mcp-server
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseMCP Server Skill
MCP 服务器技能
Build MCP (Model Context Protocol) servers using the official Python SDK with FastMCP high-level API.
使用官方Python SDK结合FastMCP高级API构建MCP(Model Context Protocol,模型上下文协议)服务器。
Architecture
架构
┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Server (FastMCP) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ @tool() │ │ @resource() │ │ @prompt() │ │
│ │ add_task │ │ tasks:// │ │ task_prompt │ │
│ │ list_tasks │ │ user:// │ │ help_prompt │ │
│ │complete_task│ │ │ │ │ │
│ └──────┬──────┘ └──────┬──────┘ └─────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Database Layer (SQLModel) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
│
│ Transports: stdio | SSE | streamable-http
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Client (Agent) │
│ OpenAI Agents SDK / Claude / Other Clients │
└─────────────────────────────────────────────────────────────────────────┘┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Server (FastMCP) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ @tool() │ │ @resource() │ │ @prompt() │ │
│ │ add_task │ │ tasks:// │ │ task_prompt │ │
│ │ list_tasks │ │ user:// │ │ help_prompt │ │
│ │complete_task│ │ │ │ │ │
│ └──────┬──────┘ └──────┬──────┘ └─────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Database Layer (SQLModel) │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
│
│ Transports: stdio | SSE | streamable-http
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Client (Agent) │
│ OpenAI Agents SDK / Claude / Other Clients │
└─────────────────────────────────────────────────────────────────────────┘Quick Start
快速开始
Installation
安装
bash
undefinedbash
undefinedpip
pip
pip install mcp
pip install mcp
poetry
poetry
poetry add mcp
poetry add mcp
uv
uv
uv add mcp
undefineduv add mcp
undefinedEnvironment Variables
环境变量
env
DATABASE_URL=postgresql://user:password@localhost:5432/mydbenv
DATABASE_URL=postgresql://user:password@localhost:5432/mydbCore Concepts
核心概念
| Concept | Decorator | Purpose |
|---|---|---|
| Tools | | Perform actions, computations, side effects |
| Resources | | Expose data for reading (URI-based) |
| Prompts | | Reusable prompt templates |
| 概念 | 装饰器 | 用途 |
|---|---|---|
| 工具 | | 执行操作、计算和副作用 |
| 资源 | | 以URI形式暴露可读取的数据 |
| 提示词 | | 可复用的提示词模板 |
Basic MCP Server
基础MCP服务器
python
from mcp.server.fastmcp import FastMCPpython
from mcp.server.fastmcp import FastMCPCreate server instance
创建服务器实例
mcp = FastMCP("Todo Server", json_response=True)
mcp = FastMCP("Todo Server", json_response=True)
Define a tool
定义一个工具
@mcp.tool()
def add_task(title: str, description: str = None) -> dict:
"""Add a new task to the todo list."""
# Implementation here
return {"task_id": 1, "status": "created"}
@mcp.tool()
def add_task(title: str, description: str = None) -> dict:
"""向待办事项列表添加新任务。"""
# 实现代码在此处
return {"task_id": 1, "status": "created"}
Define a resource
定义一个资源
@mcp.resource("tasks://{user_id}")
def get_user_tasks(user_id: str) -> str:
"""Get all tasks for a user."""
return "task data as string"
@mcp.resource("tasks://{user_id}")
def get_user_tasks(user_id: str) -> str:
"""获取用户的所有任务。"""
return "task data as string"
Define a prompt
定义一个提示词
@mcp.prompt()
def task_assistant(task_type: str = "general") -> str:
"""Generate a task management prompt."""
return f"You are a helpful {task_type} task assistant."
@mcp.prompt()
def task_assistant(task_type: str = "general") -> str:
"""生成任务管理提示词。"""
return f"You are a helpful {task_type} task assistant."
Run the server
运行服务器
if name == "main":
mcp.run(transport="streamable-http")
undefinedif name == "main":
mcp.run(transport="streamable-http")
undefinedReference
参考文档
| Pattern | Guide |
|---|---|
| Tools | reference/tools.md |
| Resources | reference/resources.md |
| Prompts | reference/prompts.md |
| Transports | reference/transports.md |
| FastAPI Integration | reference/fastapi-integration.md |
| 模式 | 指南 |
|---|---|
| 工具 | reference/tools.md |
| 资源 | reference/resources.md |
| 提示词 | reference/prompts.md |
| 传输方式 | reference/transports.md |
| FastAPI集成 | reference/fastapi-integration.md |
Examples
示例
| Example | Description |
|---|---|
| examples/todo-server.md | Complete todo MCP server with CRUD tools |
| examples/database-integration.md | MCP server with SQLModel database |
| 示例 | 描述 |
|---|---|
| examples/todo-server.md | 包含CRUD工具的完整待办事项MCP服务器 |
| examples/database-integration.md | 集成SQLModel数据库的MCP服务器 |
Templates
模板
| Template | Purpose |
|---|---|
| templates/mcp_server.py | Basic MCP server template |
| templates/mcp_tools.py | Tool definitions template |
| templates/mcp_fastapi.py | FastAPI + MCP integration template |
| 模板 | 用途 |
|---|---|
| templates/mcp_server.py | 基础MCP服务器模板 |
| templates/mcp_tools.py | 工具定义模板 |
| templates/mcp_fastapi.py | FastAPI + MCP集成模板 |
FastAPI/Starlette Integration
FastAPI/Starlette集成
python
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCPpython
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCPCreate MCP server
创建MCP服务器
mcp = FastMCP("Todo Server", stateless_http=True, json_response=True)
@mcp.tool()
def add_task(title: str) -> dict:
"""Add a task."""
return {"status": "created"}
mcp = FastMCP("Todo Server", stateless_http=True, json_response=True)
@mcp.tool()
def add_task(title: str) -> dict:
"""添加任务。"""
return {"status": "created"}
Lifespan manager for session handling
用于会话处理的生命周期管理器
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
async with mcp.session_manager.run():
yield
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
async with mcp.session_manager.run():
yield
Create Starlette app with MCP mounted
创建挂载MCP的Starlette应用
app = Starlette(
routes=[
Mount("/mcp", app=mcp.streamable_http_app()),
],
lifespan=lifespan,
)
app = Starlette(
routes=[
Mount("/mcp", app=mcp.streamable_http_app()),
],
lifespan=lifespan,
)
Add CORS for browser clients
为浏览器客户端添加CORS支持
app = CORSMiddleware(
app,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE"],
expose_headers=["Mcp-Session-Id"],
)
app = CORSMiddleware(
app,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE"],
expose_headers=["Mcp-Session-Id"],
)
Run with: uvicorn server:app --reload
运行命令: uvicorn server:app --reload
MCP endpoint: http://localhost:8000/mcp/mcp
undefinedundefinedTool with Context (Progress, Logging)
带上下文的工具(进度、日志)
python
from mcp.server.fastmcp import Context, FastMCP
mcp = FastMCP("Progress Server")
@mcp.tool()
async def long_task(steps: int, ctx: Context) -> str:
"""Execute a task with progress updates."""
await ctx.info("Starting task...")
for i in range(steps):
progress = (i + 1) / steps
await ctx.report_progress(
progress=progress,
total=1.0,
message=f"Step {i + 1}/{steps}",
)
await ctx.debug(f"Completed step {i + 1}")
return f"Task completed in {steps} steps"python
from mcp.server.fastmcp import Context, FastMCP
mcp = FastMCP("Progress Server")
@mcp.tool()
async def long_task(steps: int, ctx: Context) -> str:
"""执行带有进度更新的任务。"""
await ctx.info("Starting task...")
for i in range(steps):
progress = (i + 1) / steps
await ctx.report_progress(
progress=progress,
total=1.0,
message=f"Step {i + 1}/{steps}",
)
await ctx.debug(f"Completed step {i + 1}")
return f"Task completed in {steps} steps"Database Integration with Lifespan
带生命周期的数据库集成
python
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppContext:
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP):
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
await db.disconnect()
mcp = FastMCP("DB Server", lifespan=app_lifespan)
@mcp.tool()
def query_tasks(user_id: str, ctx: Context) -> list:
"""Query tasks from database."""
app_ctx = ctx.request_context.lifespan_context
return app_ctx.db.query(f"SELECT * FROM tasks WHERE user_id = '{user_id}'")python
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppContext:
db: Database
@asynccontextmanager
async def app_lifespan(server: FastMCP):
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
await db.disconnect()
mcp = FastMCP("DB Server", lifespan=app_lifespan)
@mcp.tool()
def query_tasks(user_id: str, ctx: Context) -> list:
"""从数据库查询任务。"""
app_ctx = ctx.request_context.lifespan_context
return app_ctx.db.query(f"SELECT * FROM tasks WHERE user_id = '{user_id}'")Transport Options
传输方式选项
| Transport | Use Case | Command |
|---|---|---|
| stdio | CLI tools, local agents | |
| SSE | Web clients, real-time | |
| streamable-http | Production APIs | |
| 传输方式 | 使用场景 | 命令 |
|---|---|---|
| stdio | CLI工具、本地Agent | |
| SSE | Web客户端、实时场景 | |
| streamable-http | 生产环境API | |
Stateless vs Stateful
无状态 vs 有状态
python
undefinedpython
undefinedStateless (recommended for production)
无状态(推荐用于生产环境)
mcp = FastMCP("Server", stateless_http=True, json_response=True)
mcp = FastMCP("Server", stateless_http=True, json_response=True)
Stateful (maintains session state)
有状态(维护会话状态)
mcp = FastMCP("Server")
undefinedmcp = FastMCP("Server")
undefinedSecurity Considerations
安全注意事项
- Validate all inputs - Never trust user-provided data
- Use parameterized queries - Prevent SQL injection
- Authenticate requests - Verify user identity before operations
- Limit resource access - Only expose necessary data
- Log tool invocations - Audit trail for debugging
- 验证所有输入 - 绝不信任用户提供的数据
- 使用参数化查询 - 防止SQL注入
- 验证请求身份 - 操作前确认用户身份
- 限制资源访问 - 仅暴露必要的数据
- 记录工具调用 - 用于调试的审计跟踪
Troubleshooting
故障排查
Server won't start
服务器无法启动
- Check port availability
- Verify dependencies installed
- Check for syntax errors in tool definitions
- 检查端口是否可用
- 验证依赖是否已安装
- 检查工具定义中的语法错误
Client can't connect
客户端无法连接
- Verify transport matches (stdio/SSE/HTTP)
- Check CORS configuration for web clients
- Ensure MCP endpoint URL is correct
- 验证传输方式是否匹配(stdio/SSE/HTTP)
- 检查Web客户端的CORS配置
- 确保MCP端点URL正确
Tools not appearing
工具未显示
- Verify decorator is applied
@mcp.tool() - Check function has docstring (used as description)
- Restart server after adding new tools
- 确认已应用装饰器
@mcp.tool() - 检查函数是否有文档字符串(用作描述)
- 添加新工具后重启服务器".replace('"', '"'),