mcp-server（MCP server 部署）

MCP Server Skill

Build MCP (Model Context Protocol) servers using the official Python SDK with FastMCP high-level API.

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                 │
└─────────────────────────────────────────────────────────────────────────┘

Quick Start

Installation

# pip
pip install mcp

# poetry
poetry add mcp

# uv
uv add mcp

Environment Variables

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

Core Concepts

Basic MCP Server

from mcp.server.fastmcp import FastMCP

# Create server instance
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"}

# 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"

# 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."

# Run the server
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Reference

Examples

Templates

FastAPI/Starlette Integration

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 = FastMCP("Todo Server", stateless_http=True, json_response=True)

@mcp.tool()
def add_task(title: str) -> dict:
    """Add a task."""
    return {"status": "created"}

# Lifespan manager for session handling
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    async with mcp.session_manager.run():
        yield

# Create Starlette app with MCP mounted
app = Starlette(
    routes=[
        Mount("/mcp", app=mcp.streamable_http_app()),
    ],
    lifespan=lifespan,
)

# Add CORS for browser clients
app = CORSMiddleware(
    app,
    allow_origins=["*"],
    allow_methods=["GET", "POST", "DELETE"],
    expose_headers=["Mcp-Session-Id"],
)

# Run with: uvicorn server:app --reload
# MCP endpoint: http://localhost:8000/mcp/mcp

Tool with Context (Progress, Logging)

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"

Database Integration with Lifespan

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}'")

Transport Options

Stateless vs Stateful

# Stateless (recommended for production)
mcp = FastMCP("Server", stateless_http=True, json_response=True)

# Stateful (maintains session state)
mcp = FastMCP("Server")

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

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

Tools not appearing

• Verify @mcp.tool() decorator is applied
• Check function has docstring (used as description)
• Restart server after adding new tools