阶段三 MCP Python SDK(1)
概述
模型上下文协议(The Model Context Protocol)允许应用程序以标准化的方式为大型语言模型(LLMs)提供上下文,将提供上下文的职责与实际的LLM交互职责分离开来。这个Python SDK实现了完整的MCP规范,使得开发者可以轻松地:
- 构建能够连接到任何MCP服务器的MCP客户端。
- 创建能够暴露资源、提示(prompts)和工具的MCP服务器。
- 使用标准的传输方式,如标准输入输出(stdio)、服务器发送事件(SSE)和可流式传输的HTTP(Streamable HTTP)。
- 处理所有的MCP协议消息和生命周期事件。
安装
将MCP添加到你的pyhton项目中
推荐使用uv https://docs.astral.sh/uv/来管理python项目。
创建一个uv管理项目
uv init mcp-server-demo
cd mcp-server-demo

然后将MCP添加到你的项目依赖中
uv add "mcp[cli]"
对于使用pip来管理依赖的项目,可以采用另一种方式。
pip install "mcp[cli]"
运行独立的MCP开发工具
用uv运行mcp
uv run mcp
快速入门
创建一个简单的MCP服务,暴露一个计算工具和一些数据。
创建astmcp_quickstart.py文件
比如我在自己的目录下mcp-server-demo\snippets\servers\fastmcp_quickstart.py
"""
FastMCP quickstart example.
cd to the `examples/snippets/clients` directory and run:
uv run server fastmcp_quickstart stdio
"""
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
# Add a prompt
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
"""Generate a greeting prompt"""
styles = {
"friendly": "Please write a warm, friendly greeting",
"formal": "Please write a formal, professional greeting",
"casual": "Please write a casual, relaxed greeting",
}
return f"{styles.get(style, styles['friendly'])} for someone named {name}."
def main():
"""Entry point for the direct execution server."""
mcp.run()
if __name__ == "__main__":
main()
uv run mcp dev fastmcp_quickstart.py (不建议这样写uv run mcp dev ./fastmcp_quickstart.py)
他会给你启动一个mcp inspector

配置完成点击connect 连接,
注意这里不要写出run --with mcp mcp run .\fastmcp_quickstart.py 会被识识别成 .fastmcp_quickstart.py
连接成功后,就可以看到请求和响应,资源、模版、工具等,调试。

MCP是什么
模型上下文协议(MCP)可以让你构建服务器,以安全、标准化的方式向大型语言模型(LLM)应用暴露数据和功能。
类比:可以将其视为一个专为LLM交互而设计的Web API。
具体能力:
- 通过资源暴露数据:可以将其视为类似于GET端点;它们用于将信息加载到LLM的上下文中。
- 通过工具提供功能:类似于POST端点;它们用于执行代码或产生其他副作用。
- 通过提示定义交互模式:为LLM交互提供可复用的模板。
- 更多功能:还有更多!
核心概念
server
FastMCP服务器是你与MCP协议的核心接口。它负责处理连接管理、协议合规性以及消息路由。
带强类型生命周期的最小可运行示例
"""Example showing lifespan support for startup/shutdown with strong typing."""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
# Mock database class for example
class Database:
"""Mock database class for example."""
@classmethod
async def connect(cls) -> "Database":
"""Connect to database."""
return cls()
async def disconnect(self) -> None:
"""Disconnect from database."""
pass
def query(self) -> str:
"""Execute a query."""
return "Query result"
@dataclass
class AppContext:
"""Application context with typed dependencies."""
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle with type-safe context."""
# Initialize on startup
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
# Cleanup on shutdown
await db.disconnect()
# Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan)
# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(ctx: Context[ServerSession, AppContext]) -> str:
"""Tool that uses initialized resources."""
db = ctx.request_context.lifespan_context.db
return db.query()
- mock 数据库
class Database:...
- 强类型 ‘资源包’
这个上下文可以将 需要在tool/resource里访问的初始化好的依赖打包成一个数据类。在后续生命周期初始化阶段,将这个实例注入到每次请求的上下文中。
@dataclass
class AppContext:
db: Database
-
生命周期函数
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
db = await Database.connect() # 启动阶段
try:
yield AppContext(db=db) # 把资源交给框架
finally:
await db.disconnect() # 关闭阶段
被 @asynccontextmanager 装饰后,app_lifespan 就是一个异步上下文管理器。
yield 之前 = startup;yield 返回值 = 后续 tool 能拿到的「资源包」;
finally 块 = shutdown,保证进程退出前一定释放连接。
-
创建服务器并绑定 lifespan
mcp = FastMCP("My App", lifespan=app_lifespan)
把刚才写好的生命周期函数注册给服务器;FastMCP 会在启动停止时自动调用。
-
在工具里类型安全地使用资源
@mcp.tool()
def query_db(ctx: Context[ServerSession, AppContext]) -> str:
db = ctx.request_context.lifespan_context.db
return db.query()
参数 ctx: Context[ServerSession, AppContext] 把「会话信息」和「资源包」都带上了。
ctx.request_context.lifespan_context 就是 yield 出去的 AppContext 实例。
因为泛型里写了 AppContext,IDE 直接提示 .db,无需魔法字符串,也不会拿错类型。
app_lifespan 是一个 “异步上下文管理器”(由 @asynccontextmanager 装饰),它的执行时机和次数完全由 FastMCP 服务器进程的生命周期决定。
一句话结论
-
只在整个进程启动时执行 1 次(
yield之前部分) -
只在整个进程即将退出时执行 1 次(
finally/yield之后部分) -
不会因为每个客户端连接或每次工具调用而重复执行
1. 当你启动服务器,FastMCP 内部会做一次
await app_lifespan(mcp).__aenter__()
-
于是进入
async def app_lifespan函数体:-
先
Database.connect() -
再
yield AppContext(...)
这一步称为 startup,整个进程生命周期里只跑一次。
-
-
服务器现在处于运行状态,所有 tool/resource 请求共享同一个
AppContext实例。 -
当你按下
Ctrl+C或进程被 systemd/k8s 关闭时,FastMCP 会等所有正在处理的请求完成后,调用
当你按下 Ctrl+C 或进程被 systemd/k8s 关闭时,FastMCP 会等所有正在处理的请求完成后,调用
await app_lifespan(mcp).__aexit__(...)
于是进入 finally 块:
-
打印日志
-
await db.disconnect()
这一步称为 shutdown,也只跑一次。
Resources
资源是您向大型语言模型(LLMs)展示数据的方式。它们类似于REST API中的GET端点——它们提供数据,但不应该进行大量的计算,也不应该产生副作用。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="Resource Example")
@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
"""Read a document by name."""
# This would normally read from disk
return f"Content of {name}"
@mcp.resource("config://settings")
def get_settings() -> str:
"""Get application settings."""
return """{
"theme": "dark",
"language": "en",
"debug": false
}"""
@mcp.resource("任意URI模板") 装饰一个普通函数,就能把动态或静态数据暴露成只读 URI 资源,客户端通过标准 MCP 协议即可按需获取,零额外路由、零手动序列化。
Tools
工具允许大型语言模型(LLMs)通过你的服务器采取行动。与资源不同,工具被期望执行计算并产生副作用。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="Tool Example")
@mcp.tool()
def sum(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> str:
"""Get weather for a city."""
# This would normally call a weather API
return f"Weather in {city}: 22degrees{unit[0].upper()}"
工具可以通过包含一个带有上下文类型注解的参数来选择性地接收一个上下文对象。这个上下文由FastMCP框架自动注入,并且提供了访问MCP功能的能力。
Structured Output
工具默认会返回结构化的结果,前提是它们的返回类型注解是兼容的。否则,它们将返回非结构化的结果。
结构化输出支持以下返回类型:
- Pydantic模型(BaseModel的子类)
- 带有类型注解的字典(TypedDicts)
- 数据类和其他带有类型注解的类
- dict[str, T](其中T是任何可JSON序列化的类型)
- 原始类型(str、int、float、bool、bytes、None)——这些类型将被包装在{"result": value}中
- 泛型类型(list、tuple、Union、Optional等)——这些类型将被包装在{"result": value}中
没有类型注解的类无法被序列化为结构化输出。只有带有正确注解属性的类才会被转换为Pydantic模型,用于模式生成和验证。
结构化结果将自动与从注解生成的输出模式进行验证。这确保了工具返回的是类型正确、经过验证的数据,客户端可以轻松处理。
注意:为了向后兼容,也返回非结构化结果。非结构化结果是为了与MCP规范的早期版本保持向后兼容而提供的,并且与当前SDK版本中的FastMCP早期版本在处理上是兼容的。
注意:如果工具函数的返回类型注解导致工具被归类为结构化,而这是不希望的,可以通过在@tool装饰器中传递structured_output=False来抑制这种分类。
Advanced: Direct CallToolResult
允许用户直接调用工具的结果,而不是通过中间步骤。
为了对工具响应(包括 _meta 字段)进行完全控制(_meta 字段用于将数据传递给客户端应用程序,而不会将其暴露给模型),你可以直接返回 CallToolResult。
"""Example showing direct CallToolResult return for advanced control."""
from typing import Annotated
from pydantic import BaseModel
from mcp.server.fastmcp import FastMCP
from mcp.types import CallToolResult, TextContent
mcp = FastMCP("CallToolResult Example")
class ValidationModel(BaseModel):
"""Model for validating structured output."""
status: str
data: dict[str, int]
@mcp.tool()
def advanced_tool() -> CallToolResult:
"""Return CallToolResult directly for full control including _meta field."""
return CallToolResult(
content=[TextContent(type="text", text="Response visible to the model")],
_meta={"hidden": "data for client applications only"},
)
@mcp.tool()
def validated_tool() -> Annotated[CallToolResult, ValidationModel]:
"""Return CallToolResult with structured output validation."""
return CallToolResult(
content=[TextContent(type="text", text="Validated response")],
structuredContent={"status": "success", "data": {"result": 42}},
_meta={"internal": "metadata"},
)
@mcp.tool()
def empty_result_tool() -> CallToolResult:
"""For empty results, return CallToolResult with empty content."""
return CallToolResult(content=[])
不再让 FastMCP 帮你把函数返回值自动包成工具响应,而是自己构造 CallToolResult(CallToolResult 是 MCP 协议里工具调用的最终响应对象),从而能控制:
-
模型可见的内容(
content) -
客户端私有的附加数据(
_meta) -
结构化数据(
structuredContent) -
空结果(
content=[])
平时写
@mcp.tool()
def foo() -> str:
return "hello"
FastMCP 会帮你生成:CallToolResult(content=[TextContent(text="hello")])
但无法控制 _meta、无法返回空内容、无法塞私有字段——于是有了手动模式。
重要提示:必须始终返回CallToolResult(不能使用Optional或Union)。对于空结果,使用CallToolResult(content=[])。对于可选的简单类型,使用str | None,而不是CallToolResult。
Prompts
提示词是可重复使用的模板,能够帮助大型语言模型(LLMs)与你的服务器进行有效互动
basic_prompt.py
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base
mcp = FastMCP(name="Prompt Example")
@mcp.prompt(title="Code Review")
def review_code(code: str) -> str:
return f"Please review this code:\n\n{code}"
@mcp.prompt(title="Debug Assistant")
def debug_error(error: str) -> list[base.Message]:
return [
base.UserMessage("I'm seeing this error:"),
base.UserMessage(error),
base.AssistantMessage("I'll help debug that. What have you tried so far?"),
]
Icons
MCP服务器可以为用户界面显示提供图标。图标可以添加到服务器实现、工具、资源和提示中。
先动过main启动,调试是否成功运行。下述代码需要当前目录下加个mcp.png的图片。
"""
FastMCP Icons Demo Server
Demonstrates using icons with tools, resources, prompts, and implementation.
"""
import base64
from pathlib import Path
from mcp.server.fastmcp import FastMCP, Icon
# Load the icon file and convert to data URI
icon_path = Path(__file__).parent / "mcp.png"
icon_data = base64.standard_b64encode(icon_path.read_bytes()).decode()
icon_data_uri = f"data:image/png;base64,{icon_data}"
icon_data = Icon(src=icon_data_uri, mimeType="image/png", sizes=["64x64"])
# Create server with icons in implementation
mcp = FastMCP("Icons Demo Server", website_url="https://github.com/modelcontextprotocol/python-sdk", icons=[icon_data])
@mcp.tool(icons=[icon_data])
def demo_tool(message: str) -> str:
"""A demo tool with an icon."""
return message
@mcp.resource("demo://readme", icons=[icon_data])
def readme_resource() -> str:
"""A demo resource with an icon"""
return "This resource has an icon"
@mcp.prompt("prompt_with_icon", icons=[icon_data])
def prompt_with_icon(text: str) -> str:
"""A demo prompt with an icon"""
return text
@mcp.tool(
icons=[
Icon(src=icon_data_uri, mimeType="image/png", sizes=["16x16"]),
Icon(src=icon_data_uri, mimeType="image/png", sizes=["32x32"]),
Icon(src=icon_data_uri, mimeType="image/png", sizes=["64x64"]),
]
)
def multi_icon_tool(action: str) -> str:
"""A tool demonstrating multiple icons."""
return "multi_icon_tool"
if __name__ == "__main__":
# Run the server
mcp.run()

Images
FastMCP 提供了一个名为 Image 的类,该类能够自动处理图像数据。
"""Example showing image handling with FastMCP."""
from PIL import Image as PILImage
from mcp.server.fastmcp import FastMCP, Image
mcp = FastMCP("Image Example")
@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
"""Create a thumbnail from an image"""
img = PILImage.open(image_path)
img.thumbnail((100, 100))
return Image(data=img.tobytes(), format="png")
https://github.com/modelcontextprotocol/python-sdk
更多推荐


所有评论(0)