MCP 协议——部署一个 Model Context Protocol Server

日期: 2026-10-01 方向: AI系统工程 / Agent 阶段: Phase 3 - Agent架构与多Agent (Day 149-162) 标签: #MCP #ModelContextProtocol #ClaudeDesktop #ClaudeCode #StdioServer

---

今日目标

---

一、MCP 是什么 / 为什么需要

1.1 痛点

2024 年大量 LLM 应用都在做同一件事：把"工具"喂给 LLM。每家都自己造轮子：

• Anthropic SDK：tool_use
• OpenAI：function calling
• LangChain：BaseTool
• LlamaIndex：FunctionTool

结果：

• 工具不可复用——给 ChatGPT 写的 PDF tool，Claude 不能直接用
• 用户每装一个新 LLM 客户端要重新接所有工具
• 缺乏权限/凭据管理的标准

1.2 MCP 的定位

"USB-C for AI applications"

MCP（Model Context Protocol）是 Anthropic 2024-11 开源的协议，把 LLM 应用与"工具/资源/prompt"解耦：

┌──────────────────┐      MCP (JSON-RPC 2.0)       ┌──────────────────┐
│  Client (Host)   │◄──────────────────────────────►│   MCP Server     │
│                  │                                │                  │
│  e.g. Claude     │      stdio / sse / http        │  - Resources     │
│  Desktop /       │                                │  - Prompts       │
│  Claude Code /   │                                │  - Tools         │
│  Cursor / VS     │                                │                  │
│  Code            │                                │  (you build this)│
└──────────────────┘                                └──────────────────┘

1.3 三类原语

关键区分：tools 是 LLM 主动用，resources 是 user/client 决定加载到 context，prompts 是 UI slash command。

1.4 协议层

• 传输：stdio（本地进程）/ SSE（HTTP）/ Streamable HTTP（2025 推荐）
• 编码：JSON-RPC 2.0
• 能力协商：initialize 握手时 server 声明它支持哪些原语

1.5 当前生态（2026）

• 官方 SDK：Python (mcp), TypeScript (@modelcontextprotocol/sdk), Java, Rust, Swift
• 客户端：Claude Desktop, Claude Code, Cursor, VS Code, Continue, Zed
• 公共 servers: filesystem, github, postgres, slack, brave-search, sentry, sqlite, puppeteer, alchemy（Web3）, etc.
• 上千个第三方 server

---

二、架构图

┌────────────────────────────────────────────────────────────────────┐
│                         Claude Desktop                              │
│                                                                    │
│   User: "List my recent SEC filings for AAPL"                     │
│       │                                                            │
│       ▼                                                            │
│   Claude (claude-opus-4-7) sees registered MCP tools              │
│   from claude_desktop_config.json                                 │
│       │                                                            │
│       ▼ JSON-RPC over stdio                                       │
└──────────────────────┬─────────────────────────────────────────────┘
                       │
                       ▼
┌────────────────────────────────────────────────────────────────────┐
│              fin-mcp-server (Python, our code)                     │
│                                                                    │
│   initialize  ──► capabilities = {tools, resources, prompts}      │
│   tools/list  ──► [search_filings, fetch_section, ...]            │
│   tools/call  ──► search_filings(ticker=AAPL)                     │
│   resources/list ──► sec://AAPL/10-Q/2026-08-01                   │
│   resources/read ──► raw filing content                           │
│   prompts/list ──► ["credit_memo_template", "risk_review"]        │
│   prompts/get  ──► template messages                              │
│                                                                    │
│   stdio / SSE                                                     │
└────────────────────────────────────────────────────────────────────┘

---

三、代码——mcp_server.py

3.1 安装

pip install "mcp[cli]>=1.6.0"

3.2 服务器实现

# mcp_server.py
"""
Day 153 - A financial data MCP server.

Exposes:
  - 3 tools (search_filings, fetch_section, calc_ratio)
  - 1 resource template (sec://{ticker}/{form}/{date})
  - 2 prompts (credit_memo, risk_review)

Run:
  python mcp_server.py        # stdio mode (for Claude Desktop)
  python mcp_server.py --sse  # SSE mode for HTTP clients
"""
from __future__ import annotations
import asyncio
import json
import sys
from datetime import datetime
from typing import Any

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool, TextContent, Resource, ResourceTemplate,
    Prompt, PromptArgument, PromptMessage, GetPromptResult,
    ReadResourceResult,
)

server = Server("fin-mcp-server")

# ====================================================================
# Mock data
# ====================================================================
FILINGS = {
    "AAPL": [
        {"form": "10-Q", "date": "2026-08-01",
         "mda": "Revenue $94.9B (+3% YoY). Services $24.2B."},
        {"form": "10-K", "date": "2025-11-01",
         "mda": "FY25 revenue $385B. Services 25% of total."},
    ],
    "MSFT": [
        {"form": "10-Q", "date": "2026-07-25",
         "mda": "Revenue $65.6B (+15% YoY). Cloud $35B."},
    ],
}

# ====================================================================
# Tools
# ====================================================================
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_filings",
            description=(
                "Search SEC filings for a US-listed company. "
                "Returns a list of {form, date}. Use before fetch_section."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "ticker": {"type": "string", "pattern": "^[A-Z]{1,5}$"},
                    "form": {"type": "string", "enum": ["10-K", "10-Q", "8-K"]},
                },
                "required": ["ticker"],
            },
        ),
        Tool(
            name="fetch_section",
            description="Fetch the MD&A section of a specific filing.",
            inputSchema={
                "type": "object",
                "properties": {
                    "ticker": {"type": "string"},
                    "form": {"type": "string"},
                    "date": {"type": "string"},
                },
                "required": ["ticker", "form", "date"],
            },
        ),
        Tool(
            name="calc_ratio",
            description="Compute a financial ratio from two numbers.",
            inputSchema={
                "type": "object",
                "properties": {
                    "numerator": {"type": "number"},
                    "denominator": {"type": "number"},
                    "as_pct": {"type": "boolean", "default": False},
                },
                "required": ["numerator", "denominator"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    if name == "search_filings":
        ticker = arguments["ticker"].upper()
        form = arguments.get("form")
        items = FILINGS.get(ticker, [])
        if form:
            items = [i for i in items if i["form"] == form]
        return [TextContent(type="text", text=json.dumps([
            {"form": i["form"], "date": i["date"]} for i in items
        ]))]

    if name == "fetch_section":
        t, f, d = arguments["ticker"].upper(), arguments["form"], arguments["date"]
        for it in FILINGS.get(t, []):
            if it["form"] == f and it["date"] == d:
                return [TextContent(type="text", text=it["mda"])]
        return [TextContent(type="text", text=f"NOT_FOUND: {t}/{f}/{d}")]

    if name == "calc_ratio":
        n = arguments["numerator"]
        d = arguments["denominator"]
        if d == 0:
            return [TextContent(type="text", text=json.dumps({"error": "div_by_zero"}))]
        v = n / d
        if arguments.get("as_pct"):
            v *= 100
        return [TextContent(type="text", text=json.dumps({"result": round(v, 4)}))]

    return [TextContent(type="text", text=f"unknown_tool: {name}")]

# ====================================================================
# Resources
# ====================================================================
@server.list_resources()
async def list_resources() -> list[Resource]:
    out = []
    for ticker, items in FILINGS.items():
        for it in items:
            out.append(Resource(
                uri=f"sec://{ticker}/{it['form']}/{it['date']}",
                name=f"{ticker} {it['form']} {it['date']}",
                description=f"SEC {it['form']} filing for {ticker}",
                mimeType="text/plain",
            ))
    return out

@server.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
    return [
        ResourceTemplate(
            uriTemplate="sec://{ticker}/{form}/{date}",
            name="SEC filing",
            description="Access a specific SEC filing by ticker/form/date",
            mimeType="text/plain",
        ),
    ]

@server.read_resource()
async def read_resource(uri: str) -> ReadResourceResult:
    # uri: sec://AAPL/10-Q/2026-08-01
    if not uri.startswith("sec://"):
        raise ValueError(f"unsupported uri scheme: {uri}")
    parts = uri.removeprefix("sec://").split("/")
    if len(parts) != 3:
        raise ValueError(f"malformed: {uri}")
    ticker, form, date = parts
    for it in FILINGS.get(ticker.upper(), []):
        if it["form"] == form and it["date"] == date:
            return ReadResourceResult(contents=[
                TextContent(type="text", text=it["mda"])
            ])
    raise ValueError(f"not_found: {uri}")

# ====================================================================
# Prompts
# ====================================================================
@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="credit_memo",
            description="Generate a 1-page credit memo for a public company.",
            arguments=[
                PromptArgument(name="ticker", description="Stock ticker", required=True),
                PromptArgument(name="depth", description="brief|standard|deep", required=False),
            ],
        ),
        Prompt(
            name="risk_review",
            description="Review the Risk Factors section of the latest 10-K.",
            arguments=[
                PromptArgument(name="ticker", required=True),
            ],
        ),
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict[str, str] | None) -> GetPromptResult:
    args = arguments or {}
    if name == "credit_memo":
        ticker = args.get("ticker", "AAPL").upper()
        depth = args.get("depth", "standard")
        text = (
            f"You are a senior credit analyst. Produce a {depth} credit memo for "
            f"{ticker}.\n\n"
            f"Steps:\n"
            f"1. Use search_filings to find latest 10-Q.\n"
            f"2. Use fetch_section to read MD&A.\n"
            f"3. Compute key ratios with calc_ratio.\n"
            f"4. Write the memo with: business overview, financials, risks, recommendation.\n"
        )
        return GetPromptResult(
            description=f"Credit memo prompt for {ticker} ({depth})",
            messages=[PromptMessage(role="user", content=TextContent(type="text", text=text))],
        )

    if name == "risk_review":
        ticker = args.get("ticker", "AAPL").upper()
        text = (
            f"Read the Risk Factors section of {ticker}'s latest 10-K and produce a "
            f"prioritized list of top 5 risks with mitigations."
        )
        return GetPromptResult(
            description=f"Risk review for {ticker}",
            messages=[PromptMessage(role="user", content=TextContent(type="text", text=text))],
        )

    raise ValueError(f"unknown_prompt: {name}")

# ====================================================================
# Run
# ====================================================================
async def amain():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    if "--sse" in sys.argv:
        # Optional: SSE mode for HTTP-based clients
        from mcp.server.sse import SseServerTransport
        from starlette.applications import Starlette
        from starlette.routing import Route, Mount
        import uvicorn

        sse = SseServerTransport("/messages/")
        async def handle_sse(request):
            async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
                await server.run(streams[0], streams[1], server.create_initialization_options())

        app = Starlette(routes=[
            Route("/sse", endpoint=handle_sse),
            Mount("/messages/", app=sse.handle_post_message),
        ])
        uvicorn.run(app, host="0.0.0.0", port=8765)
    else:
        asyncio.run(amain())

3.3 Claude Desktop 配置

~/Library/Application Support/Claude/claude_desktop_config.json (macOS) or %APPDATA%\Claude\claude_desktop_config.json (Windows):

{
  "mcpServers": {
    "fin": {
      "command": "python",
      "args": ["/abs/path/to/mcp_server.py"],
      "env": {}
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem",
               "/Users/me/Documents/research"]
    }
  }
}

重启 Claude Desktop 后，Claude 就能看到 search_filings / fetch_section / calc_ratio tools，UI 里会出现 /credit_memo slash command 和 @sec://AAPL/10-Q/2026-08-01 resource autocomplete。

3.4 Claude Code 配置

.mcp.json 在项目根目录：

{
  "mcpServers": {
    "fin": {
      "command": "python",
      "args": ["./mcp_server.py"]
    }
  }
}

---

四、金融领域应用——内部知识库 MCP

把内部系统包装成 MCP server，分析师/客户经理用 Claude Desktop 直接对接：

典型部署：每个业务域一个 MCP server（CRM-MCP / Trade-MCP / Risk-MCP），员工本地装 Claude Desktop，配置自己有权限的 servers。鉴权由 server 内部用员工 OAuth token 完成。

---

五、Web3 集成——Web3 MCP servers

社区已经有几个 Web3 MCP server：

自己写一个 onchain MCP server 骨架

@server.list_tools()
async def list_tools():
    return [
        Tool(name="get_eth_balance", inputSchema={
            "type": "object",
            "properties": {"address": {"type": "string", "pattern": "^0x[a-fA-F0-9]{40}$"}},
            "required": ["address"],
        }, description="ETH balance of an address"),
        Tool(name="simulate_swap", inputSchema={...},
             description="Simulate a Uniswap swap (does NOT execute)"),
    ]

关键设计：write tool（实际执行交易）应该 不通过 MCP，由 user 在 wallet UI 里手签。MCP 仅暴露 read + simulate。这样即使 LLM 跑飞，也只是浪费 RPC 流量，不会动钱。

---

六、生产经验与陷阱

1. stdio 模式下的 print() 会污染协议 MCP server 通过 stdout 发 JSON-RPC，任何 print 都会破坏协议。日志必须走 stderr 或文件。Python 里：logging.basicConfig(stream=sys.stderr).

2. 大文件 resource 撑爆 context 返回 50MB filing 给 LLM = 直接炸。Resource 应当返回摘要/分块或要求 LLM 通过 tool 进一步获取 ranged content。

3. schema 版本兼容 2024-11 → 2024-12 → 2025 mid 协议有过 schema 改动（streamable HTTP 替代 SSE）。生产里 pin SDK 版本：mcp >= 1.6, < 2.0。

4. 权限耦合 把"列出客户"和"删除客户"放同一个 server，user 装一次就拥有两种权限。最小权限——为不同权限范围拆 server。

5. 冷启动慢 重 server（要连 DB、加载模型）每次 Claude Desktop 启动都要起，体验慢。两个办法：① 长驻 daemon + thin MCP wrapper；② SSE/HTTP server 共享一个长跑实例。

6. prompt 参数没校验 credit_memo(ticker="DROP TABLE")——prompt 字段直接被拼到模板。永远 sanitize，把 prompt 当成不可信输入。

7. 测试痛 Claude Desktop 调试不好看错误。开发时用 mcp dev mcp_server.py 命令启动 inspector（官方 SDK 自带），有 web UI 看请求响应。

---

七、Cost & Latency

MCP 本身不影响 LLM 成本（它只在 client 侧聚合 tools），但影响：

• First-call latency：cold start 1-3s（启 Python 进程）
• JSON-RPC overhead：每次 tool call 加 ~5ms（local stdio）
• SSE/HTTP：每次 + 30-100ms（视网络）

---

八、关键速查

MCP 三类原语决策

常用 MCP server 列表（2026）

Server 健康检查 checklist

• stdout 只有 JSON-RPC，无任何 print
• 错误用 JSON-RPC error response，不抛 Python exception
• Schema 严格（pattern / enum）
• 大输出有截断
• log 走 stderr / file
• 鉴权 token 不写死，从 env 读
• graceful shutdown

---

九、面试题

Q1: MCP 解决什么 OpenAI function calling 没解决的问题？

A: ① 跨厂商互操作——同一个工具能给 Claude / Cursor / Continue 用，无需重写；② 三类原语区分——把"LLM 自主调"（tools）、"user 加载"（resources）、"UI 模板"（prompts）拆开；③ 标准化部署——server 是独立进程/服务，权限/凭据隔离；④ 生态——已有上千个开源 server。Function calling 是 API schema，MCP 是协议+生态。

Q2: 在企业里部署 MCP，安全设计要点？

A: ① 每个 MCP server 跑在独立进程/容器，最小权限；② 鉴权 token 通过 env 注入，不写死；③ 敏感 destructive 工具单独 server，用户必须显式装；④ Server 输入做 schema 验证 + sanitize（防 prompt injection）；⑤ Audit log 所有 tool call 含调用者；⑥ 生产用 streamable HTTP + TLS + mTLS。

Q3: 解释 resource 和 tool 的区别。同一个 SEC filing，何时用哪个？

A: Resource 适合"用户/客户端主动引用进 context"——例如分析师从 UI 选 @sec://AAPL/10-Q，把它注入对话 context。Tool 适合"LLM 自主决定要不要调用"——LLM 看到 user 问"AAPL 服务收入是多少"，自己调 search_filings + fetch_section。两者底层数据可能相同，但 UX 完全不同：resource 是显式的，tool 是隐式的。

Q4: MCP server 生产部署遇到 stdout 污染、cold start 慢，如何解？

A: ① 所有日志走 stderr 或 logging file，仅 JSON-RPC 走 stdout；② cold start 用长驻 SSE/HTTP server（不每次启进程）；③ 重依赖（DB 连接、ML model）lazy load 但保活；④ 容器化部署，多实例 + load balancer；⑤ 健康检查 endpoint。

Q5: 设计一个内部 MCP 战略，10000 员工用 Claude Desktop 接到内部数据，怎么做？

A: ① 平台团队建 5-10 个核心 server（CRM/Trades/Risk/Compliance/Knowledge），各业务域单独维护；② 集中的 OAuth/SSO，员工 SSO 后下发 server token；③ Server 在内部 K8s，HTTPS+streamable HTTP；④ Claude Desktop 配置文件由 IT 推送，员工无感；⑤ Audit pipeline：所有 tool call 上报到 SIEM；⑥ 每个 server 有 owner + on-call。

---

明日预告

Day 154: A2A 通信——Agent 之间怎么对话

• 多 agent 系统的消息协议
• 共享 state vs message passing
• 实现 2 个 agent 协同对话
• AutoGen / CrewAI 内部如何做的