MCP协议深度解析：2026年AI应用开发的核心标准，从原理到实战

AI客户端（Cursor/Claude/ChatGPT）
        ↕ MCP协议（JSON-RPC over stdio/SSE）
MCP客户端（内置于AI应用中）
        ↕
MCP服务器（你开发的工具集成）
        ↕
外部系统（数据库/API/文件系统）

{
  "name": "query_database",
  "description": "执行SQL查询并返回结果",
  "inputSchema": {
    "type": "object",
    "properties": {
      "sql": {"type": "string", "description": "SQL查询语句"}
    },
    "required": ["sql"]
  }
}

{
  "uri": "file:///project/src/main.ts",
  "name": "main.ts",
  "description": "项目主入口文件",
  "mimeType": "text/typescript"
}

{
  "name": "code_review",
  "description": "代码审查提示模板",
  "arguments": [
    {"name": "language", "description": "编程语言"}
  ]
}

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  { name: "my-mcp-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// 注册工具
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "get_weather",
      description: "获取指定城市的天气信息",
      inputSchema: {
        type: "object",
        properties: {
          city: { type: "string", description: "城市名称" }
        },
        required: ["city"]
      }
    }
  ]
}));

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  if (name === "get_weather") {
    const city = args.city;
    // 实际调用天气API
    const weather = await fetchWeather(city);
    return {
      content: [{ type: "text", text: JSON.stringify(weather) }]
    };
  }
});

// 启动服务器
const transport = new StdioServerTransport();
await server.connect(transport);

from mcp import Server
from mcp.types import Tool, TextContent

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

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="搜索技术文档",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        results = await search_documents(arguments["query"])
        return [TextContent(type="text", text=json.dumps(results))]

if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    
    async def main():
        async with stdio_server() as (read, write):
            await server.run(read, write)
    
    asyncio.run(main())