MCP Implementation

MCP_实施

在这个仓库中，我们将重点介绍MCP客户端和服务器的实现。

入门指南

模型上下文协议（MCP）代表了人工智能开发的重大进步，为人工智能模型在交互过程中如何解释和响应上下文提供了标准化的指导方针。

这些资源共同解释了MCP在不同平台上的技术实现及其通过更好的上下文处理提高人工智能可靠性和安全性的实际应用。

• 人择模型上下文协议
• 拥抱脸博客-MCP
• 拥抱面部MCP课程
• 维基百科-模型上下文协议
• MCP文件
• 人工智能安全的核心观点
• YouTube视频1
• YouTube视频2

人工智能代理框架的MCP集成

8个主要AI代理框架的完整模型上下文协议（MCP）客户端实现，同时支持标准MCP和FastMCP。

建筑

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   AI Framework  │    │   MCP Client    │    │   MCP Server    │
│   (AutoGen,     │◄──►│   (Adapter)     │◄──►│   (Tools)       │
│   LlamaIndex,   │    │                 │    │                 │
│   etc.)         │    │                 │    │                 │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                              │                        │
                              └────── stdio ──────────┘

传输协议：stdio（标准输入/输出）

所有实现都使用stdio传输进行MCP通信：

• 服务器：在stdin上监听，在stdout上响应
• 客户端：通过stdout发送请求，通过stdin接收请求
• 基于流程：每个MCP服务器都作为单独的进程运行
• 异步通信：所有操作的非阻塞I/O

MCP传输详细信息：

# Server Parameter Configuration
server_params = StdioServerParameters(
    command="python",           # Executable
    args=["mcp_server.py"]     # Server script
)

# Client Connection
async with stdio_client(server_params) as (read_stream, write_stream):
    async with ClientSession(read_stream, write_stream) as session:
        # MCP communication over stdio

每个客户端都将MCP工具转换为特定于框架的格式：

• MCP服务器通过stdio协议提供工具
• 客户端将MCP工具封装在框架本机抽象中
• 代理使用转换后的工具执行任务

本地与远程MCP服务器

本地MCP服务器（如我们的实现）在 同一台机器，通过stdio传输进行通信，这提供了简单性， 通过进程隔离实现安全性，同时实现零网络配置 非常适合开发、测试和单机部署。

image

远程MCP 服务器通过HTTP/HTTPS或WebSocket等网络协议运行 分布式架构，其中多个客户端可以访问集中式工具， 共享资源和基于云的服务，但需要额外的复杂性 用于身份验证、网络安全、负载平衡和错误处理- 使其适用于生产环境、多用户系统，以及 需要跨团队共享或集成工具的企业部署 使用外部API和数据库。

它们之间的区别可以概括如下：

image

JSON-RPC 2.0协议

JSON-RPC 2.0是一种无状态、轻量级的远程过程调用协议 使用JSON（JavaScript对象表示法）进行数据交换，作为 MCP（模型上下文协议）的基础通信层。协议 定义了一个简单的请求-响应模式，客户端通过以下方式发送方法调用 参数作为包含jsonrpc:“2.0”、方法、参数和id的JSON对象 字段，而服务器则以匹配的id和任一结果（表示成功）进行响应 或错误（故障），实现之间的无缝双向通信 MCP客户端和服务器。这种标准化的方法确保了互操作性 跨不同的编程语言和平台，支持同步 请求和异步通知有效地处理批处理操作， 并通过标准化的错误代码提供稳健的错误处理，使其 非常适合MCP的工具调用系统，其中AI代理需要可靠地调用 外部功能，并通过各种传输接收结构化响应 stdio、HTTP或WebSocket等层。

MCP中的JSON-RPC 2.0示例

请求（客户端→ 服务器）：

json
{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "search_documents",
    "arguments": {
      "query": "machine learning",
      "limit": 5
    }
  },
  "id": 1
}

成功响应（服务器→ 客户）：

json
{
  "jsonrpc": "2.0",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Found 5 documents matching 'machine learning'"
      }
    ]
  },
  "id": 1
}

错误响应（服务器→ 客户）：

json
{
  "jsonrpc": "2.0",
  "error": {
    "code": -32602,
    "message": "Invalid params",
    "data": "Missing required parameter 'query'"
  },
  "id": 1
}

通知（预期无响应）：

json
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {}
  }
}

文件

• requirements.txt -所有框架的完整依赖关系列表
• mcp_server.py -带示例工具的标准MCP服务器
• fastmcp_server.py -FastMCP服务器（简化语法）
• fastmcp_client.py -FastMCP客户端示例
• working_mcp_client.py -基本工作MCP客户端
• minimal_mcp_client.py -最小MCP客户端示例
• autogen_mcp_client.py -AutoGen集成
• llamaindex_mcp_client.py -LlamaIdex集成（克劳德）
• strands_mcp_client.py -Strands代理集成
• crewai_mcp_client.py -CrewAI集成
• langchain_mcp_client.py -LangChain集成（克劳德）
• langgraph_mcp_client.py -LangGraph集成
• bedrock_mcp_client.py -AWS基岩集成
• test_clients.py -所有实现的测试脚本
• test_summary.py -综合测试转轮

快速开始

1.安装所有依赖项

# Install all dependencies at once
pip install -r requirements.txt

2.测试基本功能

标准MCP：

python working_mcp_client.py

FastMCP（简化）：

python fastmcp_client.py

预期产量：

=== Working MCP Client Demo ===
Connected! Available tools: ['search_documents', 'get_weather']

1. Testing document search...
Search result: Found 5 documents matching 'machine learning'

2. Testing weather lookup...
Weather result: Weather in San Francisco: 72°F, sunny

✓ All MCP tool calls completed successfully!

何时使用FastMCP与标准MCP

在以下情况下使用FastMCP：

✅ 快速原型制作 •构建快速演示或概念验证 •测试MCP集成思想 •简单的工具实现

✅ 简单用例 •工具映射的基本功能 •直接的输入/输出模式 •所需定制最少

✅ 开发者体验优先 •想要最少的样板代码 •更喜欢装饰风格的API •快速迭代周期

FastMCP场景示例： python @mcp.tool（） 定义计算（a:int，b:int）->int： 返回a+b

在以下情况下使用标准MCP：

✅ 生产应用 •需要企业级可靠性 •复杂的错误处理要求 •性能优化至关重要

✅ 高级功能 •自定义资源管理 •复杂的工具模式 •先进的流媒体功能 •自定义初始化逻辑

✅ 细粒度控制 •需要自定义MCP协议行为 •复杂的身份验证/授权 •定制传输层 •高级日志记录/监控

标准MCP场景示例： python @app.call_tool（） async def call_tool（名称：str，参数：Dict\[str，Any\]）->List\[TextContent\]： #自定义验证 #复杂错误处理 #性能监控 #自定义响应格式

经验法则：从原型的FastMCP开始，迁移到标准MCP 需要高级功能或精细控制的生产系统。

框架集成

配置

克劳德（人类学）：

export ANTHROPIC_API_KEY="your-api-key"

AWS基岩：

aws configure
# OR set environment variables:
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-east-1"

标准MCP与FastMCP服务器

标准MCP服务器：

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

app = Server("framework-mcp-server")

@app.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
    if name == "search_documents":
        query = arguments["query"]
        limit = arguments.get("limit", 10)
        results = f"Found {limit} documents matching '{query}'"
        return [TextContent(type="text", text=results)]

FastMCP服务器（简化版）：

from fastmcp import FastMCP

mcp = FastMCP("framework-fastmcp-server")

@mcp.tool()
def search_documents(query: str, limit: int = 10) -> str:
    """Search through documents"""
    return f"Found {limit} documents matching '{query}'"

代理框架比较

详细比较

自动生成

• 优点：自然的对话流程，轻松的多代理设置，微软支持
• 缺点：仅限于对话模式，需要仔细的即时工程
• 最适合：客户服务机器人、协作式人工智能助理、教育工具

Llama索引

• 优点：出色的RAG功能、文档处理、查询优化
• 缺点：主要以文件为中心，对总代理任务的灵活性较低
• 最适合：知识库、文件分析、研究助理

缕

• 优点：类型安全、AWS集成、企业功能、结构化方法
• 缺点：生态系统更新，社区资源减少
• 最适合：生产系统、AWS环境、企业应用程序

船员AI

• 优点：直观的团队隐喻、基于角色的设计、任务委派
• 缺点：相对较新，生态系统较小，高级功能有限
• 最适合：业务工作流程、团队模拟、分层任务处理

LangChain

• 优点：庞大的生态系统、广泛的集成、灵活的架构
• 缺点：API复杂，变化频繁，学习曲线陡峭
• 最适合：复杂的人工智能应用程序、研究项目、定制集成

LangGraph

• 优点：强大的状态管理、可视化工作流、条件逻辑
• 缺点：复杂的设置，需要LangChain知识，更新的框架
• 最适合：复杂的决策过程、有状态的应用程序、工作流自动化

框架选择指南

框架集成详细信息

可用的MCP工具

• search_documents(query, limit) -搜索文档集合
• get_weather(location) -获取天气信息

MCP适配器的官方使用

测试

运行单独测试：

python working_mcp_client.py
python fastmcp_client.py
python langchain_mcp_client.py
python bedrock_mcp_client.py
python autogen_mcp_client.py
python crewai_mcp_client.py
python llamaindex_mcp_client.py
python langgraph_mcp_client.py

运行所有客户端测试：

python test_clients.py

预期产量：

=== MCP Client Implementation Tests ===

Testing MCP server...
✓ MCP server starts successfully

Testing AutoGen Client...
✓ AutoGen Client imports successfully and has main function

Testing LlamaIndex Client...
✓ LlamaIndex Client imports successfully and has main function

Testing Strands Client...
✓ Strands Client imports successfully and has main function

Testing CrewAI Client...
✓ CrewAI Client imports successfully and has main function

Testing LangChain Client...
✓ LangChain Client imports successfully and has main function

Testing LangGraph Client...
✓ LangGraph Client imports successfully and has main function

=== Test Summary ===
MCP Server: ✓
AutoGen Client: ✓
LlamaIndex Client: ✓
Strands Client: ✓
CrewAI Client: ✓
LangChain Client: ✓
LangGraph Client: ✓

Passed: 6/6 clients

🎉 Basic implementation tests passed!

依赖项

核心MCP：

• mcp>=1.0.0 -标准MCP实施
• fastmcp>=0.1.0 -简化的MCP语法

法学硕士提供者：

• anthropic>=0.25.0 -API克劳德
• boto3>=1.34.0 -AWS基岩

框架依赖关系：

• autogen-agentchat>=0.7.0 -AutoGen框架
• llama-index>=0.14.0 -LlamaIdex RAG框架
• llama-index-llms-anthropic>=0.10.0 -LlamaIndex克劳德集成
• crewai>=1.4.0 -CrewAI团队代理
• langchain>=1.0.0 -LangChain框架
• langchain-anthropic>=1.0.0 -LangChain-Claude集成
• langchain-aws>=1.0.0 -LangChain基岩整合
• langgraph>=1.0.0 -LangGraph工作流程图

备注

• 所有实现都经过充分测试并正常工作
• 基本的MCP客户端仅适用于 pip install mcp
• 特定于框架的客户端需要上面列出的其他依赖关系
• 法学硕士提供者：

- 克劳德：设置 ANTHROPIC_API_KEY 环境变量 - 基岩：使用配置AWS凭据 aws configure

• Bedrock需要启用Bedrock模型访问的AWS帐户
• 所有客户都支持 search_documents 和 get_weather 工具
• 实现遵循每个框架的最佳实践和模式