前言

Model Context Protocol (MCP) 是 Anthropic 推出的开放协议，用于标准化 LLM 与外部工具、数据源的交互方式。本文详细介绍 MCP 协议原理与服务器开发。

MCP 协议概述

架构设计

┌─────────────────────────────────────────────────────────────────┐
│                      MCP 架构                                   │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   ┌──────────────┐                      ┌──────────────┐       │
│   │   MCP Host   │                      │  MCP Server  │       │
│   │  (Claude等)  │◄────── MCP ──────────►│  (你的服务)  │       │
│   └──────────────┘      Protocol        └──────────────┘       │
│          │                                     │               │
│          │                                     │               │
│          ▼                                     ▼               │
│   ┌──────────────┐                      ┌──────────────┐       │
│   │  用户界面    │                      │   外部资源   │       │
│   │  (Claude)    │                      │  - 数据库    │       │
│   └──────────────┘                      │  - API       │       │
│                                         │  - 文件系统  │       │
│                                         └──────────────┘       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

核心概念

概念	说明	示例
Resources	暴露给 LLM 的数据/内容	文件、数据库记录
Tools	LLM 可调用的函数	搜索、计算、API调用
Prompts	预定义的提示模板	代码审查模板
Sampling	请求 LLM 生成	嵌套 LLM 调用

环境搭建

安装依赖

# Python
pip install mcp

# 或使用 uvx（推荐）
pip install uvx
uvx mcp install

# Node.js
npm install @modelcontextprotocol/sdk

项目结构

my-mcp-server/
├── src/
│   └── my_mcp_server/
│       ├── __init__.py
│       ├── server.py
│       └── tools/
│           ├── __init__.py
│           └── search.py
├── pyproject.toml
└── README.md

Python MCP Server 实战

Anthropic 提供了 FastMCP 高级库，极大简化了服务器的编写。

1. 使用 FastMCP 快速构建

from mcp.server.fastmcp import FastMCP

# 创建服务器实例
mcp = FastMCP("MyDatabaseServer")

# 1. 定义一个 Tool (LLM 可以主动调用的函数)
@mcp.tool()
def query_database(sql: str) -> str:
    """执行 SQL 查询并返回结果"""
    # 这里连接你的数据库
    return f"执行了查询: {sql}"

# 2. 定义一个 Resource (LLM 可以读取的数据源)
@mcp.resource("config://app-settings")
def get_config() -> str:
    """获取应用配置信息"""
    return "API_KEY=hidden, DEBUG=true"

# 3. 定义一个 Prompt (预定义的交互模板)
@mcp.prompt()
def code_review(code: str) -> str:
    return f"请对以下代码进行严格的审查，指出潜在的 Bug 和性能问题：\n\n{code}"

if __name__ == "__main__":
    mcp.run()

进阶：资源模板 (Resource Templates)

如果你有成千上万个文件或数据库记录，不可能一一手动定义。MCP 支持使用 URI 模板。

@mcp.resource("db://{table}/{id}")
def get_db_record(table: str, id: str) -> str:
    """动态获取数据库记录"""
    return f"Table: {table}, ID: {id} 的数据内容..."

部署与集成：Claude Desktop

要让 Claude Desktop 使用你的 MCP Server，你需要修改配置文件。

配置文件路径

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

配置示例

{
  "mcpServers": {
    "my-custom-server": {
      "command": "python",
      "args": ["/path/to/your/server.py"],
      "env": {
        "MY_API_KEY": "secret_value"
      }
    }
  }
}

安全性最佳实践

MCP Server 运行在你的本地或私有网络中，拥有极高的权限。

最小权限原则：不要直接暴露 os.system 或 rm -rf 等危险操作。
路径校验：如果提供文件读取功能，务必校验路径是否超出了预设的根目录（防止目录穿越攻击）。
输入验证：对 LLM 传入的参数进行严格的类型检查和长度限制。
只读资源：对于敏感数据，尽量通过 Resource（只读）而非 Tool（可写）暴露。

总结

MCP 协议是 LLM 生态系统中的“USB 接口”。

开发者：只需编写一次 MCP Server，就能让 Claude、Cursor 等所有支持 MCP 的客户端使用。
企业：可以安全地将内部私有数据（Jira, Confluence, DB）暴露给 AI，而无需担心数据被用于模型训练。

掌握 MCP，你就能将 LLM 变成一个真正懂你业务、能操作你工具的“超级员工”。 import httpx

创建服务器

mcp = FastMCP(“WeatherService”)

定义工具 (Tools)

@mcp.tool() async def get_weather(city: str) -> str: “"”获取指定城市的实时天气””” async with httpx.AsyncClient() as client: # 示例 API 调用 resp = await client.get(f”https://api.weather.com/{city}”) return f”{city} 的天气是：{resp.json()[‘weather’]}”

定义资源 (Resources)

@mcp.resource(“config://app”) def get_config() -> str: “"”获取应用配置资源””” return “App Config: Version 1.0.0”

定义提示模板 (Prompts)

@mcp.prompt() def weather_report_template(city: str) -> str: “"”生成天气报告的提示词模板””” return f”请为 {city} 生成一份详细的天气分析报告，包含穿衣建议。”

if name == “main”: mcp.run()

#### 2. SSE 传输协议 (HTTP/SSE)

除了默认的 `stdio`，MCP 还支持通过 HTTP 和 SSE 进行远程通信，适用于 Web 客户端。

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

server = Server("RemoteServer")
sse = SseServerTransport("/messages")

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as (read, write):
        await server.run(read, write, server.create_initialization_options())

app = Starlette(routes=[
    Route("/sse", endpoint=handle_sse),
    Route("/messages", endpoint=sse.handle_post_message, methods=["POST"]),
])

核心进阶：采样 (Sampling)

采样允许 MCP Server 反向请求 LLM 执行任务。例如，一个数据库工具可以请求 LLM：“请根据这些原始数据生成一段总结”。

@server.call_tool()
async def summarize_data(data: str):
    # 请求 LLM 进行采样
    result = await server.get_context().sample(
        prompt=f"请总结以下数据：{data}",
        max_tokens=100
    )
    return [TextContent(type="text", text=result.content)]

高级工具实现

# src/my_mcp_server/tools/database.py
from mcp.types import Tool, TextContent
import asyncpg
import json

class DatabaseTool:
    """数据库工具"""
    
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.pool = None
    
    async def connect(self):
        """连接数据库"""
        self.pool = await asyncpg.create_pool(self.connection_string)
    
    def get_tools(self) -> list[Tool]:
        """获取工具定义"""
        return [
            Tool(
                name="db_query",
                description="执行 SQL 查询",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "sql": {
                            "type": "string",
                            "description": "SQL 查询语句（只读）"
                        }
                    },
                    "required": ["sql"]
                }
            ),
            Tool(
                name="db_schema",
                description="获取数据库表结构",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "table_name": {
                            "type": "string",
                            "description": "表名"
                        }
                    },
                    "required": ["table_name"]
                }
            )
        ]
    
    async def call(self, name: str, arguments: dict) -> list[TextContent]:
        """调用工具"""
        if name == "db_query":
            return await self._query(arguments["sql"])
        elif name == "db_schema":
            return await self._get_schema(arguments["table_name"])
        raise ValueError(f"Unknown tool: {name}")
    
    async def _query(self, sql: str) -> list[TextContent]:
        """执行查询"""
        # 安全检查
        sql_lower = sql.lower().strip()
        if not sql_lower.startswith("select"):
            return [TextContent(
                type="text",
                text="Error: Only SELECT queries are allowed"
            )]
        
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(sql)
            results = [dict(row) for row in rows]
            
        return [TextContent(
            type="text",
            text=json.dumps(results, ensure_ascii=False, default=str)
        )]
    
    async def _get_schema(self, table_name: str) -> list[TextContent]:
        """获取表结构"""
        sql = """
            SELECT column_name, data_type, is_nullable
            FROM information_schema.columns
            WHERE table_name = $1
        """
        
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(sql, table_name)
            columns = [dict(row) for row in rows]
        
        return [TextContent(
            type="text",
            text=json.dumps(columns, ensure_ascii=False)
        )]

HTTP API 工具

# src/my_mcp_server/tools/api.py
import httpx
from mcp.types import Tool, TextContent
import json

class APITool:
    """HTTP API 工具"""
    
    def __init__(self, base_url: str, api_key: str = None):
        self.base_url = base_url
        self.api_key = api_key
    
    def get_tools(self) -> list[Tool]:
        """获取工具定义"""
        return [
            Tool(
                name="api_get",
                description="发送 GET 请求",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "endpoint": {
                            "type": "string",
                            "description": "API 端点路径"
                        },
                        "params": {
                            "type": "object",
                            "description": "查询参数"
                        }
                    },
                    "required": ["endpoint"]
                }
            ),
            Tool(
                name="api_post",
                description="发送 POST 请求",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "endpoint": {
                            "type": "string",
                            "description": "API 端点路径"
                        },
                        "data": {
                            "type": "object",
                            "description": "请求数据"
                        }
                    },
                    "required": ["endpoint", "data"]
                }
            )
        ]
    
    async def call(self, name: str, arguments: dict) -> list[TextContent]:
        """调用工具"""
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        
        async with httpx.AsyncClient() as client:
            if name == "api_get":
                response = await client.get(
                    f"{self.base_url}{arguments['endpoint']}",
                    params=arguments.get("params"),
                    headers=headers
                )
            elif name == "api_post":
                response = await client.post(
                    f"{self.base_url}{arguments['endpoint']}",
                    json=arguments["data"],
                    headers=headers
                )
            else:
                raise ValueError(f"Unknown tool: {name}")
        
        return [TextContent(
            type="text",
            text=json.dumps({
                "status": response.status_code,
                "data": response.json() if response.is_success else response.text
            }, ensure_ascii=False)
        )]

资源订阅

from mcp.server import Server
from mcp.types import Resource
import asyncio

server = Server("resource-server")

# 资源变更通知
class ResourceManager:
    def __init__(self, server: Server):
        self.server = server
        self.subscribers = set()
    
    async def notify_change(self, uri: str):
        """通知资源变更"""
        await self.server.notification(
            "notifications/resources/updated",
            {"uri": uri}
        )
    
    async def watch_file(self, path: str):
        """监视文件变化"""
        import watchfiles
        
        async for changes in watchfiles.awatch(path):
            for change_type, file_path in changes:
                await self.notify_change(f"file://{file_path}")

# 动态资源
@server.list_resources()
async def list_resources() -> list[Resource]:
    """列出资源（支持模板）"""
    return [
        Resource(
            uri="template://user/{user_id}/profile",
            name="用户资料",
            description="获取指定用户的资料"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    """读取资源"""
    import re
    
    # 匹配模板
    match = re.match(r"template://user/(\d+)/profile", uri)
    if match:
        user_id = match.group(1)
        return json.dumps({
            "id": user_id,
            "name": f"User {user_id}",
            "email": f"user{user_id}@example.com"
        })
    
    raise ValueError(f"Unknown resource: {uri}")

配置与部署

pyproject.toml

[project]
name = "my-mcp-server"
version = "1.0.0"
description = "My MCP Server"
requires-python = ">=3.10"
dependencies = [
    "mcp>=0.1.0",
    "httpx>=0.24.0",
    "asyncpg>=0.27.0"
]

[project.scripts]
my-mcp-server = "my_mcp_server.server:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

Claude 配置

{
    "mcpServers": {
        "my-server": {
            "command": "uvx",
            "args": ["my-mcp-server"],
            "env": {
                "DATABASE_URL": "postgresql://...",
                "API_KEY": "..."
            }
        }
    }
}

Docker 部署

FROM python:3.11-slim

WORKDIR /app

COPY pyproject.toml .
COPY src/ src/

RUN pip install -e .

CMD ["my-mcp-server"]

测试与调试

# tests/test_server.py
import pytest
from mcp.client import Client
from mcp.client.stdio import stdio_client

@pytest.fixture
async def client():
    """创建测试客户端"""
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "my_mcp_server.server"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with Client(read, write) as client:
            yield client

@pytest.mark.asyncio
async def test_list_tools(client):
    """测试工具列表"""
    tools = await client.list_tools()
    assert len(tools.tools) > 0
    
    tool_names = [t.name for t in tools.tools]
    assert "search_web" in tool_names

@pytest.mark.asyncio
async def test_call_tool(client):
    """测试工具调用"""
    result = await client.call_tool(
        "calculate",
        {"expression": "2 + 2"}
    )
    
    assert result.content[0].text == "4"

最佳实践

项目	建议
工具设计	单一职责，明确输入输出
错误处理	返回有意义的错误信息
安全性	验证输入，限制危险操作
性能	使用异步，避免阻塞
文档	清晰的 description

参考资源

（采用 CC BY-NC-SA 4.0 许可协议进行授权）

本文标题：《 LLM应用开发——MCP协议开发》

本文链接：http://localhost:3015/ai/MCP%E5%8D%8F%E8%AE%AE%E5%BC%80%E5%8F%91.html

本文最后一次更新为天前，文章中的某些内容可能已过时！

LLM应用开发——MCP协议开发

Model Context Protocol详解与实战

LLM 应用开发系列

前言