给自己几分钟时间，快速了解一下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 ClientSessionclass 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 提供了：

通用架构

MCP 核心采用客户端-服务器架构，主机应用可以连接多个服务器：



MCP核心概念

MCP 服务器可以提供三种主要类型的能力：

1.2 尝试解读

是一个桥接层



2 如果你暂时没有着急的事情

👉 mcp的由来

2.1 再看mcp支持的三种资源

恰好适配大模型的三种应用方式，一切都是为了方便LLM主导的技术时代



2.2 未来展望

mcp的背后逻辑：大家只对能力开源，且大家均遵守君子协定的假设

这种假设过于强，会存在很多隐患

参考文档