mcp-server

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

MCP 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
undefined
bash
undefined

pip

pip

pip install mcp
pip install mcp

poetry

poetry

poetry add mcp
poetry add mcp

uv

uv

uv add mcp
undefined
uv add mcp
undefined

Environment Variables

环境变量

env
DATABASE_URL=postgresql://user:password@localhost:5432/mydb
env
DATABASE_URL=postgresql://user:password@localhost:5432/mydb

Core Concepts

核心概念

ConceptDecoratorPurpose
Tools
@mcp.tool()
Perform actions, computations, side effects
Resources
@mcp.resource()
Expose data for reading (URI-based)
Prompts
@mcp.prompt()
Reusable prompt templates
概念装饰器用途
工具
@mcp.tool()
执行操作、计算和副作用
资源
@mcp.resource()
以URI形式暴露可读取的数据
提示词
@mcp.prompt()
可复用的提示词模板

Basic MCP Server

基础MCP服务器

python
from mcp.server.fastmcp import FastMCP
python
from mcp.server.fastmcp import FastMCP

Create 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")
undefined
if name == "main": mcp.run(transport="streamable-http")
undefined

Reference

参考文档

PatternGuide
Toolsreference/tools.md
Resourcesreference/resources.md
Promptsreference/prompts.md
Transportsreference/transports.md
FastAPI Integrationreference/fastapi-integration.md
模式指南
工具reference/tools.md
资源reference/resources.md
提示词reference/prompts.md
传输方式reference/transports.md
FastAPI集成reference/fastapi-integration.md

Examples

示例

ExampleDescription
examples/todo-server.mdComplete todo MCP server with CRUD tools
examples/database-integration.mdMCP server with SQLModel database
示例描述
examples/todo-server.md包含CRUD工具的完整待办事项MCP服务器
examples/database-integration.md集成SQLModel数据库的MCP服务器

Templates

模板

TemplatePurpose
templates/mcp_server.pyBasic MCP server template
templates/mcp_tools.pyTool definitions template
templates/mcp_fastapi.pyFastAPI + MCP integration template
模板用途
templates/mcp_server.py基础MCP服务器模板
templates/mcp_tools.py工具定义模板
templates/mcp_fastapi.pyFastAPI + 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 FastMCP
python
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCP

Create 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

undefined
undefined

Tool 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

传输方式选项

TransportUse CaseCommand
stdioCLI tools, local agents
mcp.run()
or
mcp.run(transport="stdio")
SSEWeb clients, real-time
mcp.run(transport="sse", port=8000)
streamable-httpProduction APIs
mcp.run(transport="streamable-http")
传输方式使用场景命令
stdioCLI工具、本地Agent
mcp.run()
mcp.run(transport="stdio")
SSEWeb客户端、实时场景
mcp.run(transport="sse", port=8000)
streamable-http生产环境API
mcp.run(transport="streamable-http")

Stateless vs Stateful

无状态 vs 有状态

python
undefined
python
undefined

Stateless (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")
undefined
mcp = FastMCP("Server")
undefined

Security Considerations

安全注意事项

  1. Validate all inputs - Never trust user-provided data
  2. Use parameterized queries - Prevent SQL injection
  3. Authenticate requests - Verify user identity before operations
  4. Limit resource access - Only expose necessary data
  5. Log tool invocations - Audit trail for debugging
  1. 验证所有输入 - 绝不信任用户提供的数据
  2. 使用参数化查询 - 防止SQL注入
  3. 验证请求身份 - 操作前确认用户身份
  4. 限制资源访问 - 仅暴露必要的数据
  5. 记录工具调用 - 用于调试的审计跟踪

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
    @mcp.tool()
    decorator is applied
  • Check function has docstring (used as description)
  • Restart server after adding new tools
  • 确认已应用
    @mcp.tool()
    装饰器
  • 检查函数是否有文档字符串(用作描述)
  • 添加新工具后重启服务器".replace('"', '"'),