使用说明
MCP_实施
在这个仓库中,我们将重点介绍MCP客户端和服务器的实现。
入门指南
模型上下文协议(MCP)代表了人工智能开发的重大进步,为人工智能模型在交互过程中如何解释和响应上下文提供了标准化的指导方针。
这些资源共同解释了MCP在不同平台上的技术实现及其通过更好的上下文处理提高人工智能可靠性和安全性的实际应用。
人工智能代理框架的MCP集成
8个主要AI代理框架的完整模型上下文协议(MCP)客户端实现,同时支持标准MCP和FastMCP。
建筑
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ AI Framework │ │ MCP Client │ │ MCP Server │
│ (AutoGen, │◄──►│ (Adapter) │◄──►│ (Tools) │
│ LlamaIndex, │ │ │ │ │
│ etc.) │ │ │ │ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
每个客户端都将MCP工具转换为特定于框架的格式:
- MCP服务器通过stdio协议提供工具
- 客户端将MCP工具封装在框架本机抽象中
- 代理使用转换后的工具执行任务
本地与远程MCP服务器
本地MCP服务器(如我们的实现)在 同一台机器,通过stdio传输进行通信,这提供了简单性, 通过进程隔离实现安全性,同时实现零网络配置 非常适合开发、测试和单机部署。
远程MCP 服务器通过HTTP/HTTPS或WebSocket等网络协议运行 分布式架构,其中多个客户端可以访问集中式工具, 共享资源和基于云的服务,但需要额外的复杂性 用于身份验证、网络安全、负载平衡和错误处理- 使其适用于生产环境、多用户系统,以及 需要跨团队共享或集成工具的企业部署 使用外部API和数据库。
它们之间的区别可以概括如下:
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": {}
}
}
文件
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!
框架集成
安装框架依赖项
# AutoGen
pip install autogen-agentchat
# LlamaIndex with Claude
pip install llama-index llama-index-llms-anthropic
# LangChain with Claude
pip install langchain langchain-anthropic
# AWS Bedrock
pip install langchain-aws boto3
# CrewAI
pip install crewai
# LangGraph (included with LangChain)
pip install langgraph
# LLM providers
pip install anthropic
export ANTHROPIC_API_KEY="your-api-key"
# For Bedrock (configure AWS credentials)
aws configure
用法示例
标准MCP客户端
from working_mcp_client import WorkingMCPClient
import asyncio
async def main():
client = WorkingMCPClient()
await client.connect_and_demo()
asyncio.run(main())
AutoGen客户端
from autogen_mcp_client import AutoGenMCPClient
import asyncio
async def main():
client = AutoGenMCPClient()
await client.connect_and_test()
asyncio.run(main())
LlamaIdex客户端
from llamaindex_mcp_client import LlamaIndexMCPClient
import asyncio
async def main():
client = LlamaIndexMCPClient()
await client.connect_and_test()
asyncio.run(main())
Strands代理客户
from strands_mcp_client import StrandsMCPClient
import asyncio
async def main():
client = StrandsMCPClient(["python", "mcp_server.py"])
await client.connect()
response = await client.chat("Search for AI research documents")
asyncio.run(main())
CrewAI客户端
from crewai_mcp_client import CrewAIMCPClient
import asyncio
async def main():
client = CrewAIMCPClient()
await client.connect_and_test()
asyncio.run(main())
朗链客户端
from langchain_mcp_client import LangChainMCPClient
import asyncio
async def main():
client = LangChainMCPClient(["python", "mcp_server.py"])
await client.connect()
asyncio.run(main())
LangGraph客户端
from langgraph_mcp_client import LangGraphMCPClient
import asyncio
async def main():
client = LangGraphMCPClient()
await client.connect_and_test()
asyncio.run(main())
AWS基岩客户端
from bedrock_mcp_client import BedrockMCPClient
import asyncio
async def main():
client = BedrockMCPClient(["python", "mcp_server.py"], region="us-east-1")
await client.connect()
asyncio.run(main())
配置
克劳德(人类学):
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}'"
代理框架比较
| 框架 | 类型 | 关键优势 | 最佳用例 | 学习曲线 | 社区 | 法学硕士支持 |
|---|---|---|---|---|---|---|
| 自动生成 | 多代理 | 对话驱动的代理、基于角色的交互 | 复杂的多代理工作流、协作解决问题 | 中型 | 大型(微软) | OpenAI、Azure、自定义 |
| Llama索引 | RAG聚焦 | 文档索引、检索增强、查询引擎 | 知识密集型应用程序、文档问答、RAG系统 | 中型 | 大型 | OpenAI、Anthropic、本地模型 |
| 缕 | 结构化 | 类型安全代理、结构化工作流、企业就绪 | 生产应用程序、结构化数据处理 | 低 | 增长(AWS) | Claude、Bedrock、自定义 |
| 船员AI | 基于团队 | 基于角色的代理、分层任务委托、团队协调 | 业务流程自动化、团队模拟 | 中等 | 中等 | OpenAI、Anthropic、本地模型 |
| LangChain | 通用 | 广泛的生态系统、链、内存、工具集成 | 快速原型制作、多样化的人工智能应用、工具编排 | 高 | 非常大 | 所有主要供应商 |
| LangGraph | 工作流 | 状态机、条件流、复杂路由 | 复杂决策树、有状态工作流、过程自动化 | 高级 | 中级(LangChain) | 支持所有LangChain |
详细比较
自动生成
- 优点:自然的对话流程,轻松的多代理设置,微软支持
- 缺点:仅限于对话模式,需要仔细的即时工程
- 最适合:客户服务机器人、协作式人工智能助理、教育工具
Llama索引
- 优点:出色的RAG功能、文档处理、查询优化
- 缺点:主要以文件为中心,对总代理任务的灵活性较低
- 最适合:知识库、文件分析、研究助理
缕
- 优点:类型安全、AWS集成、企业功能、结构化方法
- 缺点:生态系统更新,社区资源减少
- 最适合:生产系统、AWS环境、企业应用程序
船员AI
- 优点:直观的团队隐喻、基于角色的设计、任务委派
- 缺点:相对较新,生态系统较小,高级功能有限
- 最适合:业务工作流程、团队模拟、分层任务处理
LangChain
- 优点:庞大的生态系统、广泛的集成、灵活的架构
- 缺点:API复杂,变化频繁,学习曲线陡峭
- 最适合:复杂的人工智能应用程序、研究项目、定制集成
LangGraph
- 优点:强大的状态管理、可视化工作流、条件逻辑
- 缺点:复杂的设置,需要LangChain知识,更新的框架
- 最适合:复杂的决策过程、有状态的应用程序、工作流自动化
框架选择指南
| 用例 | 推荐框架 | 备选方案 |
|---|---|---|
| 文件问答 | LlamaIdex | 朗链 |
| 多代理聊天 | AutoGen | CrewAI |
| 业务流程 | CrewAI | Strands |
| 企业生产 | Strands | LangChain |
| 研究/原型制作 | LangChain | AutoGen |
| 复杂的工作流程 | LangGraph | LangChain |
| AWS集成 | 绞线 | LangChain+基岩 |
| 快速发展 | AutoGen | CrewAI |
框架集成详细信息
| 框架 | 集成类型 | 关键功能 |
|---|---|---|
| 基本MCP | 直接MCP调用 | 纯MCP功能-直接工具调用 |
| 自动生成 | ConversableAgent | 多代理对话 |
| Llama索引 | BaseTool+ReActAgent | RAG和推理 |
| 缕 | 代理+工具 | 结构化代理工作流 |
| 船员AI | Agent+Task+Crew | 基于团队的任务执行 |
| LangChain | BaseTool+代理 | 函数调用代理 |
| LangGraph | 状态图 | 有状态的工作流图 |
| 基岩 | ChatBedrock | AWS原生LLM集成 |
可用的MCP工具
search_documents(query, limit)-搜索文档集合get_weather(location)-获取天气信息
测试
运行单独测试:
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_summary.py
预期结果: 8/8实施工作
依赖项
核心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工具 - 实现遵循每个框架的最佳实践和模式
何时使用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 |
|---|---|---|
| 学习曲线 | 简单 | 中等 |
| 代码量 | 最小 | 详细 |
| 灵活性 | 有限 | 完全 |
| 性能 | 良好 | 优化 |
| 生产就绪 | 基础 | 企业 |
| 定制 | 有限 | 广泛 |
经验法则:从原型的FastMCP开始,迁移到标准MCP 需要高级功能或精细控制的生产系统。
基于LLM的应用程序与MCP工具的集成
AWS研讨会
- MCP研讨会-MCP与亚马逊Q
- 使用Amazon Q CLI和Rust编程构建MCP服务器
- 使用Amazon Q CLI和Python编程构建MCP服务器
- 日常生产力加速器:使用模型上下文协议(MCP)构建GenAI工具
- 韩语-MCP研讨会
- 韩语-MCP与亚马逊Q研讨会
- 使用人工智能和MCP来原型化和管理您的直播应用程序
- 亚马逊Q开发者CLI和MCP研讨会
- 使用亚马逊智能IDE的MCP SecKiro自动化安全审查
- 亚马逊EKS的代理开发经验
- 使用AWS MCP服务器进行Vibe编码
- 用Python构建MCP服务器
- EKS研讨会上的AI代理
- 在亚马逊基岩上与Claude一起构建MCP驱动的代理
- 使用仅支持MCP的AWS托管事件开发DynamoDB数据模型
- 通过代理AIOps加速平台工程师的生产力
- AI驱动的容器操作:在AWS上使用MCP服务器构建和部署
- 在Amazon EKS上部署代理AI应用程序
- 具有Athena、DBT、Iceberg、Strands SDK和MCP的经济高效的现代数据平台
- 亚马逊基岩MCP医疗保健和生命科学实践实验室
- 使用Strands、AgentCore、MCP、Q CLI进行AWS EOL检测和补救-研讨会
- 使用Amazon Nova Act和MCP构建代理
- 使用Strands、MCP和S3表构建无服务器分析代理
- 利用AWS MCP服务器,使用Amazon Q CLI简化SAP操作
