模块9: MCP — 构建你自己的服务器

模型上下文协议（MCP）是Anthropic创建的开放标准，用于将AI智能体连接到外部工具和数据源。在MCP之前，每个AI应用都必须为每个工具构建自定义集成 — 一个应用的Slack集成和另一个应用的Slack集成完全不同。MCP标准化了这些连接，使得一次构建的工具集成可以与任何兼容MCP的主机一起工作。

可以把它想象成AI的USB-C。就像USB-C在设备和外设之间提供通用连接器一样，MCP在AI智能体和工具之间提供通用协议。Anthropic于2025年12月将MCP捐赠给Linux基金会，目前所有语言的SDK月下载量已超过9700万次。

MCP之所以重要，是因为它解决了N x M集成问题。没有MCP时，N个AI应用各需要M个自定义集成，总共N*M个集成。有了MCP，每个工具构建一个服务器，每个应用构建一个客户端，总共只需N+M个集成。

MCP遵循客户端-服务器架构，有三个不同的角色。理解这些角色是正确构建和集成MCP服务器的关键。

通信流程为：用户 → 主机 → 客户端 → 服务器 → 外部系统。主机内的LLM决定调用哪些工具，客户端将请求发送到相应的服务器，服务器执行操作并返回结果，客户端将结果传回主机。

MCP原语

Python MCP SDK提供了 FastMCP，一个高级API，使构建服务器像编写带装饰器的Python函数一样简单。每个装饰过的函数都成为任何MCP主机可以发现和使用的工具、资源或提示。

pip install mcp[cli]

完整的示例服务器

from mcp.server.fastmcp import FastMCP
import json
import sqlite3
from datetime import datetime

# Create the MCP server
mcp = FastMCP("Company Tools", version="1.0.0")

# --- TOOLS (model-controlled) ---

@mcp.tool()
def get_stock_price(symbol: str) -> str:
    """Get the current stock price for a given ticker symbol.

    Args:
        symbol: Stock ticker symbol, e.g., 'AAPL', 'GOOGL', 'MSFT'

    Returns:
        Current price and daily change as a formatted string.
    """
    # In production, call a real API like Alpha Vantage or Yahoo Finance
    prices = {
        "AAPL": {"price": 227.50, "change": +1.2},
        "GOOGL": {"price": 178.30, "change": -0.8},
        "MSFT": {"price": 445.20, "change": +2.1},
    }
    data = prices.get(symbol.upper())
    if not data:
        return f"Unknown symbol: {symbol}"
    sign = "+" if data["change"] >= 0 else ""
    return f"{symbol.upper()}: ${data['price']:.2f} ({sign}{data['change']}%)"

@mcp.tool()
def query_employees(department: str = "", role: str = "") -> str:
    """Search the employee directory by department and/or role.

    Args:
        department: Filter by department name (optional)
        role: Filter by job role (optional)

    Returns:
        JSON array of matching employees with name, role, department, email.
    """
    conn = sqlite3.connect("company.db")
    conn.row_factory = sqlite3.Row
    query = "SELECT name, role, department, email FROM employees WHERE 1=1"
    params = []
    if department:
        query += " AND department LIKE ?"
        params.append(f"%{department}%")
    if role:
        query += " AND role LIKE ?"
        params.append(f"%{role}%")
    rows = conn.execute(query, params).fetchall()
    conn.close()
    return json.dumps([dict(r) for r in rows], indent=2)

@mcp.tool()
def create_calendar_event(
    title: str, date: str, time: str, duration_minutes: int = 60
) -> str:
    """Create a new calendar event.

    Args:
        title: Event title
        date: Date in YYYY-MM-DD format
        time: Start time in HH:MM format (24-hour)
        duration_minutes: Duration in minutes (default: 60)
    """
    # In production, call Google Calendar or Outlook API
    return json.dumps({
        "status": "created",
        "event": {"title": title, "date": date, "time": time,
                  "duration": duration_minutes}
    })

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

除了工具之外，MCP服务器还暴露资源（只读数据）和提示（可复用模板）。资源让主机应用可以浏览和读取数据，而无需LLM决定这样做。提示提供用户可以触发的预构建交互模式。

资源

@mcp.resource("company://docs/{doc_name}")
def get_document(doc_name: str) -> str:
    """Read a company document by name.

    Exposes documents like:
      company://docs/vacation-policy
      company://docs/expense-guidelines
    """
    docs_dir = "./company_docs"
    path = f"{docs_dir}/{doc_name}.md"
    try:
        with open(path, "r") as f:
            return f.read()
    except FileNotFoundError:
        return f"Document '{doc_name}' not found."

@mcp.resource("company://metrics/dashboard")
def get_dashboard_metrics() -> str:
    """Get current company metrics dashboard data."""
    # In production, query your metrics database
    return json.dumps({
        "revenue_mtd": "$2.4M",
        "active_users": 15420,
        "uptime": "99.97%",
        "open_tickets": 23
    }, indent=2)

提示模板

from mcp.server.fastmcp import Context

@mcp.prompt()
def review_code(language: str, code: str) -> str:
    """Review code for bugs, style issues, and improvements."""
    return f"""Please review the following {language} code for:
1. Bugs and potential errors
2. Style and readability issues
3. Performance improvements
4. Security concerns

Code to review:
```{language}
{code}
```

Provide specific, actionable feedback with code examples for each issue found."""

@mcp.prompt()
def summarise_meeting(notes: str) -> str:
    """Summarise meeting notes into action items and decisions."""
    return f"""Analyse these meeting notes and produce:
1. A 2-3 sentence summary
2. Key decisions made
3. Action items with owners and deadlines

Meeting notes:
{notes}"""

Claude Desktop是测试和个人使用中最常见的MCP主机。要连接您的服务器，您需要编辑Claude Desktop配置文件并添加您的服务器定义。

配置文件位置

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json

配置格式

{
  "mcpServers": {
    "company-tools": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"],
      "env": {
        "DATABASE_URL": "sqlite:///company.db",
        "API_KEY": "your-api-key-here"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxx"
      }
    }
  }
}

保存配置并重启Claude Desktop后，您的工具将出现在工具菜单中（锤子图标）。Claude会自动发现您服务器暴露的所有工具、资源和提示。

使用MCP Inspector测试

# Install the CLI tools
pip install "mcp[cli]"

# Launch the inspector (opens a web UI)
mcp dev server.py

# Or test from the command line
mcp run server.py

MCP生态系统自协议发布以来增长迅速。有大量用于常见集成的预构建服务器、多种语言的SDK，以及不断壮大的开发者社区在构建和共享服务器。