MCP协议开发实战：从零搭建AI Agent工具链，终于有人讲清楚了！

小二·

121人浏览 · 2026-07-28 11:35:25

小二· · 2026-07-28 11:35:25 发布

📅 2024年7月28日，Model Context Protocol（MCP）正式规范发布，AI Agent开发正式进入"即插即用"时代。本文从协议原理、架构设计到完整代码实现，手把手带你从零掌握MCP开发。

在这里插入图片描述

前言

大型语言模型（LLM）正从"对话机器人"进化为"智能代理Agent"。然而现实很骨感：每个AI应用想要调用外部工具，都得自己写一套插件逻辑——天气插件、数据库插件、文件读写插件……协议不统一、代码不可复用、生态极度碎片化。

Model Context Protocol（MCP） 的出现，就是为了解决这个问题。它就像AI世界的"USB接口"——一套协议，统一规范，插上就能用。

本文将深入MCP协议的每一个核心细节，配合完整的Python + TypeScript双语言代码示例，让你真正从零掌握MCP开发。全文硬核，建议先收藏。

一、MCP协议背景：7月28日，正式启航

1.1 为什么需要MCP？

让我们先看一个真实的开发困境：

传统方案：每个LLM应用 × 每个工具 = N×M 个集成

一家公司接入了ChatGPT、Claude、Gemini三个模型，又需要天气、地图、数据库三个工具，就需要维护9套集成代码。每个工具的认证方式、数据格式、调用接口各不相同，维护成本极高。

MCP的出现，将这个复杂度降为 N + M：

MCP方案：LLM应用（支持MCP）× MCP Server（工具生态）= 自由组合

1.2 官方里程碑

时间	事件
2023年11月	Anthropic开源MCP协议早期版本
2024年5月	社区生态快速扩张，Server数量破百
2024年7月28日	MCP 1.0 正式规范发布，协议稳定
2024年后	VS Code、Cursor、Claude Desktop等主流工具全面支持

1.3 MCP vs 既有方案对比

特性	MCP	Function Calling	LangChain Tools
标准化程度	✅ 官方统一协议	⚠️ 各家实现各异	⚠️ 框架绑定
双向通信	✅ 支持	❌ 仅单向调用	❌ 仅单向调用
工具发现	✅ 内置机制	❌ 手动注册	⚠️ 部分支持
安全模型	✅ 细粒度权限	❌ 应用层负责	❌ 应用层负责
多语言SDK	✅ Python/TS/Go	⚠️ 依赖模型厂商	✅ 官方支持

二、MCP核心概念：Model Context Protocol详解

2.1 协议架构一览

MCP采用客户端-服务器架构，整体分为三层：

┌─────────────────────────────────────────┐
│           MCP Host（主机环境）            │
│  Claude Desktop / VS Code / 自研应用     │
└────────────────┬────────────────────────┘
                 │  stdio / HTTP + SSE
┌────────────────▼────────────────────────┐
│       MCP Client（客户端SDK）            │
│  管理连接生命周期、协议握手、消息路由      │
└────────────────┬────────────────────────┘
                 │  JSON-RPC 2.0
┌────────────────▼────────────────────────┐
│       MCP Server（服务端SDK）            │
│  暴露工具(Tools)、资源(Resources)、      │
│  提示模板(Prompts)                       │
└─────────────────────────────────────────┘

2.2 三大核心能力

MCP Server可以向外暴露三类能力：

① Tools（工具）— AI主动调用

// 工具：AI Agent可以"调用"的函数
// AI模型生成参数 → MCP Server执行 → 返回结果给AI
{
  name: "get_weather",
  description: "查询指定城市的天气",
  inputSchema: {
    type: "object",
    properties: {
      city: { type: "string", description: "城市名称" },
      unit: { type: "string", enum: ["celsius", "fahrenheit"] }
    },
    required: ["city"]
  }
}

② Resources（资源）— AI按需读取

// 资源：AI可以"读取"的数据，类似文件系统
// AI模型在生成响应前主动读取上下文数据
{
  uri: "file:///project/config.json",
  name: "项目配置文件",
  mimeType: "application/json"
}

③ Prompts（提示模板）— AI可复用的提示词

// 提示模板：预定义的提示词片段，可带参数
{
  name: "code_review",
  description: "代码审查模板",
  arguments: [
    { name: "language", required: true },
    { name: "diff", required: true }
  ]
}

2.3 传输层：两种通信方式

MCP支持两种传输机制，适用不同场景：

传输方式	协议	适用场景	特点
stdio	标准输入/输出	本地进程、CLI工具	简单、安全、无网络依赖
HTTP + SSE	HTTP + Server-Sent Events	远程服务、分布式部署	支持网络调用，可水平扩展

大多数本地开发场景（如Claude Desktop插件）使用 stdio 模式。

2.4 消息协议：JSON-RPC 2.0

所有MCP通信均基于 JSON-RPC 2.0 规范：

// 请求示例
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": { "city": "北京", "unit": "celsius" }
  }
}

// 响应示例
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "北京今天晴，温度25°C，适合出行。"
      }
    ]
  }
}

// 错误示例
{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params: city is required"
  }
}

三、无状态架构设计原理

3.1 为什么MCP选择无状态设计？

这是MCP最容易被误解的设计决策之一。很多人会问：“AI Agent需要上下文，为什么协议是无状态的？”

答案在于关注点分离（Separation of Concerns）：

┌─────────────────────┐    有状态（会话级别）    ┌─────────────────────┐
│   AI Host（上层应用）  │ ◄───────────────────► │  对话历史、上下文窗口  │
│  管理会话、决策、UI    │                       │  用户偏好、任务状态   │
└──────────┬──────────┘                       └─────────────────────┘
           │ 无状态（协议级别）
┌──────────▼──────────┐
│   MCP Server        │
│  工具执行、数据获取   │
│  不关心"谁在调用"    │
└─────────────────────┘

无状态的优势：

横向扩展：Server可以部署多个副本，通过负载均衡分担请求
幂等性：相同的请求总能得到相同的结果（工具层面）
调试友好：每个请求都是完整的，日志记录简单清晰
安全隔离：Server不存储会话状态，被攻击后无法追溯历史

3.2 有状态需求如何满足？

如果你的工具确实需要状态（例如购物车、用户会话），MCP的设计哲学是：将状态存储在上层（Host或工具调用方），通过参数传递：

# ❌ 反面示例：在Server中存储会话状态
class ShoppingCartServer:
    def __init__(self):
        self.carts = {}  # 不推荐！违反无状态原则

# ✅ 正面示例：将状态ID作为参数传递
class ShoppingCartServer:
    @mcp.tool()
    def add_to_cart(self, cart_id: str, item: str):
        """cart_id由客户端管理，Server只执行操作"""
        # cart_id 对应客户端维护的购物车状态
        return {"success": True, "cart_id": cart_id, "item": item}

3.3 协议握手流程

完整的MCP连接建立过程：

Client                               Server
  │                                    │
  │ ─── initialize (capabilities) ───► │  1. 发送客户端能力
  │                                    │
  │ ◄── initialized (serverInfo) ──── │  2. 确认协议版本，交换元信息
  │                                    │
  │ ─── tools/list ──────────────────► │  3. 发现可用工具
  │ ◄── tools/list (result) ────────── │  4. 获取工具清单
  │                                    │
  │ ─── tools/call ──────────────────► │  5. 调用工具
  │ ◄── tools/call (result) ────────── │  6. 返回执行结果
  │                                    │
  │ ◄══ notification (心跳/错误) ══════ │  7. 服务端推送（如有必要）

四、实战：搭建MCP Server（完整代码）

4.1 环境准备

# Python SDK
pip install mcp

# TypeScript SDK
npm install @modelcontextprotocol/sdk
# 或使用 npx 直接运行（无需安装）

4.2 Python版：完整MCP Server实现

下面实现一个多功能工具服务器，包含文件操作、天气查询、数据库查询三个场景：

#!/usr/bin/env python3
"""
MCP Server 完整实现示例
功能：文件管理 + 天气查询 + 数据库操作
传输方式：stdio（默认）
"""

import json
import os
from pathlib import Path
from typing import Any
import sqlite3
from datetime import datetime

# MCP SDK（2024年7月后推荐使用mcp包）
try:
    from mcp.server import Server
    from mcp.server.stdio import stdio_server
    from mcp.types import (
        Tool,
        TextContent,
        CallToolResult,
        ListToolsResult,
    )
except ImportError:
    # 兼容旧版本
    from mcp import Server, stdio_server, Tool, TextContent

# ============================================================
# 第一部分：工具定义
# ============================================================

# 工具1：读取文件
async def read_file_handler(arguments: dict) -> CallToolResult:
    """
    读取指定路径的文件内容
    """
    file_path = arguments.get("path")
    if not file_path:
        return CallToolResult(
            content=[TextContent(type="text", text="错误：缺少 path 参数")],
            isError=True
        )
    
    try:
        # 安全检查：防止路径遍历攻击
        resolved = Path(file_path).resolve()
        if not resolved.exists():
            return CallToolResult(
                content=[TextContent(type="text", text=f"文件不存在: {file_path}")],
                isError=True
            )
        
        with open(resolved, "r", encoding="utf-8") as f:
            content = f.read()
        
        # 限制返回内容长度
        max_chars = arguments.get("max_chars", 10000)
        if len(content) > max_chars:
            content = content[:max_chars] + f"\n\n... (内容已截断，共 {len(content)} 字符)"
        
        return CallToolResult(
            content=[TextContent(type="text", text=content)]
        )
    except Exception as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"读取文件失败: {str(e)}")],
            isError=True
        )

# 工具2：模拟天气查询
async def get_weather_handler(arguments: dict) -> CallToolResult:
    """
    查询指定城市的天气信息（模拟数据）
    实际项目中替换为真实API调用
    """
    city = arguments.get("city", "")
    unit = arguments.get("unit", "celsius")
    
    if not city:
        return CallToolResult(
            content=[TextContent(type="text", text="错误：缺少 city 参数")],
            isError=True
        )
    
    # 模拟天气数据（实际使用时接入 weather API）
    weather_data = {
        "北京": {"temp": 28, "condition": "晴", "humidity": 45, "wind": "东南风3级"},
        "上海": {"temp": 33, "condition": "多云", "humidity": 72, "wind": "东风2级"},
        "深圳": {"temp": 35, "condition": "雷阵雨", "humidity": 85, "wind": "西南风4级"},
        "成都": {"temp": 26, "condition": "阴", "humidity": 68, "wind": "北风1级"},
    }
    
    data = weather_data.get(city, None)
    if not data:
        return CallToolResult(
            content=[TextContent(type="text", text=f"未找到城市 '{city}' 的天气数据")],
            isError=True
        )
    
    temp = data["temp"]
    if unit == "fahrenheit":
        temp = temp * 9 / 5 + 32
        unit_label = "°F"
    else:
        unit_label = "°C"
    
    result = f"""{city}当前天气：
🌡️ 温度：{temp}{unit_label}
🌤️ 天气：{data['condition']}
💧 湿度：{data['humidity']}%
🌬️ 风力：{data['wind']}
⏰ 查询时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""
    
    return CallToolResult(content=[TextContent(type="text", text=result)])

# 工具3：SQLite数据库查询
async def query_db_handler(arguments: dict) -> CallToolResult:
    """
    执行只读SQL查询
    仅支持 SELECT 语句，防止注入
    """
    query = arguments.get("query", "")
    db_path = arguments.get("db_path", "data.db")
    
    if not query:
        return CallToolResult(
            content=[TextContent(type="text", text="错误：缺少 query 参数")],
            isError=True
        )
    
    # 安全检查：仅允许 SELECT
    if not query.strip().upper().startswith("SELECT"):
        return CallToolResult(
            content=[TextContent(type="text", text="安全限制：仅支持 SELECT 查询")],
            isError=True
        )
    
    try:
        conn = sqlite3.connect(db_path)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute(query)
        rows = cursor.fetchall()
        conn.close()
        
        if not rows:
            return CallToolResult(
                content=[TextContent(type="text", text="查询结果为空")]
            )
        
        # 格式化输出
        headers = rows[0].keys()
        header_line = " | ".join(headers)
        separator = "-" * len(header_line)
        
        result_lines = [f"查询到 {len(rows)} 条记录：", "", header_line, separator]
        for row in rows:
            result_lines.append(" | ".join(str(row[h]) for h in headers))
        
        return CallToolResult(
            content=[TextContent(type="text", text="\n".join(result_lines))]
        )
    except Exception as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"数据库查询失败: {str(e)}")],
            isError=True
        )

# ============================================================
# 第二部分：Server主程序
# ============================================================

# 创建Server实例
server = Server("demo-mcp-server")

@server.list_tools()
async def list_tools() -> ListToolsResult:
    """
    返回所有可用工具的元信息
    AI模型通过这个列表了解可以调用哪些工具
    """
    return ListToolsResult(tools=[
        Tool(
            name="read_file",
            description="读取指定路径的文件内容。用于查看文本文件、配置文件、代码等。注意：仅支持文本文件。",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件路径（绝对路径或相对路径）"
                    },
                    "max_chars": {
                        "type": "integer",
                        "description": "最大返回字符数，默认10000",
                        "default": 10000
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="get_weather",
            description="查询指定城市的当前天气信息，包括温度、天气状况、湿度和风力。",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如：北京、上海、深圳"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认 celsius",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="query_database",
            description="在SQLite数据库中执行只读SELECT查询，返回格式化结果。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL SELECT查询语句"
                    },
                    "db_path": {
                        "type": "string",
                        "description": "数据库文件路径，默认 data.db",
                        "default": "data.db"
                    }
                },
                "required": ["query"]
            }
        ),
    ])

@server.call_tool()
async def call_tool(name: str, arguments: Any) -> CallToolResult:
    """
    工具调用入口：根据工具名路由到对应的处理函数
    """
    handler_map = {
        "read_file": read_file_handler,
        "get_weather": get_weather_handler,
        "query_database": query_db_handler,
    }
    
    handler = handler_map.get(name)
    if not handler:
        return CallToolResult(
            content=[TextContent(type="text", text=f"未知工具: {name}")],
            isError=True
        )
    
    return await handler(arguments)

# ============================================================
# 第三部分：启动Server
# ============================================================

async def main():
    """stdio模式启动，Claude Desktop等Host会自动连接"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

4.3 TypeScript版：完整MCP Server实现

#!/usr/bin/env npx ts-node
/**
 * MCP Server - TypeScript 实现
 * 功能：文件管理 + 天气查询 + 数据库操作
 */

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  Tool,
} from "@modelcontextprotocol/sdk/types.js";
import * as fs from "fs";
import * as path from "path";

// ============================================================
// 第一部分：工具定义
// ============================================================

interface ToolArguments {
  path?: string;
  max_chars?: number;
  city?: string;
  unit?: "celsius" | "fahrenheit";
  query?: string;
  db_path?: string;
}

// 工具1：读取文件
async function readFileHandler(args: ToolArguments): Promise<{ content: { type: "text"; text: string }[]; isError?: boolean }> {
  if (!args.path) {
    return { content: [{ type: "text", text: "错误：缺少 path 参数" }], isError: true };
  }

  try {
    const resolved = path.resolve(args.path);
    if (!fs.existsSync(resolved)) {
      return { content: [{ type: "text", text: `文件不存在: ${args.path}` }], isError: true };
    }

    let content = fs.readFileSync(resolved, "utf-8");
    const maxChars = args.max_chars ?? 10000;
    if (content.length > maxChars) {
      content = content.slice(0, maxChars) + `\n\n... (内容已截断，共 ${content.length} 字符)`;
    }

    return { content: [{ type: "text", text: content }] };
  } catch (e) {
    return { content: [{ type: "text", text: `读取文件失败: ${e}` }], isError: true };
  }
}

// 工具2：天气查询
async function getWeatherHandler(args: ToolArguments): Promise<{ content: { type: "text"; text: string }[]; isError?: boolean }> {
  const { city, unit = "celsius" } = args;
  if (!city) {
    return { content: [{ type: "text", text: "错误：缺少 city 参数" }], isError: true };
  }

  const weatherData: Record<string, { temp: number; condition: string; humidity: number; wind: string }> = {
    "北京": { temp: 28, condition: "晴", humidity: 45, wind: "东南风3级" },
    "上海": { temp: 33, condition: "多云", humidity: 72, wind: "东风2级" },
    "深圳": { temp: 35, condition: "雷阵雨", humidity: 85, wind: "西南风4级" },
    "成都": { temp: 26, condition: "阴", humidity: 68, wind: "北风1级" },
  };

  const data = weatherData[city];
  if (!data) {
    return { content: [{ type: "text", text: `未找到城市 '${city}' 的天气数据` }], isError: true };
  }

  const temp = unit === "fahrenheit" ? data.temp * 9 / 5 + 32 : data.temp;
  const unitLabel = unit === "fahrenheit" ? "°F" : "°C";

  const result = `${city}当前天气：
🌡️ 温度：${temp}${unitLabel}
🌤️ 天气：${data.condition}
💧 湿度：${data.humidity}%
🌬️ 风力：${data.wind}`;

  return { content: [{ type: "text", text: result }] };
}

// ============================================================
// 第二部分：Server主程序
// ============================================================

const server = new Server(
  {
    name: "demo-mcp-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},  // 声明支持工具调用能力
    },
  }
);

// 注册工具列表
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "read_file",
        description: "读取指定路径的文件内容，用于查看文本文件、配置文件、代码等。",
        inputSchema: {
          type: "object",
          properties: {
            path: {
              type: "string",
              description: "文件路径（绝对路径或相对路径）",
            },
            max_chars: {
              type: "integer",
              description: "最大返回字符数，默认10000",
              default: 10000,
            },
          },
          required: ["path"],
        },
      } as Tool,
      {
        name: "get_weather",
        description: "查询指定城市的当前天气信息，包括温度、天气状况、湿度和风力。",
        inputSchema: {
          type: "object",
          properties: {
            city: {
              type: "string",
              description: "城市名称，例如：北京、上海、深圳",
            },
            unit: {
              type: "string",
              enum: ["celsius", "fahrenheit"],
              description: "温度单位，默认 celsius",
              default: "celsius",
            },
          },
          required: ["city"],
        },
      } as Tool,
    ],
  };
});

// 注册工具调用处理器
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "read_file":
      return await readFileHandler(args as ToolArguments);
    case "get_weather":
      return await getWeatherHandler(args as ToolArguments);
    default:
      return { content: [{ type: "text", text: `未知工具: ${name}` }], isError: true };
  }
});

// ============================================================
// 第三部分：启动Server
// ============================================================

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server 已启动，等待连接...");  // 注意：stderr用于日志，stdout用于协议
}

main().catch(console.error);

4.4 在Claude Desktop中配置使用

创建配置文件 ~/.claude/desktops/config.json（Windows: %USERPROFILE%\.claude\desktops\config.json）：

{
  "mcpServers": {
    "demo-server": {
      "command": "python",
      "args": ["C:/path/to/your/mcp_server.py"]
    }
  }
}

重启Claude Desktop，即可在设置中看到新工具。

五、工具定义与注册：深入inputSchema

5.1 inputSchema的艺术

inputSchema是MCP中工具定义的灵魂。一个好的schema能让AI模型准确理解工具的用法，减少幻觉（hallucination）调用。

// ❌ 糟糕的Schema - AI很难猜到如何调用
{
  name: "process",
  inputSchema: {
    "type": "object",
    "properties": {
      "data": { "type": "string" }
    }
  }
}

// ✅ 优秀的Schema - AI能准确理解每个参数
{
  name: "create_calendar_event",
  description: "在日历中创建新事件，支持单次事件和重复事件",
  inputSchema: {
    "type": "object",
    "properties": {
      "title": {
        "type": "string",
        "description": "事件标题，简明扼要，例如：'团队周会'",
        minLength: 1,
        maxLength: 100
      },
      "start_time": {
        "type": "string",
        "description": "开始时间，ISO 8601格式，例如：'2024-07-28T14:00:00+08:00'",
        format: "date-time"
      },
      "end_time": {
        "type": "string",
        "description": "结束时间，ISO 8601格式"
      },
      "attendees": {
        "type": "array",
        "description": "参与者邮箱列表",
        "items": { "type": "string", "format": "email" },
        "default": []
      },
      "recurrence": {
        "type": "string",
        "description": "重复规则，RFC 5545 RRULE格式",
        "enum": ["NONE", "DAILY", "WEEKLY", "MONTHLY"],
        "default": "NONE"
      },
      "location": {
        "type": "string",
        "description": "会议地点或视频会议链接"
      }
    },
    "required": ["title", "start_time"],
    "additionalProperties": false  // 拒绝未知参数，防止AI注入
  }
}

5.2 嵌套对象与数组Schema

# 复杂Schema示例：批量处理任务
Tool(
    name="batch_process",
    description="批量处理一组数据转换任务",
    inputSchema={
        "type": "object",
        "properties": {
            "tasks": {
                "type": "array",
                "description": "任务列表，每个任务包含转换规则",
                "items": {
                    "type": "object",
                    "properties": {
                        "id": {"type": "string", "description": "任务唯一标识"},
                        "input": {"type": "string", "description": "输入内容"},
                        "operations": {
                            "type": "array",
                            "description": "要执行的操作序列",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "type": {
                                        "type": "string",
                                        "enum": ["uppercase", "lowercase", "trim", "replace"],
                                        "description": "操作类型"
                                    },
                                    "params": {
                                        "type": "object",
                                        "description": "操作参数",
                                        "properties": {
                                            "pattern": {"type": "string"},
                                            "replacement": {"type": "string"}
                                        }
                                    }
                                },
                                "required": ["type"]
                            }
                        }
                    },
                    "required": ["id", "input"]
                },
                "minItems": 1,
                "maxItems": 100
            },
            "parallel": {
                "type": "boolean",
                "description": "是否并行执行，默认false（串行）",
                "default": False
            }
        },
        "required": ["tasks"]
    }
)

六、客户端调用实现

6.1 Python客户端调用

#!/usr/bin/env python3
"""
MCP客户端示例 - 演示如何通过代码调用MCP Server
"""

import asyncio
import json

try:
    from mcp.client import ClientSession
    from mcp.client.stdio import stdio_client
except ImportError:
    print("请安装: pip install mcp")
    exit(1)

async def main():
    # 方式一：连接本地stdio Server
    async with stdio_client(
        command="python",
        args=["mcp_server.py"]  # 你的Server路径
    ) as (read, write):
        async with ClientSession(read, write) as session:
            # 1. 初始化连接
            await session.initialize()
            print("✅ 已连接到MCP Server")
            
            # 2. 发现可用工具
            tools = await session.list_tools()
            print(f"\n可用工具 ({len(tools.tools)}个)：")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            # 3. 调用工具：查询天气
            print("\n📡 调用 get_weather...")
            result = await session.call_tool(
                "get_weather",
                arguments={"city": "北京", "unit": "celsius"}
            )
            print(f"结果: {result.content[0].text}")
            
            # 4. 调用工具：读取文件
            print("\n📡 调用 read_file...")
            result = await session.call_tool(
                "read_file",
                arguments={"path": "README.md", "max_chars": 500}
            )
            print(f"结果: {result.content[0].text[:200]}...")
            
            # 5. 批量调用演示
            print("\n📡 批量查询天气...")
            cities = ["上海", "深圳", "成都"]
            tasks = [
                session.call_tool("get_weather", {"city": city})
                for city in cities
            ]
            results = await asyncio.gather(*tasks)
            for city, res in zip(cities, results):
                print(f"  {city}: {res.content[0].text.split(chr(10))[0]}")

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

6.2 TypeScript客户端调用

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function main() {
  const transport = new StdioClientTransport({
    command: "npx",
    args: ["ts-node", "mcp_server.ts"],
  });

  const client = new Client({
    name: "mcp-client-demo",
    version: "1.0.0",
  }, {
    capabilities: {
      tools: {},  // 声明需要工具调用能力
    },
  });

  await client.connect(transport);
  console.log("✅ 已连接到MCP Server");

  // 发现工具
  const toolsResponse = await client.request(
    { method: "tools/list" },
    { method: "tools/list", params: {} }
  );
  console.log(`\n可用工具：`, toolsResponse.tools);

  // 调用工具
  const result = await client.request(
    { method: "tools/call" },
    {
      method: "tools/call",
      params: {
        name: "get_weather",
        arguments: { city: "北京", unit: "celsius" },
      },
    }
  );
  console.log("\n查询结果：", result.content);

  await client.close();
}

main().catch(console.error);

6.3 HTTP+SSE远程调用

对于需要远程部署的场景，使用HTTP + SSE传输：

# Server端（HTTP模式）
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("远程服务")

@mcp.tool()
def heavy_compute(data: list) -> dict:
    """耗时的计算任务，适合远程部署"""
    # 实际计算逻辑
    return {"result": sum(data), "count": len(data)}

# 使用ASGI服务器（如uvicorn）托管
# 注意：HTTP模式下需要自行处理认证和TLS

# 客户端（HTTP模式）
from mcp.client import ClientSession
from mcp.client.http import http_client

async def remote_call():
    async with http_client("https://your-mcp-server.com/mcp") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("heavy_compute", {"data": [1, 2, 3]})
            print(result)

七、与LangChain/LlamaIndex集成

7.1 LangChain集成

LangChain从0.3版本开始原生支持MCP，通过langchain-mcp包即可接入：

pip install langchain-mcp

"""
LangChain + MCP 集成示例
使用MCP工具增强LangChain Agent的能力
"""

import asyncio
from langchain_mcp import MCPtoolkit
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

async def main():
    # 方式一：通过MCPtoolkit自动发现并注册工具
    mcp_tools = await MCPtoolkit(
        command="python",
        args=["mcp_server.py"]
    ).get_tools()
    
    print(f"✅ 自动发现 {len(mcp_tools)} 个MCP工具")
    for tool in mcp_tools:
        print(f"  - {tool.name}")
    
    # 方式二：手动指定工具
    from langchain_mcp import MCPTool
    specific_tool = MCPTool(
        command="python",
        args=["mcp_server.py"],
        tool_name="get_weather"
    )
    
    # 构建Agent
    llm = ChatOpenAI(model="gpt-4")
    
    agent = create_react_agent(
        llm,
        tools=mcp_tools,  # MCP工具直接作为LangChain工具使用
        state_modifier="""
        你是一个有用的AI助手，可以调用各种工具来完成任务。
        当需要天气信息、文件操作或数据查询时，直接调用相应工具。
        """
    )
    
    # 运行Agent
    result = await agent.ainvoke({
        "messages": [
            ("user", "帮我查一下北京和上海的天气，然后告诉我哪里更适合户外运动")
        ]
    })
    
    for message in result["messages"]:
        print(f"[{message.type}]: {message.content[:200] if hasattr(message, 'content') else str(message)}")

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

7.2 LlamaIndex集成

"""
LlamaIndex + MCP 集成示例
使用MCP工具作为查询引擎的数据源
"""

import asyncio
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
from langchain_mcp import MCPtoolkit

async def setup_llamaindex_with_mcp():
    # 获取MCP工具
    mcp_toolkit = await MCPtoolkit(
        command="python",
        args=["mcp_server.py"]
    )
    mcp_tools = await mcp_toolkit.get_tools()
    
    # 转换为LlamaIndex的FunctionTool格式
    llamaindex_tools = [
        FunctionTool.from_defaults(
            fn=tool.fn,
            name=tool.name,
            description=tool.description
        )
        for tool in mcp_tools
    ]
    
    # 创建Agent
    llm = OpenAI(model="gpt-4")
    agent = ReActAgent.from_tools(llamaindex_tools, llm=llm, verbose=True)
    
    # 查询
    response = agent.chat("查询成都今天的天气并列出湿度")
    print(response)

asyncio.run(setup_llamaindex_with_mcp())

7.3 两者对比

维度	LangChain	LlamaIndex
MCP集成方式	`langchain-mcp` 包	手动转换 FunctionTool
Agent类型	ReAct / OpenAI Functions	ReAct /上下文检索
学习曲线	中等，文档丰富	较低，API简洁
适用场景	通用Agent构建	知识检索增强

八、安全考虑与权限控制

8.1 常见安全威胁

MCP Server以主进程的权限运行，一旦被攻破，攻击者可以：

┌──────────────────────────────────────────────┐
│              MCP Server 安全威胁模型           │
├──────────────────────────────────────────────┤
│ 威胁1：路径遍历                                │
│  输入:  "../../../etc/passwd"                 │
│  后果:  读取服务器敏感文件                      │
├──────────────────────────────────────────────┤
│ 威胁2：命令注入                                │
│  输入:  "; rm -rf /" (如果工具执行shell)        │
│  后果:  服务器数据被删除                        │
├──────────────────────────────────────────────┤
│ 威胁3：资源耗尽（DoS）                         │
│  输入:  读取超大文件 / 无限循环查询              │
│  后果:  Server无响应                           │
├──────────────────────────────────────────────┤
│ 威胁4：数据泄露                                │
│  工具返回包含敏感信息（API密钥、用户数据）        │
│  后果:  隐私数据暴露给AI                        │
├──────────────────────────────────────────────┤
│ 威胁5：权限提升                                │
│  Server被植入恶意代码，通过stdin/stdout外传数据  │
│  后果:  横向渗透                               │
└──────────────────────────────────────────────┘

8.2 防御策略

① 输入验证与净化

from pathlib import Path
import re

def safe_path_resolve(user_path: str, allowed_base: str) -> Path:
    """
    安全路径解析：防止路径遍历攻击
    """
    allowed = Path(allowed_base).resolve()
    try:
        resolved = (allowed / user_path).resolve()
        # 确保解析后的路径在允许范围内
        if not str(resolved).startswith(str(allowed)):
            raise ValueError(f"路径 '{user_path}' 超出允许范围")
        return resolved
    except (ValueError, OSError) as e:
        raise ValueError(f"无效路径: {user_path}")

# 使用示例
@app.tool()
def safe_read_file(path: str) -> str:
    base_dir = "/app/user_workspace"  # 允许的根目录
    safe_path = safe_path_resolve(path, base_dir)
    return open(safe_path).read()

② 资源限制

import resource
import signal

class ResourceGuard:
    """限制工具执行的资源使用"""
    
    @staticmethod
    def set_limits(max_memory_mb=100, timeout_sec=30):
        # 内存限制
        resource.setrlimit(resource.RLIMIT_AS, (max_memory_mb * 1024 * 1024, -1))
        # CPU时间限制
        resource.setrlimit(resource.RLIMIT_CPU, (timeout_sec, timeout_sec + 5))
        # 文件大小限制
        resource.setrlimit(resource.RLIMIT_FSIZE, (10 * 1024 * 1024, -1))
    
    @staticmethod
    def with_timeout(timeout_sec):
        """装饰器：超时自动终止"""
        def decorator(func):
            def wrapper(*args, **kwargs):
                def timeout_handler(signum, frame):
                    raise TimeoutError(f"执行超时（{timeout_sec}秒）")
                
                old_handler = signal.signal(signal.SIGALRM, timeout_handler)
                signal.alarm(timeout_sec)
                try:
                    return func(*args, **kwargs)
                finally:
                    signal.alarm(0)
                    signal.signal(signal.SIGALRM, old_handler)
            return wrapper
        return decorator

③ 权限级别设计

from enum import Enum
from typing import Set

class PermissionLevel(Enum):
    READ_ONLY = "read_only"      # 仅读取，不能写入
    READ_WRITE = "read_write"    # 可读写文件
    NETWORK = "network"          # 可发起网络请求
    ADMIN = "admin"              # 管理员权限

class SecureTool:
    """带权限控制的工具基类"""
    
    def __init__(self, required_permissions: Set[PermissionLevel]):
        self.required_permissions = required_permissions
    
    def check_permission(self, user_level: PermissionLevel) -> bool:
        hierarchy = {
            PermissionLevel.READ_ONLY: 0,
            PermissionLevel.READ_WRITE: 1,
            PermissionLevel.NETWORK: 2,
            PermissionLevel.ADMIN: 3,
        }
        return hierarchy[user_level] >= hierarchy[min(self.required_permissions)]

# 使用示例
class FileWriteTool(SecureTool):
    def __init__(self):
        super().__init__({PermissionLevel.READ_WRITE})
    
    def execute(self, path: str, content: str, user_level: PermissionLevel):
        if not self.check_permission(user_level):
            raise PermissionError("权限不足：需要READ_WRITE级别")
        # 执行写操作
        pass

④ 敏感数据过滤

import re

class SensitiveDataFilter:
    """自动过滤工具返回结果中的敏感信息"""
    
    PATTERNS = {
        "api_key": r'[a-zA-Z0-9]{20,64}',
        "email": r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
        "phone": r'1[3-9]\d{9}',
        "id_card": r'\d{17}[\dXx]',
        "password": r'(password|pwd|passwd)\s*[=:]\s*\S+',
    }
    
    REPLACEMENTS = {
        "api_key": "[API_KEY_REDACTED]",
        "email": "[EMAIL_REDACTED]",
        "phone": "[PHONE_REDACTED]",
        "id_card": "[ID_REDACTED]",
        "password": r'\1= [REDACTED]',
    }
    
    @classmethod
    def filter(cls, text: str, level: str = "medium") -> str:
        result = text
        if level == "high":
            # 高敏感模式：过滤所有匹配项
            for key, pattern in cls.PATTERNS.items():
                result = re.sub(pattern, cls.REPLACEMENTS.get(key, "[REDACTED]"), result)
        elif level == "medium":
            # 中敏感模式：只过滤明文密码和API Key
            result = re.sub(r'password\s*[=:]\s*\S+', 'password= [REDACTED]', result)
            result = re.sub(r'api[_-]?key\s*[=:]\s*[a-zA-Z0-9]{20,}', 'api_key= [REDACTED]', result, flags=re.IGNORECASE)
        return result

8.3 MCP官方的安全建议

✅ 始终在隔离环境中运行MCP Server
✅ 实施最小权限原则——每个Server只暴露必要工具
✅ 对所有输入进行严格验证和清理
✅ 设置资源限制（超时、内存、文件大小）
✅ 敏感数据返回前必须过滤
✅ 定期审查和审计工具定义
❌ 禁止在Server中存储用户认证凭据
❌ 禁止使用shell执行用户输入的内容
❌ 禁止将Server暴露在不可信的网络中（除非使用TLS）

九、性能优化与扩展

9.1 连接池与复用

频繁创建销毁MCP连接会产生巨大开销：

# ❌ 低效：每次调用都创建新连接
async def inefficient_call():
    for i in range(100):
        async with stdio_client(command="python", args=["server.py"]) as (r, w):
            async with ClientSession(r, w) as session:
                await session.initialize()
                await session.call_tool("get_weather", {"city": "北京"})

# ✅ 高效：连接池复用
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client
from contextlib import asynccontextmanager

class MCPConnectionPool:
    """MCP连接池：复用已建立的连接"""
    
    def __init__(self, command: str, args: list, max_size: int = 5):
        self.command = command
        self.args = args
        self.max_size = max_size
        self._pool: asyncio.Queue = asyncio.Queue(maxsize=max_size)
        self._sessions: list[ClientSession] = []
    
    async def initialize(self):
        """预热：建立多个连接备用"""
        for _ in range(self.max_size):
            read, write = await self._connect()
            session = ClientSession(read, write)
            await session.initialize()
            self._sessions.append(session)
            await self._pool.put(session)
    
    async def _connect(self):
        return await stdio_client(
            command=self.command,
            args=self.args
        ).__aenter__()
    
    async def acquire(self) -> ClientSession:
        """获取一个可用连接（阻塞直到可用）"""
        return await self._pool.get()
    
    async def release(self, session: ClientSession):
        """归还连接"""
        await self._pool.put(session)
    
    async def close(self):
        for session in self._sessions:
            await session.close()

9.2 工具调用缓存

import hashlib
import json
from functools import wraps
from typing import Optional
import asyncio

class ToolCache:
    """工具结果缓存：避免重复计算"""
    
    def __init__(self, ttl_seconds: int = 300):
        self.cache: dict[str, tuple[str, float]] = {}  # key -> (result, timestamp)
        self.ttl = ttl_seconds
        self.lock = asyncio.Lock()
    
    def _make_key(self, tool_name: str, arguments: dict) -> str:
        """生成缓存键"""
        content = json.dumps({"tool": tool_name, "args": arguments}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:16]
    
    async def get_or_call(self, tool_name: str, arguments: dict, callable_fn):
        key = self._make_key(tool_name, arguments)
        
        async with self.lock:
            if key in self.cache:
                result, timestamp = self.cache[key]
                if asyncio.get_event_loop().time() - timestamp < self.ttl:
                    return result
        
        # 执行实际调用
        result = await callable_fn()
        
        async with self.lock:
            self.cache[key] = (result, asyncio.get_event_loop().time())
        
        return result
    
    def clear(self):
        self.cache.clear()

# 使用示例
cache = ToolCache(ttl_seconds=300)

@app.tool()
async def cached_weather(city: str) -> str:
    return await cache.get_or_call(
        "get_weather",
        {"city": city},
        lambda: real_weather_api_call(city)
    )

9.3 异步批量处理

import asyncio
from typing import List

async def batch_call_tools(
    session: ClientSession,
    calls: List[dict]
) -> List[CallToolResult]:
    """
    批量并行调用多个工具
    大幅提升需要同时使用多个工具的场景效率
    """
    tasks = [
        session.call_tool(call["name"], call.get("arguments", {}))
        for call in calls
    ]
    return await asyncio.gather(*tasks, return_exceptions=True)

# 使用示例
async def research_task(session: ClientSession):
    """并行查询多个数据源"""
    results = await batch_call_tools(session, [
        {"name": "get_weather", "arguments": {"city": "北京"}},
        {"name": "get_weather", "arguments": {"city": "上海"}},
        {"name": "read_file", "arguments": {"path": "market_report.txt"}},
        {"name": "query_database", "arguments": {"query": "SELECT * FROM trends LIMIT 10"}},
    ])
    
    for result in results:
        if isinstance(result, Exception):
            print(f"❌ 调用失败: {result}")
        else:
            print(f"✅ 结果: {result.content[0].text[:100]}")

9.4 水平扩展架构

                    ┌─────────────┐
                    │   Load      │
                    │   Balancer  │
                    │  (Nginx)    │
                    └──────┬──────┘
                           │
         ┌─────────────────┼─────────────────┐
         │                 │                 │
    ┌────▼────┐      ┌────▼────┐       ┌────▼────┐
    │ MCP     │      │ MCP     │       │ MCP     │
    │ Server 1│      │ Server 2│       │ Server 3│
    │(Node.js)│      │(Python) │       │(Python) │
    └────┬────┘      └────┬────┘       └────┬────┘
         │                 │                 │
    ┌────▼────┐      ┌────▼────┐       ┌────▼────┐
    │ DB/FS   │      │ DB/FS   │       │ DB/FS   │
    └─────────┘      └─────────┘       └─────────┘
    
    说明：HTTP+SSE模式下可部署多个Server副本，
    通过Nginx做负载均衡，实现高可用和水平扩展

9.5 监控与指标

import time
from typing import Callable
import logging

logger = logging.getLogger(__name__)

def instrument_tool(func: Callable):
    """工具性能监控装饰器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            result = await func(*args, **kwargs)
            duration = (time.perf_counter() - start) * 1000
            logger.info(f"工具 {func.__name__} 执行成功，耗时 {duration:.2f}ms")
            return result
        except Exception as e:
            duration = (time.perf_counter() - start) * 1000
            logger.error(f"工具 {func.__name__} 执行失败，耗时 {duration:.2f}ms，错误: {e}")
            raise
    
    # 记录调用次数统计
    wrapper.call_count = 0
    wrapper.total_duration = 0.0
    return wrapper

# 暴露Prometheus指标端点
# /metrics 端点返回格式化的指标数据
# 可接入Grafana进行可视化监控

十、总结与资源

10.1 核心要点回顾

┌────────────────────────────────────────────────────────────┐
│                    MCP开发核心要点                           │
├────────────────────────────────────────────────────────────┤
│  协议层   │ JSON-RPC 2.0 + stdio/HTTP+SSE 传输              │
│  架构     │ 客户端-服务器模式，关注点分离，无状态设计          │
│  三大能力  │ Tools（调用）/ Resources（读取）/ Prompts（模板） │
│  开发语言  │ Python SDK + TypeScript SDK                    │
│  集成生态  │ LangChain / LlamaIndex / Claude Desktop        │
│  安全红线  │ 输入验证 + 资源限制 + 敏感数据过滤               │
│  性能优化  │ 连接池 + 结果缓存 + 批量并行调用                 │
└────────────────────────────────────────────────────────────┘

10.2 适用场景

场景	推荐方案
Claude Desktop / Cursor 插件	stdio模式，本地Python/TS Server
自研AI应用（单实例）	stdio模式，直接进程调用
企业内部多团队共享工具	HTTP+SSE模式，部署到K8s
需要RAG增强的知识库	MCP + LlamaIndex
复杂多步骤Agent	MCP + LangChain

10.3 官方资源

资源	链接
MCP官方文档	https://modelcontextprotocol.io
MCP Python SDK	`pip install mcp`
MCP TypeScript SDK	`npm install @modelcontextprotocol/sdk`
官方Server示例	https://github.com/modelcontextprotocol/servers
MCP规范（1.0）	https://spec.modelcontextprotocol.io

10.4 社区生态

MCP的生态正在快速增长，以下是截至目前的主流Server：

📦 官方维护
├── filesystem        文件系统操作
├── git              Git操作
├── slack            Slack集成
└── github           GitHub API

🌐 社区贡献
├── brave-search     Brave搜索
├── aws-kb-retrieval AWS知识库
├── postgres         PostgreSQL数据库
├── sequential-thinking  顺序思考推理
└── puppeteer        浏览器自动化

结语

MCP的出现标志着AI Agent开发进入了一个新阶段。它不仅仅是一个"工具调用协议"，更是一种生态级的互操作标准。就像USB让设备即插即用一样，MCP正在让AI工具生态走向即插即用。

本文从协议原理出发，深入剖析了无状态设计、工具注册、客户端调用、与主流框架集成、安全防护和性能优化六大核心主题，提供了完整的Python + TypeScript双语言代码示例。

现在，就从搭建你的第一个MCP Server开始吧！

💡 后续学习建议：

尝试为你的现有工具编写MCP Server

深入阅读MCP 1.0规范文档

探索社区Server生态，贡献自己的Server

研究MCP在多Agent协作中的应用场景

如果本文对你有帮助，欢迎在评论区交流讨论！

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

Agent应用开发工程师，AI落地层核心岗！掌握这7大模块

AI Agent技术社区

AI Agent如何实现长期记忆？

在人工智能领域，尤其是大型语言模型（LLM）驱动的智能体（Agent）应用中，长期记忆（Long-term Memory）是实现持续、个性化交互的关键能力。与仅能处理单次对话的短期记忆不同，长期记忆允许Agent跨越多个会话记住用户偏好、历史对话、任务上下文和个人知识，从而提供更连贯、更智能的服务。长期记忆的实现与检索增强生成（RAG）技术密不可分，可以看作是一种针对“用户/会话历史”的特化RAG

AI Agent技术社区

从Hello World到生产落地：AI Agent工程师的实战进阶秘籍

AI Agent技术社区

所有评论(0)

查看更多评论

小二·

@weixin_52208686

已为社区贡献57条内容

MCP协议开发实战：从零搭建AI Agent工具链，终于有人讲清楚了！

小二·

前言

一、MCP协议背景：7月28日，正式启航

1.1 为什么需要MCP？

1.2 官方里程碑

1.3 MCP vs 既有方案对比

二、MCP核心概念：Model Context Protocol详解

2.1 协议架构一览

2.2 三大核心能力

① Tools（工具）— AI主动调用

② Resources（资源）— AI按需读取

③ Prompts（提示模板）— AI可复用的提示词

2.3 传输层：两种通信方式

2.4 消息协议：JSON-RPC 2.0

三、无状态架构设计原理

3.1 为什么MCP选择无状态设计？

3.2 有状态需求如何满足？

3.3 协议握手流程

四、实战：搭建MCP Server（完整代码）

4.1 环境准备

4.2 Python版：完整MCP Server实现

4.3 TypeScript版：完整MCP Server实现

4.4 在Claude Desktop中配置使用

五、工具定义与注册：深入inputSchema

5.1 inputSchema的艺术

5.2 嵌套对象与数组Schema

六、客户端调用实现

6.1 Python客户端调用

6.2 TypeScript客户端调用

6.3 HTTP+SSE远程调用

七、与LangChain/LlamaIndex集成

7.1 LangChain集成

7.2 LlamaIndex集成

7.3 两者对比

八、安全考虑与权限控制

8.1 常见安全威胁

8.2 防御策略

① 输入验证与净化

② 资源限制

③ 权限级别设计

④ 敏感数据过滤

8.3 MCP官方的安全建议

九、性能优化与扩展

9.1 连接池与复用

9.2 工具调用缓存

9.3 异步批量处理

9.4 水平扩展架构

9.5 监控与指标

十、总结与资源

10.1 核心要点回顾

10.2 适用场景

10.3 官方资源

10.4 社区生态

结语

所有评论(0)

温馨提示：您尚未绑定手机号

小二·