基于 MCP 实现 Client 与 Server 通信：三种方式详解

在 MCP（Model Control Protocol）框架中，Client 与 Server 之间的通信是核心环节。本文将详细介绍 MCP Client 与 Server 之间的三种通信方式：stdio、HTTP+SSE 和 Streamable HTTP，并通过具体代码实现帮助读者快速掌握每种方式的应用。

MCP 通信方式概述

MCP Client 与 Server 之间主要支持三种通信方式，每种方式适用于不同的场景：

1. stdio 方式：通过标准输入（stdin）和标准输出（stdout）实现双向通信。特点是 Client 启动 Server 子进程，且 Server 进程只能与启动它的 MCP Client 通信，适用于本地快速集成测试场景。

2. HTTP+SSE 方式：基于 HTTP 请求 / 响应和 Server-Sent Events 实现通信。Server 作为独立进程运行，可部署在远端，支持多个客户端随时连接和断开。

3. Streamable HTTP 方式：流式 HTTP 通信方式，支持在数据未传输完成前就开始处理，是 MCP 项目 PR#206 引入的全新机制，用于替代 HTTP+SSE。

使用 stdio 通信方式

stdio 方式是本地开发测试的首选，下面我们一步步实现基于 stdio 的 MCP Client。

1. MCP Client 项目初始化

使用 UV 工具初始化项目：

# 初始化项目
uv init mcp-client-demo

# 添加MCP客户端依赖
uv add "mcp[cli]"

初始化完成后，删除默认的hello.py，创建client.py文件，项目结构如下：

mcp-client-demo/
├── .venv/
├── .gitignore
├── python_version
├── client.py
├── pyproject.toml
├── README.md
└── uv.lock

2. MCP Client 代码实现

（1）导入必要的包

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

各包功能说明：

• ClientSession：创建客户端会话对象，封装与服务器通信的逻辑（发送请求、接收响应）
• StdioServerParameters：配置 stdio 连接参数，启动本地服务器进程
• stdio_client：异步客户端工厂函数，使用 stdio 协议与服务器交互

（2）配置服务器连接参数

# 创建stdio方式的服务器参数结构
server_params = StdioServerParameters(
    command="uv",  # 可执行文件
    args=[
        "run",
        "--with",
        "mcp[cli]",
        "--with-editable",
        "D:\\workspace\\python\\mcp-test\\achievement",
        "mcp",
        "run",
        "D:\\workspace\\python\\mcp-test\\achievement\\server.py"
    ],  # 可选的命令行参数
    env=None  # 可选的环境变量
)

这段代码配置了启动 MCP Server 的命令和参数，相当于通过 UV 工具运行指定的 MCP Server，并建立 stdio 连接。

（3）实现异步主函数

async def run():
    # 建立stdio客户端连接
    async with stdio_client(server_params) as (read, write):
        # 创建客户端会话
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            
            # 列出可用的工具
            tools = await session.list_tools()
            print("Tools:", tools)
            
            # 调用工具
            score = await session.call_tool(
                name="get_score_by_name",
                arguments={"name": "张三"}
            )
            print("score: ", score)
            
            # 列出可用的资源
            resources = await session.list_resources()
            
            # 读取资源
            content, mime_type = await session.read_resource("file:///info.md")
            print("resource: ", mime_type)
            
            # 列出可用的提示词模板
            prompts = await session.list_prompts()
            
            # 获取提示词模板
            prompt = await session.get_prompt(
                "prompt", 
                arguments={"name": "张三"}
            )
            print("prompt: ", prompt)

代码解析：

• 第 2 行：使用stdio_client建立连接，返回read和write两个异步函数，分别用于从子进程读取数据和向子进程写入数据
• 第 3 行：创建ClientSession对象封装通信逻辑
• 第 5 行：初始化连接
• 第 7-31 行：演示了 MCP 的核心功能：列出工具、调用工具、操作资源和提示词模板

（4）程序入口

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

（5）运行测试

uv run client.py

使用 HTTP+SSE 通信方式

SSE（Server-Sent Events）方式允许服务器向客户端实时单向推送数据，目前已逐步被 Streamable HTTP 替代，但仍需了解其实现方式。

1. 支持 SSE 的 MCP Server 实现

MCP Python SDK 使用 Starlette 框架实现 SSE 支持

def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:
    """创建Starlette应用，使Mcp Server支持SSE通信方式"""
    # 创建SseServerTransport对象
    sse = SseServerTransport("/messages/")
    
    # 在MCP Client发起SSE连接请求时调用
    async def handle_sse(request: Request) -> None:
        async with sse.connect_sse(
            request.scope,
            request.receive,
            request.send,
        ) as (read_stream, write_stream):
            await mcp_server.run(
                read_stream,
                write_stream,
                mcp_server.create_initialization_options(),
            )
    
    return Starlette(
        debug=debug,
        routes=[
            # 当MCP Client访问该路径时，触发handle_sse函数以处理SSE连接
            Route("/sse", endpoint=handle_sse),
            # 用于处理MCP Client通过POST请求发送的消息
            Mount("/messages/", app=sse.handle_post_message),
        ],
    )

代码解析：

• 第 4 行：创建SseServerTransport对象，指定路径为/messages/，用于管理 SSE 连接和消息传输
• 第 7-17 行：定义异步请求处理函数handle_sse，建立 SSE 连接并返回读写流
• 第 19-27 行：创建 Starlette 应用，配置两条路由：/sse处理 SSE 连接，/messages/处理 POST 请求

2. 启动 SSE Server

if __name__ == "__main__":
    mcp_server = mcp._mcp_server
    parser = argparse.ArgumentParser(description='Run MCP SSE-based server')
    parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
    parser.add_argument('--port', type=int, default=18080, help='Port to listen on')
    args = parser.parse_args()
    
    # 为MCP Server创建SSE连接
    starlette_app = create_starlette_app(mcp_server, debug=True)
    uvicorn.run(starlette_app, host=args.host, port=args.port)

启动命令：

uv run server.py

3. SSE Client 实现

将 stdio 客户端改造为 SSE 客户端只需替换通信工厂函数：

async def connect_to_sse_server(server_url: str):
    """使用SSE连接MCP Server"""
    # 建立SSE客户端连接
    async with sse_client(url=server_url) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 列出可用工具
            tools = await session.list_tools()
            print("Tools:", tools)
            
            # 调用工具
            score = await session.call_tool(
                name="get_score_by_name",
                arguments={"name": "张三"}
            )
            print("score: ", score)
            
            # 列出可用资源
            resources = await session.list_resources()
            
            # 读取资源
            content, mime_type = await session.read_resource("file:///info.md")
            print("resource: ", mime_type)
            
            # 列出可用prompt模板
            prompts = await session.list_prompts()
            
            # 获取prompt模板
            prompt = await session.get_prompt(
                "prompt", 
                arguments={"name": "张三"}
            )
            print("prompt: ", prompt)

# 程序入口
async def main():
    if len(sys.argv) < 2:
        print("Usage: uv run client.py <URL of SSE MCP server (i.e, http://localhost:18080/sse)>")
        sys.exit(1)
    await connect_to_sse_server(server_url=sys.argv[1])

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

运行命令：

uv run client-sse.py http://localhost:18080/sse

使用 Streamable HTTP 通信方式

Streamable HTTP 是 MCP 推荐的通信方式，解决了 HTTP+SSE 的资源消耗、复杂性和兼容性问题。

1. Streamable HTTP Server 实现

MCP 在 FastMCP 中封装了 Streamable HTTP 服务器的实现：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("achievement")

# 省略定义工具、资源和提示词的代码
...

# 启动Streamable HTTP服务器
mcp.run(transport="streamable-http")

代码解析：

• 与 stdio 方式的代码几乎相同，唯一区别是调用run()时指定transport="streamable-http"参数
• 启动后默认监听 8000 端口

2. Streamable HTTP Client 实现

将 SSE 客户端改造为 Streamable HTTP 客户端只需替换通信工厂函数：

from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
import asyncio
import sys

async def connect_to_streamable_server(server_url: str):
    # 连接到Streamable HTTP服务器
    async with streamablehttp_client(url=server_url) as (
        read_stream,
        write_stream,
    ):
        # 以下逻辑与之前的客户端实现相同
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            # ... 工具、资源和提示词操作 ...

# 运行客户端
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: uv run client-streamable.py <URL of Streamable MCP server>")
        sys.exit(1)
    asyncio.run(connect_to_streamable_server(sys.argv[1]))

运行命令：

uv run client-streamable.py http://localhost:8000/mcp

Streamable HTTP 的优势：

• 统一端点：所有通信整合到一个端点，简化交互
• 按需流式传输：灵活选择响应方式，优化性能
• 状态管理：引入 Session 机制，支持状态管理和恢复

致谢

谢谢大家的阅读，还有很多不足支出，欢迎大家在评论区指出，如果我的内容对你有帮助，可以点赞 ， 收藏 ，大家的支持就是我坚持下去的动力！

“请赐予我平静，去接受我无法改变的 ；赐予我勇气，去改变我能改变的。”