作者：京东物流 张华
给自己几分钟时间，快速了解一下 mcp 技术~
0 1 分钟时间入门
👉 Python 实现一个基于 sse 通信的完整 mcp-demo
0.1 环境依赖
Python >= 3.10
Package Version----------------- ---------annotated-types 0.7.0anyio 4.9.0certifi 2025.4.26click 8.2.1h11 0.16.0httpcore 1.0.9httpx 0.28.1httpx-sse 0.4.0idna 3.10mcp 1.9.2pydantic 2.11.5pydantic-core 2.33.2pydantic-settings 2.9.1python-dotenv 1.1.0python-multipart 0.0.20sniffio 1.3.1sse-starlette 2.3.6starlette 0.47.0typing-extensions 4.14.0typing-inspection 0.4.1uvicorn 0.34.3
复制代码
0.2 实现 mcp-server
server-1-带 2 个 tool
#!/usr/bin/env python#-*- coding:UTF-8 -*-########################################################################## File Name: sse_server_1.py# Author: AI_Team# Mail: example@jd.com# Created Time: 17:19:01 2025-06-04#########################################################################import sysimport httpximport asynciofrom mcp.server import FastMCP
app = FastMCP('web-search', port=9000)
@app.tool()async def web_search(query: str) -> str: """ 搜索互联网内容 Args: query: 要搜索内容 Returns: 搜索结果的总结 """ res_data = ["query", query] return ':'.join(res_data)
@app.tool()async def hello_world() -> str: """ 给世界打招呼 Args: None
Returns: 打招呼的内容 """ return "Hi mcp world!"
if __name__ == "__main__": app.run(transport='sse')
复制代码
server-2-带 1 个 tool
#!/usr/bin/env python#-*- coding:UTF-8 -*-########################################################################## File Name: sse_server_2.py# Author: AI_Team# Mail: example@jd.com# Created Time: 17:19:01 2025-06-04#########################################################################import sysimport httpximport asynciofrom mcp.server import FastMCP
app = FastMCP('test-sse', port=9001)
@app.tool()async def hi_world() -> str: """ 给世界打招呼的另一种说法 Args: None
Returns: 带中文的打招呼内容 """ return "你好 mcp world!"
if __name__ == "__main__": app.run(transport='sse')
复制代码
0.3 实现 mcp-client
#!/usr/bin/env python#-*- coding:UTF-8 -*-########################################################################## File Name: sse_cli.py# Author: AI_Team# Mail: example@jd.com# Created Time: 18:43:37 2025-06-04#########################################################################import sysimport asynciofrom contextlib import AsyncExitStackfrom mcp.client.sse import sse_clientfrom mcp import ClientSession
class MCPClient: def __init__(self, server_urls: list[str]): """ 初始化 MCP 客户端。 :param server_urls: SSE 服务地址列表，用于连接多个服务器。 """ self.server_urls = server_urls self.sessions = {} # 存储每个服务器的会话及其上下文 ：server_id -> session self.tool_mapping = {} # 工具映射 ：prefixed_name -> (session, original_tool_name) self.exit_stack = AsyncExitStack()
async def initialize_sessions(self): """ 初始化与所有 SSE 服务器的连接，并获取可用工具列表。 """ for i, server_url in enumerate(self.server_urls): server_id = f"server{i}" # 为每个服务器生成唯一标识符 # 创建 SSE 客户端并进入上下文 streams_context = sse_client(url=server_url) streams = await self.exit_stack.enter_async_context(streams_context) session = await self.exit_stack.enter_async_context(ClientSession(*streams))
await session.initialize() # 存储会话 self.sessions[server_id] = session # 获取工具列表并建立映射 response = await session.list_tools() print(response) for tool in response.tools: prefixed_name = f"{server_id}_{tool.name}" # 为工具名添加服务器前缀 self.tool_mapping[prefixed_name] = (session, tool.name) print(f"已连接到 {server_url}，工具列表：{[tool.name for tool in response.tools]}")
async def chat_loop(self): for i in range(5): try: for prefixed_name, (session, tool_name) in self.tool_mapping.items(): if tool_name == "web_search": res = await session.call_tool('web_search', {'query': '杭州今天天气'}) elif tool_name == "hello_world": res = await session.call_tool('hello_world') elif tool_name == "hi_world": res = await session.call_tool('hi_world') else: res = "tool不存在！" print (tool_name + ": ") print (res) except Exception as e: import traceback traceback.print_exc()
async def cleanup(self): """ 清理所有会话和连接资源，确保无资源泄漏。 """ await self.exit_stack.aclose() print("所有会话已清理。")
async def main(): # 定义 SSE 服务器地址列表 server_urls = ["http://localhost:9000/sse", "http://localhost:9001/sse"] # 创建并运行客户端 client = MCPClient(server_urls=server_urls) try: await client.initialize_sessions() await client.chat_loop() finally: await client.cleanup()
if __name__ == '__main__': asyncio.run(main())
复制代码
0.4 执行一下吧😁
# 启动server-1python sse_server_1.py
# 启动server-2python sse_server_2.py
# cliend端执行一下python sse_cli.py
复制代码
1 如果你还有 1 分钟时间
👉 mcp 是什么
1.1 先看官方的定义
MCP 简介
MCP 是一个开放协议，它为应用程序向 LLM 提供上下文的方式进行了标准化。你可以将 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 为设备连接各种外设和配件提供了标准化的方式一样，MCP 为 AI 模型连接各种数据源和工具提供了标准化的接口。
为什么选择 MCP？
MCP 帮助你在 LLM 的基础上构建代理（agents）和复杂的工作流。LLM 经常需要与数据和工具集成，而 MCP 提供了：
•持续增长的预构建集成列表，LLM 可直接使用
•灵活切换不同的 LLM 提供商和厂商
•在你的基础设施内安全地处理数据的最佳实践
通用架构
MCP 核心采用客户端-服务器架构，主机应用可以连接多个服务器：
•MCP Hosts: 如 Claude Desktop、IDE 或 AI 工具，希望通过 MCP 访问数据的程序
•MCP Clients: 维护与服务器一对一连接的协议客户端
•MCP Servers: 轻量级程序，通过标准的 Model Context Protocol 提供特定能力
•本地数据源: MCP 服务器可安全访问的计算机文件、数据库和服务
•远程服务: MCP 服务器可连接的互联网上的外部系统（如通过 APIs）
MCP 核心概念
MCP 服务器可以提供三种主要类型的能力：
1.Resources: 可以被 clients 读取的类文件数据（如 API 响应或文件内容）
2.Tools: 可以被 LLM 调用的函数（需要用户批准）
3.Prompts: 预先编写的模板，帮助用户完成特定任务
1.2 尝试解读
是一个桥接层
•支持本机进程通信和远程网络通信两种模式
•把大模型和应用依赖的资源间的联系规范化
2 如果你暂时没有着急的事情
👉 mcp 的由来
2.1 再看 mcp 支持的三种资源
恰好适配大模型的三种应用方式，一切都是为了方便 LLM 主导的技术时代
2.2 未来展望
mcp 的背后逻辑：大家只对能力开源，且大家均遵守君子协定的假设
这种假设过于强，会存在很多隐患
参考文档
• https://mcp-docs.cn/introduction
评论