ITPub博客

首页 > 应用开发 > Java > MCP快速入门—快速构建自己的服务器

MCP快速入门—快速构建自己的服务器

原创 Java 作者:lxz2266 时间:2025-07-16 07:57:37 0 删除 编辑

正文内容

1. MCP核心概念与技术背景

MCP服务器主要提供三种能力类型:

  • 资源(Resources):客户端可读取的类似文件的数据(如API响应或文件内容)
  • 工具(Tools):经用户批准后LLM可调用的函数
  • 提示(Prompts):帮助用户完成特定任务的预编写模板

本教程将重点介绍工具类型的实现,通过构建两个实用工具(get-alerts和get-forecast)来扩展Claude的功能,使其能够获取天气预报和恶劣天气警报。

2. 环境准备与项目初始化

2.1 系统要求

  • Python 3.10或更高版本
  • Python MCP SDK 1.2.0或更高版本

2.2 安装uv并创建项目

# 安装uv包管理器curl -LsSf https://astral.sh/uv/install.sh | sh# 创建项目目录uv init weathercd weather# 创建并激活虚拟环境uv venvsource .venv/bin/activate# 安装依赖uv add "mcp[cli]" httpx# 创建服务器文件touch weather.py

3. 构建天气服务器

3.1 初始化FastMCP实例

from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCP# 初始化FastMCP服务器mcp = FastMCP("weather")# 常量定义NWS_API_BASE = "https://api.weather.gov"USER_AGENT = "weather-app/1.0"```
FastMCP类利用Python类型提示和文档字符串自动生成工具定义,简化了MCP工具的创建和维护过程。

3.2 实现辅助函数

async def make_nws_request(url: str) -> dict[str, Any] | None:    """向NWS API发起请求并处理错误"""
    headers = {        "User-Agent": USER_AGENT,        "Accept": "application/geo+json"
    }    async with httpx.AsyncClient() as client:        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()            return response.json()        except Exception:            return Nonedef format_alert(feature: dict) -> str:    """格式化警报特征为可读字符串"""
    props = feature["properties"]    return f"""
Event: {props.get('event', 'Unknown')}Area: {props.get('areaDesc', 'Unknown')}Severity: {props.get('severity', 'Unknown')}Description: {props.get('description', 'No description available')}Instructions: {props.get('instruction', 'No specific instructions provided')}"""

3.3 实现工具功能

@mcp.tool()async def get_alerts(state: str) -> str:    """获取美国各州的天气警报
    
    Args:
        state: 两字母州代码(如CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)    if not data or "features" not in data:        return "无法获取警报或未发现警报"
    if not data["features"]:        return "该州无活跃警报"
    alerts = [format_alert(feature) for feature in data["features"]]    return "\n---\n".join(alerts)@mcp.tool()async def get_forecast(latitude: float, longitude: float) -> str:    """获取某地天气预报
    
    Args:
        latitude: 纬度
        longitude: 经度
    """
    # 首先获取预测网格端点
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)    if not points_data:        return "无法获取该位置的预测数据"
    # 从points响应中获取预测URL
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)    if not forecast_data:        return "无法获取详细预测"
    # 将时间段格式化为可读预测
    periods = forecast_data["properties"]["periods"]
    forecasts = []    for period in periods[:5]:  # 仅显示接下来5个时段
        forecast = f"""{period['name']}:
温度: {period['temperature']}°{period['temperatureUnit']}风速: {period['windSpeed']} {period['windDirection']}预测: {period['detailedForecast']}"""
        forecasts.append(forecast)    return "\n---\n".join(forecasts)PYTHON折叠复制全屏

3.4 运行服务器

if __name__ == "__main__":    # 初始化并运行服务器
    mcp.run(transport='stdio')

4. 连接Claude for Desktop进行测试

4.1 配置客户端

{


来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70046763/viewspace-3091250/ ,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2025-06-14

  • 博文量
    14
  • 访问量
    13508