MCP 架构

┌─────────────────────────────────────────┐
│              Host (宿主应用)              │
│    (Claude Desktop / IDE / 你的应用)      │
├─────────────────────────────────────────┤
│  ┌─────────────┐    ┌─────────────┐     │
│  │  Client 1   │    │  Client 2   │     │
│  │ (连接Server)│    │ (连接Server)│     │
│  └──────┬──────┘    └──────┬──────┘     │
└─────────┼──────────────────┼────────────┘
          │                  │
    ┌─────┴─────┐      ┌─────┴─────┐
    │ MCP Server│      │ MCP Server│
    │ (工具/数据)│      │ (工具/数据)│
    └───────────┘      └───────────┘

核心角色：

• Host：承载 LLM 的应用（如 Claude Desktop）
• Client：Host 内的 MCP 客户端，维护与 Server 的连接
• Server：提供资源、工具、Prompts 的独立进程

---

📦 MCP 三大核心概念

1. Resources（资源）

类似 REST API 的 GET，提供只读数据：

• 文件内容
• 数据库查询结果
• API 响应数据

{
  "uri": "file:///project/src/main.py",
  "mimeType": "text/x-python",
  "text": "def hello(): ..."
}

2. Tools（工具）

LLM 可调用的函数，可以产生副作用：

• 执行命令
• 修改文件
• 调用外部 API

{
  "name": "execute_command",
  "description": "执行 shell 命令",
  "inputSchema": {
    "type": "object",
    "properties": {
      "command": { "type": "string" }
    },
    "required": ["command"]
  }
}

3. Prompts（提示词模板）

预定义的提示词，可参数化：

{
  "name": "review_code",
  "description": "代码审查模板",
  "arguments": [
    { "name": "code", "required": true }
  ]
}

---

🔌 传输层：stdio vs SSE

MCP 支持两种传输方式：

方式	适用场景	特点
stdio	本地 Server	通过 stdin/stdout 通信，最简单
SSE	远程 Server	HTTP + Server-Sent Events，支持网络部署

---

📝 从零实现一个 MCP Server

下面是一个完整的 stdio 模式 MCP Server 示例（Python）：

1. 基础 Server 结构

# weather_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent
import asyncio

# 创建 Server 实例
app = Server("weather-server")

@app.list_resources()
async def list_resources() -> list[Resource]:
    """列出可用资源"""
    return [
        Resource(
            uri="weather://beijing/current",
            name="北京当前天气",
            mimeType="application/json",
            description="北京市实时天气数据"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """读取资源内容"""
    if uri == "weather://beijing/current":
        return '{"temp": 25, "condition": "sunny"}'
    raise ValueError(f"Unknown resource: {uri}")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """列出可用工具"""
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的天气",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {"type": "string", "enum": ["C", "F"], "default": "C"}
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """执行工具调用"""
    if name == "get_weather":
        city = arguments["city"]
        unit = arguments.get("unit", "C")
        # 实际这里调用天气 API
        result = f"{city} 当前温度: 25°{unit}, 晴朗"
        return [TextContent(type="text", text=result)]
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

2. 配置 Claude Desktop 使用

编辑 ~/Library/Application Support/Claude/claude_desktop_config.json（macOS）：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"],
      "env": {
        "API_KEY": "your-api-key"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/xxx/Desktop"]
    }
  }
}

3. 测试流程

1. 重启 Claude Desktop
2. 看到工具图标出现即连接成功
3. 对话中可以直接使用：
   • "查一下北京天气" → 调用 get_weather 工具
   • "读取 weather://beijing/current" → 读取资源

---

🔧 进阶：SSE 模式部署

适合远程服务或需要 HTTP 接口的场景：

from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

sse = SseServerTransport("/messages/")

async def handle_sse(request):
    async with sse.connect_sse(
        request.scope, request.receive, request.send
    ) as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

async def handle_messages(request):
    await sse.handle_post_message(request.scope, request.receive, request.send)

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages/", endpoint=handle_messages, methods=["POST"]),
    ]
)

---

🎯 设计建议

1. 粒度控制：Tools 应该原子化，一个 Tool 做一件事
2. 描述清晰：Tool 的 description 直接影响 LLM 调用决策
3. 错误处理：返回明确的错误信息，LLM 会据此调整策略
4. 安全边界：Tools 应该有权限控制，避免危险操作

---

📚 生态现状

• 官方 SDKs：Python @modelcontextprotocol/sdk, TypeScript @modelcontextprotocol/sdk
• 官方 Servers：文件系统、Git、PostgreSQL、Brave 搜索等
• 社区 Servers：Slack、Notion、Figma、Spotify 等集成