当前位置：首页 > news >正文

知识点2：MCP：python-sdk 核心概念

news 2025/7/15 14:37:10

0 前言

官方网址：https://github.com/modelcontextprotocol/python-sdk
所有内容均以官方为主，可结合食用～
系统：ubuntu 20.04

MCP让我们可以通过一种标准且安全的方式创建一个服务端用来将数据和函数功能块暴露给LLM应用程序，可以把它想象成一个专门为LLM交互设计的web API。MCP服务器可以：

通过Resources暴露数据(可以把它想象成GET；它们用于将信息加载到LLM的上下文中。)
通过Tools提供功能（类似与POST；用来执行代码或触发外部系统的状态变化。）
通过Prompts定义交互模型(它直接作用于 LLM 的提示词层面。)

1 Server

FastMCP 是 MCP 协议的核心枢纽，如同 Web 开发中的 API 网关。它充当 LLM 应用与后端服务（数据、工具、业务逻辑）之间的安全中介层，确保所有交互符合 MCP 标准。

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass# 创建一个简单的内存数据库
class SimpleDatabase:def __init__(self):# 初始化数据库，存储一些示例数据self.data = [{"id": 1, "name": "Alice"},{"id": 2, "name": "Bob"},]self.connected = False # 标记数据库连接状态@classmethodasync def connect(cls):# 异步连接数据库，返回连接后的实例db = cls()db.connected = Truereturn dbasync def disconnect(self):# 异步断开数据库连接self.connected = Falsedef query(self):# 返回全部数据return self.datafrom mcp.server.fastmcp import FastMCP
# 创建了一个名为"My App"的FastMCP服务器实例。在创建服务器时指定了依赖项（如pandas和numpy），方便部署和开发环境自动安装依赖。
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])@dataclass
class AppContext:db: SimpleDatabase@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:# 该异步上下文管理器用于管理应用的生命周期（启动和关闭）。# 启动时连接数据库，关闭时断开连接，确保资源正确释放。db = await SimpleDatabase.connect() # 启动时异步连接数据库try:yield AppContext(db=db) # 将数据库连接封装到AppContext并yield出去finally:await db.disconnect() # 关闭时断开数据库连接# 将生命周期管理器传递给服务器，服务器会在启动和关闭时自动调用app_lifespan。
mcp = FastMCP("My App", lifespan=app_lifespan)@mcp.tool()
def query_db() -> str:# 工具函数：演示如何访问生命周期上下文中的数据库连接ctx = mcp.get_context() # 获取当前请求的上下文对象db = ctx.request_context.lifespan_context.db # 从生命周期上下文中获取数据库连接# 查询数据并返回字符串return str(db.query())

运行uv run mcp dev server.py
在这里插入图片描述

2 Resources

Resources是资源（Resources）是用于向大语言模型（LLM）暴露数据的一种方式。它们的作用类似于 REST API 中的 GET 端点：主要用于提供数据，而不是执行复杂的计算或产生副作用。

from mcp.server.fastmcp import FastMCP# 创建一个 FastMCP 服务器实例，名称为 "Resource Example"
mcp = FastMCP(name="Resource Example")# 定义一个资源，路径为 "file://documents/{name}"
# 资源类似于 REST API 的 GET 端点，用于向 LLM 提供数据
@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:"""根据文档名称读取文档内容"""# 实际应用中，这里通常会从磁盘读取文件内容# 此处仅做演示，返回一个字符串，包含文档名称return f"Content of {name}"# 定义另一个资源，路径为 "config://settings"
# 用于向 LLM 提供应用配置数据
@mcp.resource("config://settings")
def get_settings() -> str:"""获取应用设置"""# 返回一个 JSON 格式的字符串，包含主题、语言和调试状态等配置信息return """{"theme": "dark","language": "en","debug": false
}"""# 总结：
# 1. @mcp.resource 装饰器用于注册资源，资源只负责数据的读取和暴露，不做复杂计算或产生副作用。
# 2. 资源的路径可以包含参数（如 {name}），方便按需获取不同数据。
# 3. 资源函数返回的数据会被 LLM 访问和使

在这里插入图片描述

3 Tools

工具（Tools）允许大语言模型（LLM）通过你的服务器执行操作。与资源不同，工具通常会进行计算，并且可能会产生副作用。

import requests
from mcp.server.fastmcp import FastMCPmcp = FastMCP(name="Tool Example")@mcp.tool()
def get_weather(city: str, unit: str = "metric") -> str:"""实时获取指定城市的天气信息（使用 OpenWeatherMap API）。unit 可选 "metric"（摄氏度）或 "imperial"（华氏度）。"""# 你需要在 openweathermap.org 注册并获取 API keyAPI_KEY = "你的API密钥"  # 请替换为你自己的 API keyurl = (f"https://api.openweathermap.org/data/2.5/weather"f"?q={city}&units={unit}&appid={API_KEY}&lang=zh_cn")try:response = requests.get(url, timeout=5)response.raise_for_status()data = response.json()temp = data["main"]["temp"]weather = data["weather"][0]["description"]unit_symbol = "°C" if unit == "metric" else "°F"return f"{city} 当前天气：{weather}，温度：{temp}{unit_symbol}"except Exception as e:return f"获取天气失败: {e}"# 示例调用：get_weather("Beijing", "metric")

根据上述代码可以获取如下结果：
在这里插入图片描述

4 结构化输出和非结构化输出

工具（Tools）默认会返回结构化结果，只要你的返回类型注解是兼容的。如果不兼容，则返回非结构化结果。

结构化输出支持以下类型：

Pydantic 模型（BaseModel 子类）
TypedDict
数据类（dataclass）和其他带类型注解的类
dict[str, T]（T 为任何可 JSON 序列化类型）
基本类型（str、int、float、bool、bytes、None）——会被包装为 {“result”: value}
泛型类型（list、tuple、Union、Optional 等）——也会被包装为 {“result”: value}

注意：

没有类型注解的类无法序列化为结构化输出。只有属性有类型注解的类才会被转换为 Pydantic 模型，用于生成 schema 和做数据校验。
结构化结果会自动根据注解生成的输出 schema 进行校验，确保工具返回的数据类型正确且易于客户端处理。

兼容性说明：

为了兼容旧版 MCP 规范，也会返回非结构化结果，保持与旧版 FastMCP 的兼容性。
如果你不希望工具被识别为结构化输出，可以在 @tool 装饰器里传递 structured_output=False 来关闭结构化输出。

总结：
结构化输出让工具返回的数据更规范、更易于校验和处理。如果你的工具返回类型注解合理，MCP 会自动帮你做数据校验和格式化。

观察之前返回的结果发现它的返回实际上是{"result": value}

我们将第三部分的代码改成下面的形式，就可以输出一个dict型的结构化结果。

import requests
from mcp.server.fastmcp import FastMCPmcp = FastMCP(name="Tool Example")@mcp.tool()
def get_weather(city: str, unit: str = "metric") -> str:"""实时获取指定城市的天气信息（使用 OpenWeatherMap API）。unit 可选 "metric"（摄氏度）或 "imperial"（华氏度）。"""# 你需要在 openweathermap.org 注册并获取 API keyAPI_KEY = "你的API密钥"  # 请替换为你自己的 API keyurl = (f"https://api.openweathermap.org/data/2.5/weather"f"?q={city}&units={unit}&appid={API_KEY}&lang=zh_cn")try:response = requests.get(url, timeout=5)response.raise_for_status()data = response.json()temp = data["main"]["temp"]weather = data["weather"][0]["description"]unit_symbol = "°C" if unit == "metric" else "°F"return {'城市':city,'当前天气':weather,'温度':str(temp)+unit_symbol}except Exception as e:return {'获取天气失败':e}# 示例调用：get_weather("Beijing", "metric")
# 实例结果：1 return ["London", "Paris", "Tokyo"]  # Returns: {"result": ["London", "Paris", "Tokyo"]}
#         2 return 22.5   # Returns: {"result": 22.5}