Loading...
Loading...
Compare original and translation side by side
mcpmcpuv add "mcp[cli]"uv add "mcp[cli]"
Requires Python ≥ 3.10. The `[cli]` extra adds the `mcp` CLI for development tooling.
要求 Python ≥ 3.10。`[cli]` 扩展会添加用于开发工具的 `mcp` 命令行工具。| Primitive | Analogy | Purpose |
|---|---|---|
| Resources | GET endpoint | Load data into LLM context (read-only) |
| Tools | POST endpoint | Execute actions, produce side effects |
| Prompts | Template | Reusable interaction patterns for LLMs |
| 原语 | 类比 | 用途 |
|---|---|---|
| Resources | GET 接口 | 将数据加载到LLM上下文(只读) |
| Tools | POST 接口 | 执行操作,产生副作用 |
| Prompts | 模板 | 供LLM使用的可复用交互模式 |
FastMCPFastMCPfrom mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"MyServer",
stateless_http=True, # recommended for production HTTP
json_response=True, # recommended for scalability
)from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"MyServer",
stateless_http=True, # 生产环境HTTP推荐配置
json_response=True, # 可扩展性推荐配置
)from pydantic import BaseModel, Field
class WeatherData(BaseModel):
temperature: float = Field(description="Temperature in Celsius")
condition: str
humidity: float
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> WeatherData:
"""Get current weather for a city.
Returns structured weather data validated against WeatherData schema.
"""
# Implementation calls a real weather API
return WeatherData(temperature=22.5, condition="sunny", humidity=45.0)BaseModelTypedDictdataclassasync defctx: Contextfrom pydantic import BaseModel, Field
class WeatherData(BaseModel):
temperature: float = Field(description="Temperature in Celsius")
condition: str
humidity: float
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> WeatherData:
"""Get current weather for a city.
Returns structured weather data validated against WeatherData schema.
"""
# 实现部分调用真实天气API
return WeatherData(temperature=22.5, condition="sunny", humidity=45.0)BaseModelTypedDictdataclassasync defctx: Context@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
"""Read a document by name from the document store."""
# Read from disk, DB, or cache
return f"Content of {name}"
@mcp.resource("config://settings")
def get_settings() -> str:
"""Return current application settings as JSON."""
return '{"theme": "dark", "debug": false}'{param}@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
"""从文档存储中按名称读取文档。"""
# 从磁盘、数据库或缓存读取
return f"Content of {name}"
@mcp.resource("config://settings")
def get_settings() -> str:
"""以JSON格式返回当前应用配置。"""
return '{"theme": "dark", "debug": false}'{param}from mcp.server.fastmcp.prompts import base
@mcp.prompt(title="Code Review")
def review_code(code: str, language: str = "python") -> list[base.Message]:
"""Generate a structured code review prompt."""
return [
base.UserMessage(f"Please review this {language} code:"),
base.UserMessage(f"```{language}\n{code}\n```"),
base.AssistantMessage("I'll analyze the code for correctness, style, and potential issues."),
]from mcp.server.fastmcp.prompts import base
@mcp.prompt(title="Code Review")
def review_code(code: str, language: str = "python") -> list[base.Message]:
"""生成结构化的代码评审提示词。"""
return [
base.UserMessage(f"Please review this {language} code:"),
base.UserMessage(f"```{language}\n{code}\n```"),
base.AssistantMessage("I'll analyze the code for correctness, style, and potential issues."),
]ctx: Contextfrom mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
@mcp.tool()
async def long_running_task(
task_name: str,
steps: int,
ctx: Context[ServerSession, None],
) -> str:
"""Run a multi-step task with progress reporting."""
await ctx.info(f"Starting task: {task_name}")
for i in range(steps):
await ctx.report_progress(
progress=(i + 1) / steps,
total=1.0,
message=f"Step {i + 1} of {steps}",
)
await ctx.info("Task complete")
return f"Completed {task_name}"| Method | Purpose |
|---|---|
| Send info log to client |
| Send debug log |
| Send warning log |
| Send error log |
| Report numeric progress |
| Read another resource from within a tool |
| Request structured input from the user |
| Unique ID for current request |
| Access server instance metadata |
ctx: Contextfrom mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
@mcp.tool()
async def long_running_task(
task_name: str,
steps: int,
ctx: Context[ServerSession, None],
) -> str:
"""Run a multi-step task with progress reporting."""
await ctx.info(f"Starting task: {task_name}")
for i in range(steps):
await ctx.report_progress(
progress=(i + 1) / steps,
total=1.0,
message=f"Step {i + 1} of {steps}",
)
await ctx.info("Task complete")
return f"Completed {task_name}"| 方法 | 用途 |
|---|---|
| 向客户端发送信息日志 |
| 发送调试日志 |
| 发送警告日志 |
| 发送错误日志 |
| 上报数值化进度 |
| 在工具内部读取其他资源 |
| 向用户请求结构化输入 |
| 当前请求的唯一ID |
| 访问服务器实例元数据 |
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
import httpx
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppState:
http_client: httpx.AsyncClient
@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppState]:
async with httpx.AsyncClient() as client:
yield AppState(http_client=client)
mcp = FastMCP("MyServer", lifespan=lifespan)
@mcp.tool()
async def fetch_url(url: str, ctx: Context) -> str:
"""Fetch content from a URL using the shared HTTP client."""
state: AppState = ctx.request_context.lifespan_context
response = await state.http_client.get(url)
return response.text@dataclassTypedDictfrom collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
import httpx
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppState:
http_client: httpx.AsyncClient
@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppState]:
async with httpx.AsyncClient() as client:
yield AppState(http_client=client)
mcp = FastMCP("MyServer", lifespan=lifespan)
@mcp.tool()
async def fetch_url(url: str, ctx: Context) -> str:
"""使用共享HTTP客户端获取URL内容。"""
state: AppState = ctx.request_context.lifespan_context
response = await state.http_client.get(url)
return response.text@dataclassTypedDict@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide a by b."""
if b == 0:
raise ValueError("Division by zero is not allowed")
return a / bCallToolResultfrom mcp.types import CallToolResult, TextContent
@mcp.tool()
def safe_parse(data: str) -> CallToolResult:
"""Parse data, returning errors inline rather than raising."""
try:
result = parse(data)
return CallToolResult(
content=[TextContent(type="text", text=str(result))]
)
except ParseError as exc:
return CallToolResult(
content=[TextContent(type="text", text=f"Parse failed: {exc}")],
isError=True,
)isError=True@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide a by b."""
if b == 0:
raise ValueError("Division by zero is not allowed")
return a / bCallToolResultfrom mcp.types import CallToolResult, TextContent
@mcp.tool()
def safe_parse(data: str) -> CallToolResult:
"""Parse data, returning errors inline rather than raising."""
try:
result = parse(data)
return CallToolResult(
content=[TextContent(type="text", text=str(result))]
)
except ParseError as exc:
return CallToolResult(
content=[TextContent(type="text", text=f"Parse failed: {exc}")],
isError=True,
)isError=Trueundefinedundefined
| Pattern | Recommendation |
|---------|---------------|
| Transport (production) | Streamable HTTP with `stateless_http=True, json_response=True` |
| Transport (local/stdio) | stdio via `mcp.run()` or `uv run mcp run server.py` |
| I/O tools | Use `async def` |
| Shared state | Use lifespan context |
| Structured output | Return Pydantic `BaseModel` subclass |
| Progress reporting | Use `ctx.report_progress()` |
| Secrets/config | Pass via environment variables, not hardcoded |
| 场景 | 推荐方案 |
|---------|---------------|
| 生产环境传输协议 | 启用`stateless_http=True, json_response=True`的流式HTTP |
| 本地/标准输入输出传输 | 标准输入输出,通过`mcp.run()`或`uv run mcp run server.py`启动 |
| I/O密集型工具 | 使用`async def` |
| 共享状态 | 使用生命周期上下文 |
| 结构化输出 | 返回Pydantic `BaseModel`子类 |
| 进度上报 | 使用`ctx.report_progress()` |
| 密钥/配置 | 通过环境变量传递,不要硬编码 |references/server-patterns.mdreferences/transports-and-deployment.mdreferences/server-patterns.mdreferences/transports-and-deployment.md