​

项目源码地址：

GitHub - brianxiadong/llm-langchain-project: llm-langchain-project

路过的朋友请帮忙点个star
​

1、模型上下文协议（Model Context Protocol）入门

2024 年 11 月，Anthropic 公司搞了个挺有意思的新玩意 - Model Context Protocol（模型上下文协议）简称为 MCP 协议。简单来说，它就是给 AI 和各类工具数据之间搭了个标准化的"桥梁"，让开发者不用再为对接问题头疼了。大模型应用可以使用别人分享的 MCP 服务来完成各种各样的工作内容，你可以从这些地方获取 MCP 服务：

• awesome-mcp-servers
• mcp.so

MCP 协议在实际的应用场景上非常广泛，列举一些比较常见的应用场景：

• 使用百度/高德地图分析旅线计算时间
• 接 Puppeteer 自动操作网页
• 使用 Github/Gitlab 让大模型接管代码仓库
• 使用数据库组件完成对 Mysql、ES、Redis 等数据库的操作
• 使用搜索组件扩展大模型的数据搜索能力

1.1 在 Claude Desktop 中体验 MCP

接下来我们使用 Claude 快速接入 Github 服务（提前申请 token），编辑一下 Claude Desktop 的配置文件：

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json

添加如下内容，注意把<YOUR_TOKEN>替换成你自己申请的 token：

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "`"
      }
    }
  }
}

重启Claude之后，可以看到已经加载了MCP对应的工具：

点开之后可以看到具体的工具内容：

此时我们就可以享受 Github 服务提供的操作仓库的能力：

从图上可以看到，通过创建仓库test-mcp这样的提示词，Claude 的大模型自行判断需要使用 mcp 中提供的create_repository能力，从而完成了仓库的创建，接下来我们打开 Github 也确实发现了这个已经创建的仓库。

通过这种方式，大模型就可以利用MCP接入各式各样的能力，完成各种更为复杂的工作。

1.2 MCP 的架构

MCP 主要分为MCP服务和MCP客户端：

• 客户端：一般指的是大模型应用，比如 Claude、通过 Spring AI Alibaba、Langchain 等框架开发的 AI 应用
• 服务端：连接各种数据源的服务和工具

整体架构如下：

整体的工作流程是这样的：AI 应用中集成MCP客户端，通过MCP协议向MCP服务端发起请求，MCP 服务端可以连接本地/远程的数据源，或者通过 API 访问其他服务，从而完成数据的获取，返回给 AI 应用去使用。

2、开发MCP应用

如果希望大模型接入我们自己的系统，开发一个MCP应用时一个非常好的选择。Mcp的开发流程可以参考https://github.com/modelcontextprotocol。

首先，我们安装 uv 并设置我们的 Python 项目和环境。uv（全称 Universal Virtual ）是 Astral 团队基于 Rust 语言开发的工具，旨在替代传统的 pip 和 venv 等。核心优势在于：

• 性能卓越：由 Rust 编写，依赖解析速度比 pip 快 10 - 100 倍 ，安装大型库如 numpy、pandas 仅需几秒。
• 功能集成：将虚拟环境管理和包安装融合，首次在项目目录使用时自动创建标准虚拟环境；支持 requirements.txt 和 pyproject.toml ，可直接使用现有项目结构；采用锁定文件（uv.lock）精确记录依赖版本，保障跨环境依赖一致性 ；还具备创建新项目、在虚拟环境执行脚本、项目打包发布等功能 ，覆盖 Python 从安装、开发到构建发行全流程。
• 跨平台支持：Windows、macOS、Linux 系统均适用。

接下来安装uv:

curl -LsSf https://astral.sh/uv/install.sh | sh

安装完之后重启你的终端，接下来进行环境的初始化:

uv init weather
cd weather

# 创建虚拟环境
uv venv
source .venv/bin/activate

# 安装依赖
uv add "mcp[cli]" httpx

# 创建mcp服务文件
touch weather.py

​完整代码如下:

from typing import Any, Dict, Optional, Union
import httpx
from fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"


async def make_nws_request(url: str) -> Optional[Dict[str, Any]]:
    """Make a request to the NWS API with proper error handling."""
    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 None

def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    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')}
"""


@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. 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 "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    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:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Get the forecast URL from the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)


if __name__ == "__main__":
    import asyncio
    # Initialize and run the server
    mcp.run()

详细解读一下这段代码：
本文档详细解析 weather.py 的代码结构、设计模式和实现细节。

📁 文件结构分析

weather.py (98行)
├── 导入模块 (1-3行)
├── 服务器初始化 (5-10行)
├── 工具函数 (12-25行)
├── 数据格式化 (27-35行)
├── MCP工具定义 (37-96行)
└── 服务器启动 (97-98行)

🔍 逐行代码解析

1. 模块导入部分 (1-3行)

from typing import Any, Dict, Optional, Union
import httpx
from fastmcp import FastMCP

分析：

• typing：提供类型注解支持，增强代码可读性和IDE支持
• httpx：现代异步HTTP客户端，比requests更适合异步编程
• fastmcp：MCP服务器框架，简化MCP协议实现

设计考虑：

• 使用类型注解提高代码质量
• 选择异步HTTP库支持高并发
• 采用高级MCP框架减少样板代码

2. 服务器初始化 (5-10行)

# Initialize FastMCP server
mcp = FastMCP("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

分析：

• 服务器实例：创建名为"weather"的MCP服务器
• 常量定义：将API基础URL和User-Agent提取为常量
• 配置集中化：便于维护和修改

设计模式：

• 单例模式：全局唯一的MCP服务器实例
• 常量模式：避免硬编码，提高可维护性

3. HTTP请求工具函数 (12-25行)

async def make_nws_request(url: str) -> Optional[Dict[str, Any]]:
    """Make a request to the NWS API with proper error handling."""
    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 None

代码分析：

函数签名

• 异步函数：async def 支持非阻塞操作
• 类型注解：Optional[Dict[str, Any]] 明确返回类型
• 参数类型：url: str 确保输入类型正确

请求头设置

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()  # 检查HTTP状态码
    return response.json()
except Exception:
    return None  # 统一错误处理，返回None

设计优点：

• 防御性编程：捕获所有异常，避免服务器崩溃
• 超时控制：30秒超时防止请求挂起
• 状态码检查：确保HTTP请求成功

4. 数据格式化函数 (27-35行)

def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    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')}
"""

代码分析：

数据提取

props = feature["properties"]

• 数据结构理解：NWS API返回GeoJSON格式，警报信息在properties字段

安全访问模式

props.get('event', 'Unknown')

• 防御性编程：使用.get()方法避免KeyError
• 默认值策略：提供有意义的默认值

字符串格式化

• f-string模板：使用三引号多行字符串
• 结构化输出：固定格式便于解析和显示

设计模式：

• 数据转换器模式：将原始API数据转换为用户友好格式
• 模板方法模式：固定的格式化模板

5. MCP工具定义

5.1 天气警报工具 (37-55行)

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. 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 "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)

代码分析：

装饰器模式

@mcp.tool()

• 声明式编程：通过装饰器将函数注册为MCP工具
• 框架集成：FastMCP自动处理工具注册和调用

URL构建

url = f"{NWS_API_BASE}/alerts/active/area/{state}"

• 字符串插值：动态构建API端点
• RESTful设计：遵循REST API约定

数据验证

if not data or "features" not in data:
    return "Unable to fetch alerts or no alerts found."

if not data["features"]:
    return "No active alerts for this state."

• 多层验证：检查数据存在性和结构完整性
• 用户友好错误：返回可理解的错误信息

数据处理

alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)

• 列表推导式：简洁的数据转换
• 分隔符连接：使用---分隔多个警报

5.2 天气预报工具 (57-96行)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Get the forecast URL from the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

代码分析：

两阶段API调用

# 第一阶段：获取预报网格信息
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)

# 第二阶段：获取实际预报数据
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)

设计分析：

• API设计理解：NWS API采用HATEOAS设计，需要两步获取数据
• 异步链式调用：两个异步请求的顺序执行
• 错误传播：每个阶段都有独立的错误处理

数据切片和格式化

for period in periods[:5]:  # Only show next 5 periods

• 数据限制：只显示前5个时段，避免信息过载
• 性能考虑：减少数据处理量

结构化输出

forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""

• 模板化输出：固定格式便于解析
• 多字段组合：整合温度、风力、详细预报

6. 服务器启动 (97-98行)

if __name__ == "__main__":
    import asyncio
    # Initialize and run the server
    mcp.run()

分析：

• 模块保护：只在直接运行时启动服务器
• 异步导入：按需导入asyncio模块
• 框架启动：调用FastMCP的run方法

3、测试

可以在Claude Desktop中测试，但是我的账号被封了，所以只能在Cursor中进行测试了。

打开配置，点击右上角的添加MCP配置：

填入如下内容:

{
    "mcpServers": {
        "weather-server": {
            "command": "uv",
            "args": [
                "--directory",
                "weather目录的全路径",
                "run",
                "weather.py"
            ],
            "env": {}
        }
    }
}

然后可以看到提供了两个工具，我们尝试调用他们。

这里一定要选择agent，否则是不能调用mcp的。

成功调用了MCP程序。