MCP Server开发 / ai #65

简介

MCP起源于2024年11月25日 Anthropic 发布的文章：Introducing the Model Context Protocol。

MCP （Model Context Protocol，模型上下文协议）定义了应用程序和 AI 模型之间交换上下文信息的方式。这使得开发者能够以一致的方式将各种数据源、工具和功能连接到 AI 模型（一个中间协议层），就像 USB-C 让不同设备能够通过相同的接口连接一样。MCP 的目标是创建一个通用标准，使 AI 应用程序的开发和集成变得更加简单和统一。

MCP 就是以更标准的方式让 LLM Chat 使用不同工具。这样你应该更容易理解“中间协议层”的概念了。Anthropic 旨在实现 LLM Tool Call 的标准。

下载与资源

文档
Servers
awesome-mcp-servers
mcpservers
LangChain MCP

uv安装环境

教程
uv 是 Astral 公司推出的一款基于 Rust 编写的 Python 包管理工具，旨在成为 “Python 的 Cargo”。它提供了快速、可靠且易用的包管理体验，在性能、兼容性和功能上都有出色表现，为 Python 项目的开发和管理带来了新的选择。

1. 安装
# 通过PIP安装（推荐）
直接使用Python自带的pip安装，兼容性最佳： - 0.7.16
pip install uv   
uv --version  //uv 0.12.1

2. 初始化项目
uv init my_project
cd my_project
# 这会创建以下基本项目结构：
my_project/
└──src
|   └──my_project
|        └──__init__.py      
├── pyproject.toml    # 项目配置和依赖声明
├── .python-version   # 固定 Python 版本
└── README.md

3. 创建虚拟环境
uv venv
# 激活虚拟环境
source .venv/Scripts/activate

4. 添加生产依赖
uv add mcp
uv add openai python-dotenv
uv add httpx
# 设置国内镜像，如果安装速度慢，可以在 pyproject.toml 中设置国内镜像源
[tool.uv]
index-url = "https://pypi.tuna.tsinghua.edu.cn/simple"
# 查看已安装的包：
uv pip list

5. 运行
uv run 是 uv 中非常实用的命令，可以无需手动激活虚拟环境直接运行脚本或命令，uv 会自动找到并使用正确的环境
# 直接运行 Python 脚本
uv run main.py
# 运行任意命令（在虚拟环境的上下文中执行）
uv run python -c "import requests; print(requests.__version__)"

Server端设计

根据MCP协议定义，Server可以提供三种类型的标准能力，Resources、Tools、Prompts，每个Server可同时提供者三种类型能力或其中一种。
Resources：资源，类似于文件数据读取，可以是文件资源或是API响应返回的内容。
Tools：工具，第三方服务、功能函数，通过此可控制LLM可调用哪些函数。
Prompts：提示词，为用户预先定义好的完成特定任务的模板。

Model Context Protocol（MCP）由Anthropic开源，用于将大型语言模型直接连接数据源。它分为本地 stdio / 远程 SSE/HTTP，支持标准输入输出（stdio） 和 基于HTTP的服务器推送事件（SSE） 两种传输方式。Stdio模式适用于本地通信，通过启动服务器作为子进程实现高效低延迟的数据交换，适合快速响应的本地应用。而基于HTTP和SSE的方式则适用于分布式或远程场景，实现客户端与服务器间实时数据推送。

1. 本地通讯：文件读写、Git、本地数据库、计算器。使用了stdio传输数据，具体流程Client启动Server程序作为子进程，其消息通讯是通过stdin/stdout进行的，消息格式为JSON-RPC 2.0。
2. 远程通讯：联网搜索、企业 API、CRM、向量库、第三方服务。 Server 对外只暴露三类标准化基元（Primitives），模型可自动发现、动态调用。Client与Server可以部署在任何地方，Client使用SSE与Server进行通讯，消息的格式为JSON-RPC 2.0，Server定义了/see与/messages接口用于推送与接收数据。

Server端设计

把工具（tool）封装进特定的方法中，以提供统一的调用方法和方式，这样可以增加规范性和可复用性。

# server.py
from mcp.server.mcpserver import MCPServer

# 初始化 MCP 服务器
mcp = MCPServer("WeatherServer")

# 给函数加上装饰器即可
@mcp.tool()
async def userful_tool(input: str):
    """
    工具的说明和使用
    """
    pass
            
if __name__ == "__main__":
    mcp.run(transport='stdio')

# 可使用inspector做测试
npx -y @modelcontextprotocol/inspector uv run server.py

Server案例

以查询天气的工具来做简单的案例

# server.py
import json
import httpx
import asyncio
from typing import Any
from mcp.server.mcpserver import MCPServer

# 初始化 MCP 服务器
mcp = MCPServer("WeatherServer")
# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "1b63xxxxx...." # 请替换为你自己的 OpenWeather API Key
USER_AGENT = "weather-app/1.0"

async def fetch_weather(city: str) -> dict[str, Any] | None:
    """
    从 OpenWeather API 获取天气信息。
    :param city: 城市名称（需使用英文，如 Beijing）
    :return: 天气数据字典；若出错返回包含 error 信息的字典
    """
    params = {
        "q": city,
        "appid": API_KEY,
        "units": "metric",
        "lang": "zh_cn"
     }
    headers = {"User-Agent": USER_AGENT}
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(OPENWEATHER_API_BASE, params=params,
            headers=headers, timeout=30)
            response.raise_for_status()
            return response.json() # 返回字典类型
        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP 错误: {e.response.status_code}"}
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}
            
def format_weather(data: dict[str, Any] | str) -> str:
    """
    将天气数据格式化为易读文本。
    :param data: 天气数据（可以是字典或 JSON 字符串）
    :return: 格式化后的天气信息字符串
    """
    # 如果传入的是字符串，则先转换为字典
    if isinstance(data, str):
        try:
            data = json.loads(data)
        except Exception as e:
            return f"无法解析天气数据: {e}"
            # 如果数据中包含错误信息，直接返回错误提示
    if "error" in data:
        return f"{data['error']}"
    # 提取数据时做容错处理
    city = data.get("name", "未知")
    country = data.get("sys", {}).get("country", "未知")
    temp = data.get("main", {}).get("temp", "N/A")
    humidity = data.get("main", {}).get("humidity", "N/A")
    wind_speed = data.get("wind", {}).get("speed", "N/A")
    # weather 可能为空列表，因此用 [0] 前先提供默认字典
    weather_list = data.get("weather", [{}])
    description = weather_list[0].get("description", "未知")
    return (
        f"城市{city}, {country}\n"
        f"温度: {temp}°C\n"
        f"湿度: {humidity}%\n"
        f"风速: {wind_speed} m/s\n"
        f"天气: {description}\n")

@mcp.tool()
async def query_weather(city: str) -> str:
    """
    输入指定城市的英文名称，返回今日天气查询结果。
    :param city: 城市名称（需使用英文）
    :return: 格式化后的天气信息
    """
    data = await fetch_weather(city)
    print(123, "weather data:", data)
    return format_weather(data)

if __name__ == "__main__":
    mcp.run(transport='stdio')

# Inspector测试
npx -y @modelcontextprotocol/inspector uv run server.py

测试界面

服务器测试 MCP Inspector

在实际开发MCP服务器的过程中，Anthropic提供了一个非常便捷的debug工具：Inspector。借助Inspector，我们能够非常快捷的调用各类server，并测试其功能
npx -y @modelcontextprotocol/inspector uv run server.py