1. Streamable HTTP #

本章主要介绍如何使用 Streamable HTTP 传输协议与 MCP 服务器进行交互。你将学会：

• 如何使用 streamablehttp_client 创建 HTTP 客户端；
• 如何使用 ClientSession 管理 MCP 会话；
• 如何列出服务器提供的工具，并调用工具。

2. Streamable HTTP #

在 C:\mcp-project 下新建 streamable_http_server.py：

# 从 mcp.server.fastmcp 模块导入 FastMCP 类
from mcp.server.fastmcp import FastMCP

# 创建 FastMCP 实例，名称为 "HTTP Server"
# 默认监听地址为 127.0.0.1，端口为 8000，路径为 /mcp
mcp = FastMCP(name="HTTP Server")


# 使用装饰器注册一个名为 greet 的工具，便于客户端验证
@mcp.tool()
# 定义 greet 函数，接收一个字符串参数 name，默认值为 "World"
def greet(name: str = "World") -> str:
    # 返回格式化的问候语字符串
    return f"Hello, {name}!"


# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 以 Streamable HTTP 方式运行服务器（阻塞运行，监听 127.0.0.1:8000，路径为 /mcp）
    mcp.run(transport="streamable-http")

在 C:\mcp-project 下新建 streamable_http_client.py：

# 使用 Streamable HTTP 连接服务器，列出工具并调用 greet

# 导入异步IO库，用于支持异步编程
import asyncio

# 从 mcp 包导入 ClientSession，用于管理客户端会话
from mcp import ClientSession

# 从 mcp.client.streamable_http 导入 streamablehttp_client 工厂方法，用于创建 HTTP 客户端
from mcp.client.streamable_http import (
    streamablehttp_client,
)


# 定义主异步函数
async def main() -> None:
    # 设置服务器地址，默认为本地 http://127.0.0.1:8000/mcp
    server_url = "http://127.0.0.1:8000/mcp"
    # 使用 streamablehttp_client 建立到服务器的读/写流
    async with streamablehttp_client(server_url) as (read, write, _):
        # 基于读/写流创建客户端会话
        async with ClientSession(read, write) as session:
            # 初始化会话，完成握手
            await session.initialize()

            # 列出服务器已注册的工具
            tools = await session.list_tools()
            # 打印所有工具的名称
            print("[Tools]", [t.name for t in tools.tools])

            # 调用 greet 工具，传入参数 name="MCP"
            result = await session.call_tool("greet", {"name": "MCP"})
            # 导入 types，用于类型判断
            from mcp import types

            # 用于存放解析到的文本内容
            texts = []
            # 遍历返回内容块，提取文本内容
            for block in result.content:
                if isinstance(block, types.TextContent):
                    texts.append(block.text)
            # 打印 greet 工具的调用结果
            print("[Call greet]", " | ".join(texts))


# 判断是否为主模块入口
if __name__ == "__main__":
    # 在事件循环中运行主异步函数
    asyncio.run(main())

运行（在两个命令行窗口中）：

窗口 1（启动服务器）：

cd C:\mcp-project
call .venv\Scripts\activate
python streamable_http_server.py

窗口 2（运行客户端）：

cd C:\mcp-project
call .venv\Scripts\activate
python streamable_http_client.py

预期输出（客户端）：

[Tools] ['greet']
[Call greet] Hello, MCP!

3. 在 Starlette 中挂载多个 FastMCP 服务 #

我们创建一个 Starlette 应用，将两个 FastMCP 服务（Echo/Math）分别挂载在 /echo 与 /math 路径。每个服务的 MCP 流式 HTTP 接口仍位于子路径下的 /mcp。

3.1 创建 ASGI 服务器 #

在 C:\mcp-project 下新建 asgi_server.py：

# 导入 contextlib，用于管理多个异步上下文（如 SessionManager）
import contextlib

# 从 starlette.applications 导入 Starlette 应用类
from starlette.applications import Starlette

# 从 starlette.routing 导入 Mount，用于路由挂载
from starlette.routing import Mount

# 从 mcp.server.fastmcp 导入 FastMCP 高层服务器类
from mcp.server.fastmcp import FastMCP

# 创建 Echo 服务器实例，命名为 "EchoServer"，并设置为无状态（stateless_http=True）
echo_mcp = FastMCP(name="EchoServer", stateless_http=True)


# 使用装饰器注册 echo 工具到 echo_mcp 服务器
@echo_mcp.tool()
# 定义 echo 工具函数，接收 message 字符串参数，返回回声字符串
def echo(message: str) -> str:
    """回声工具：原样返回输入。"""
    return f"Echo: {message}"


# 创建 Math 服务器实例，命名为 "MathServer"，并设置为无状态（stateless_http=True）
math_mcp = FastMCP(name="MathServer", stateless_http=True)


# 使用装饰器注册 add_two 工具到 math_mcp 服务器
@math_mcp.tool()
# 定义 add_two 工具函数，接收整数 n，返回 n+2
def add_two(n: int) -> int:
    """将输入数字加二。"""
    return n + 2


# 定义 Starlette 应用的生命周期管理器，负责启动和关闭两个 MCP 会话管理器
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    # 创建异步退出栈，确保多个上下文正确进入和退出
    async with contextlib.AsyncExitStack() as stack:
        # 启动 echo_mcp 的会话管理器
        await stack.enter_async_context(echo_mcp.session_manager.run())
        # 启动 math_mcp 的会话管理器
        await stack.enter_async_context(math_mcp.session_manager.run())
        # 生命周期 yield，供 Starlette 使用
        yield


# 创建 Starlette 应用，并将两个 MCP 服务分别挂载到不同子路径
# 每个 FastMCP 的 streamable_http_app() 会在其根路径下提供 /mcp 接口
app = Starlette(
    routes=[
        # 挂载 echo_mcp 到 /echo 路径，实际接口为 /echo/mcp
        Mount("/echo", app=echo_mcp.streamable_http_app()),
        # 挂载 math_mcp 到 /math 路径，实际接口为 /math/mcp
        Mount("/math", app=math_mcp.streamable_http_app()),
    ],
    # 指定生命周期管理器
    lifespan=lifespan,
)

3.2 运行 ASGI 服务器 #

方式 A（推荐，显式模块方式）：

python -m uvicorn asgi_server:app --host 127.0.0.1 --port 8000 --reload

方式 B（直接 uvicorn 命令，取决于环境是否生成脚本）：

uvicorn asgi_server:app --host 127.0.0.1 --port 8000 --reload

curl -X POST http://localhost:8000/echo/mcp \
  -H "Content-Type: application/json" \
 -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tools/call",
    "params": {
      "name": "echo",
      "arguments": {"message": "Hello World"}
    },
    "id": 1
  }'


curl -X POST http://localhost:8000/math/mcp \
  -H "Content-Type: application/json" \
 -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tools/call", 
    "params": {
      "name": "add_two",
      "arguments": {"n": 5}
    },
    "id": 1
  }'  

3.3 创建客户端 #

客户端示例：在 C:\mcp-project 下新建 asgi_client.py：

# 导入异步IO库
import asyncio

# 从mcp模块导入ClientSession和types
from mcp import ClientSession, types

# 从mcp.client.streamable_http导入streamablehttp_client
from mcp.client.streamable_http import streamablehttp_client


# 定义异步函数call_echo，用于调用Echo子服务
async def call_echo() -> None:
    # 使用streamablehttp_client连接到本地的Echo子服务（/echo/mcp）
    async with streamablehttp_client("http://127.0.0.1:8000/echo/mcp") as (
        read,
        write,
        _,
    ):
        # 创建ClientSession会话，传入read和write对象
        async with ClientSession(read, write) as session:
            # 初始化会话
            await session.initialize()
            # 获取可用工具列表
            tools = await session.list_tools()
            # 打印Echo子服务提供的工具名称
            print("[Echo Tools]", [t.name for t in tools.tools])
            # 调用名为"echo"的工具，传入参数{"message": "hello"}
            res = await session.call_tool("echo", {"message": "hello"})
            # 创建空列表用于存储文本内容
            texts = []
            # 遍历返回内容
            for b in res.content:
                # 如果内容类型为TextContent，则提取文本
                if isinstance(b, types.TextContent):
                    texts.append(b.text)
            # 打印Echo工具的返回结果
            print("[Echo Result]", " | ".join(texts))


# 定义异步函数call_math，用于调用Math子服务
async def call_math() -> None:
    # 使用streamablehttp_client连接到本地的Math子服务（/math/mcp）
    async with streamablehttp_client("http://127.0.0.1:8000/math/mcp") as (
        read,
        write,
        _,
    ):
        # 创建ClientSession会话，传入read和write对象
        async with ClientSession(read, write) as session:
            # 初始化会话
            await session.initialize()
            # 获取可用工具列表
            tools = await session.list_tools()
            # 打印Math子服务提供的工具名称
            print("[Math Tools]", [t.name for t in tools.tools])
            # 调用名为"add_two"的工具，传入参数{"n": 5}
            res = await session.call_tool("add_two", {"n": 5})
            # 创建空列表用于存储文本内容
            texts = []
            # 遍历返回内容
            for b in res.content:
                # 如果内容类型为TextContent，则提取文本
                if isinstance(b, types.TextContent):
                    texts.append(b.text)
            # 打印Math工具的返回结果，如果没有文本内容则打印structuredContent属性
            print(
                "[Math Result]",
                " | ".join(texts) or str(getattr(res, "structuredContent", None)),
            )


# 定义主异步函数main
async def main() -> None:
    # 调用call_echo函数
    await call_echo()
    # 调用call_math函数
    await call_math()


# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 运行主异步函数main
    asyncio.run(main())

3.4 运行客户端 #

在运行 ASGI 服务器（上面的 uvicorn 命令）后，另开命令行运行客户端：

python asgi_client.py

预期输出（客户端）：

[Echo Tools] ['echo']
[Echo Result] Echo: hello
[Math Tools] ['add_two']
[Math Result] 7

3.5 理解 ASGI 服务器 #

3.5.1 lifespan - 应用生命周期管理器 #

lifespan 是 Starlette 应用的"开关管理员"，负责整个应用的启动和关闭。

职责：

• 启动阶段：初始化所有需要的资源和服务
• 运行阶段：保持应用正常运行
• 关闭阶段：确保所有资源正确清理，优雅退出

类比：就像大楼的物业经理，上班时打开所有办公室的门，下班时检查并锁好所有门。

3.5.2 `@contextlib.asynccontextmanager` - 异步上下文管理器工厂 #

asynccontextmanager 是一个装饰器，将普通异步函数变成异步上下文管理器。

Starlette 要求 lifespan 必须是一个异步上下文管理器，不能是普通函数。

@contextlib.asynccontextmanager
async def my_manager():
    # 启动代码
    setup()
    try:
        yield resource  # 在这里交出控制权
    finally:
        # 清理代码
        cleanup()

3.5.3 contextlib.AsyncExitStack - 多资源管理栈 #

AsyncExitStack 是一个"资源管理栈"，可以同时管理多个异步资源。

工作原理：

• 进入时：按顺序注册多个资源
• 退出时：按相反顺序自动清理所有资源
• 异常安全：即使某个资源清理失败，也会继续清理其他资源

类比：像是一个智能的多插线板，一键开启所有设备，断电时按正确顺序关闭所有设备。

3.5.4 stack.enter_async_context() - 资源注册方法 #

enter_async_context 将异步上下文管理器注册到退出栈中。

# 将 echo_mcp 的会话管理器注册到栈中
await stack.enter_async_context(echo_mcp.session_manager.run())

执行流程：

1. 调用 echo_mcp.session_manager.run() 返回上下文管理器
2. 进入该上下文管理器（启动服务）
3. 将退出逻辑注册到栈中
4. 返回资源对象（如果有）

3.5.5 echo_mcp.session_manager.run() - 会话管理器启动器 #

echo_mcp.session_manager.run()是启动 MCP 会话管理服务的方法。

具体功能：

• 启动后台服务（可能是 HTTP 服务器、WebSocket 监听器等）
• 管理客户端连接和会话
• 处理消息路由和协议通信

返回一个异步上下文管理器，用于控制服务的启动和停止。

3.5.6 echo_mcp.streamable_http_app() - HTTP 应用工厂 #

echo_mcp.streamable_http_app() 是创建基于 HTTP 的 MCP 服务应用。

功能：

• 创建标准的 Starlette/FastAPI 应用
• 在指定路径（如 /mcp）提供 MCP 协议端点
• 处理 HTTP 请求到 MCP 协议的转换

返回一个 Starlette 应用实例，可以挂载到路由中。

3.5.7 它们之间的关系图解 #

┌─────────────────────────────────────────────────────────────┐
│                   Starlette 主应用                          │
│                                                             │
│  ┌─────────────────┐  ┌─────────────────┐                   │
│  │    /echo/*      │  │    /math/*      │                   │
│  │                 │  │                 │                   │
│  │  Echo MCP App   │  │  Math MCP App   │                   │
│  │ (HTTP接口)      │  │ (HTTP接口)      │                   │
│  └─────────────────┘  └─────────────────┘                   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                lifespan 生命周期管理器               │   │
│  │                                                     │   │
│  │  ┌─────────────────────────────────────────────┐   │   │
│  │  │          AsyncExitStack 退出栈              │   │   │
│  │  │                                             │   │   │
│  │  │  ┌─────────────────┐  ┌─────────────────┐   │   │   │
│  │  │  │ Echo Session    │  │ Math Session    │   │   │   │
│  │  │  │ Manager         │  │ Manager         │   │   │   │
│  │  │  │ (后台服务)       │  │ (后台服务)       │   │   │   │
│  │  │  └─────────────────┘  └─────────────────┘   │   │   │
│  │  └─────────────────────────────────────────────┘   │   │
│  └─────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘

3.5.8 完整的工作流程 #

3.5.8.1 启动阶段 #

1. Starlette 启动，调用 lifespan.__aenter__()
2. 创建 AsyncExitStack，准备管理资源
3. 启动 Echo 会话管理器：await stack.enter_async_context(echo_mcp.session_manager.run())
4. 启动 Math 会话管理器：await stack.enter_async_context(math_mcp.session_manager.run())
5. yield 控制权，应用开始处理 HTTP 请求

3.5.8.2 运行阶段 #

• HTTP 请求到达 /echo/mcp → 由 echo_mcp.streamable_http_app() 处理
• HTTP 请求到达 /math/mcp → 由 math_mcp.streamable_http_app() 处理
• 后台会话管理器处理 MCP 协议通信

3.5.8.3 关闭阶段 #

1. 应用收到关闭信号，调用 lifespan.__aexit__()
2. AsyncExitStack 开始清理：
   • 先关闭 Math 会话管理器（后进先出）
   • 再关闭 Echo 会话管理器
3. 所有资源清理完成，应用安全退出

3.5.8.4 总结 #

• lifespan 管理整个应用生命周期
• @asynccontextmanager 提供优雅的上下文管理接口
• AsyncExitStack 确保多个资源的正确管理
• session_manager.run() 启动后台服务
• streamable_http_app() 提供 HTTP 接口

3.6 实现 ASGI 服务器 #

# 1. contextlib.asynccontextmanager 装饰器
# 定义一个将异步生成器函数转换为异步上下文管理器的装饰器
def asynccontextmanager(async_func):
    """将异步函数转换为异步上下文管理器"""

    # 定义异步上下文管理器类
    class AsyncContextManager:
        # 初始化，保存参数
        def __init__(self, *args, **kwargs):
            self.args = args  # 参数
            self.kwargs = kwargs  # 关键字参数

        # 进入异步上下文时调用
        async def __aenter__(self):
            # 创建并进入异步生成器
            self.gen = async_func(*self.args, **self.kwargs)
            # 返回异步生成器的下一个值
            return await self.gen.__anext__()

        # 退出异步上下文时调用
        async def __aexit__(self):
            # 退出异步上下文，清理资源
            try:
                # 返回异步生成器的下一个值
                await self.gen.__anext__()
                # 捕获停止异步迭代器异常
            except StopAsyncIteration:
                pass

    # 返回异步上下文管理器类
    return AsyncContextManager


# 2. AsyncExitStack 类
# 定义一个用于管理多个异步上下文的栈
class AsyncExitStack:
    """管理多个异步上下文的栈"""

    # 初始化，创建上下文列表
    def __init__(self):
        self.contexts = []  # 存储所有上下文

    # 进入一个异步上下文并添加到栈中
    async def enter_async_context(self, context):
        """进入一个异步上下文并添加到栈中"""
        # 进入异步上下文
        result = await context.__aenter__()
        # 添加到上下文列表
        self.contexts.append(context)
        # 返回结果
        return result

    # 进入整个栈的异步上下文
    async def __aenter__(self):
        # 返回自身
        return self

    # 退出时按相反顺序清理所有上下文
    # 参数：异常类型、异常值、异常追踪
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """退出时按相反顺序清理所有上下文"""
        # 按相反顺序清理所有上下文
        for context in reversed(self.contexts):
            # 退出异步上下文
            await context.__aexit__(exc_type, exc_val, exc_tb)


# 3. SessionManager 类
# 定义 MCP 服务器会话管理器
class SessionManager:
    """管理 MCP 服务器会话"""

    # 初始化，保存服务器名称和运行状态
    def __init__(self, server_name):
        self.server_name = server_name
        self.is_running = False

    # 返回一个会话运行的上下文管理器
    def run(self):
        """返回一个会话运行的上下文管理器"""

        # 定义会话上下文管理器
        class SessionContext:
            # 初始化，保存管理器引用
            def __init__(self, manager):
                self.manager = manager  # 保存管理器引用

            # 进入上下文时，启动会话
            async def __aenter__(self):
                print(f"Starting {self.manager.server_name} session...")  # 打印启动会话
                self.manager.is_running = True  # 设置运行状态
                return self  # 返回自身

            # 退出上下文时，关闭会话
            async def __aexit__(self, exc_type, exc_val, exc_tb):
                print(f"Stopping {self.manager.server_name} session...")  # 打印停止会话
                self.manager.is_running = False  # 设置运行状态

        # 返回会话上下文实例
        return SessionContext(self)


# 4. FastMCP 类
# 定义 MCP 服务器实现类
class FastMCP:
    """MCP 服务器实现"""

    # 初始化，保存名称、无状态标志、会话管理器和工具列表
    def __init__(self, name, stateless_http=True):
        self.name = name  # 保存名称
        self.stateless_http = stateless_http  # 保存无状态标志
        self.session_manager = SessionManager(name)  # 创建会话管理器
        self.tools = []  # 创建工具列表

    # 工具注册装饰器
    def tool(self):
        """工具注册装饰器"""

        # 装饰器函数，将工具函数加入工具列表
        def decorator(func):
            self.tools.append(func)  # 将工具函数加入工具列表
            return func  # 返回工具函数

        return decorator

    # 将 MCP 服务器转换为 HTTP ASGI 应用
    def streamable_http_app(self):
        """将 MCP 服务器转换为 HTTP ASGI 应用"""

        # 定义 HTTP ASGI 应用类
        class HTTPApp:
            # 初始化，保存 MCP 实例
            def __init__(self, mcp_instance):
                self.mcp = mcp_instance  # 保存 MCP 实例

            # ASGI 调用接口
            async def __call__(self, scope, receive, send):
                # 如果路径为 /mcp，处理 MCP 协议请求
                if scope["path"] == "/mcp":
                    # 处理 MCP 协议请求
                    await self.handle_mcp_request(scope, receive, send)
                else:
                    # 路径不匹配，返回 404
                    await send(
                        {
                            "type": "http.response.start",
                            "status": 404,
                        }
                    )
                    await send(
                        {
                            "type": "http.response.body",
                            "body": b"Not Found",
                        }
                    )

            # 处理 MCP 协议请求
            async def handle_mcp_request(self, scope, receive, send):
                # 处理实际的 MCP 协议
                print(f"Handling MCP request for {self.mcp.name}")  # 打印处理 MCP 请求
                # ... MCP 协议处理逻辑 ...

        # 返回 HTTPApp 实例
        return HTTPApp(self)


# 5. Starlette 应用
# 定义 ASGI Web 框架
class Starlette:
    """ASGI Web 框架"""

    # 初始化，保存路由和生命周期管理器
    def __init__(self, routes=None, lifespan=None):
        self.routes = routes or []  # 保存路由
        self.lifespan = lifespan  # 保存生命周期管理器

    # ASGI 应用调用接口
    async def __call__(self, scope, receive, send):
        """ASGI 应用调用接口"""
        # 1. 如果是生命周期启动事件
        if scope["type"] == "lifespan":
            await self.handle_lifespan(scope, receive, send)
            return

        # 2. 如果是 HTTP 请求
        if scope["type"] == "http":
            await self.handle_http(scope, receive, send)

    # 处理应用生命周期事件
    async def handle_lifespan(self, scope, receive, send):
        """处理应用生命周期"""
        # 启动事件
        if scope["message"] == "startup":
            # 启动时调用 lifespan 管理器
            if self.lifespan:
                self.lifespan_context = self.lifespan(self)  # 创建生命周期上下文
                await self.lifespan_context.__aenter__()  # 进入生命周期上下文

            await send(
                {"type": "lifespan.startup.complete"}
            )  # 发送生命周期启动完成事件

        # 关闭事件
        elif scope["message"] == "shutdown":
            # 关闭时清理 lifespan 管理器
            if hasattr(self, "lifespan_context"):
                await self.lifespan_context.__aexit__(
                    None, None, None
                )  # 退出生命周期上下文

            await send(
                {"type": "lifespan.shutdown.complete"}
            )  # 发送生命周期关闭完成事件

    # 处理 HTTP 请求
    async def handle_http(self, scope, receive, send):
        """处理 HTTP 请求"""
        path = scope["path"]

        # 路由匹配
        for route in self.routes:
            if path.startswith(route.path):
                # 将请求转发到对应的子应用
                sub_scope = scope.copy()
                sub_scope["path"] = path[len(route.path) :]  # 移除前缀
                await route.app(sub_scope, receive, send)  # 调用子应用
                return

        # 没有匹配的路由，返回 404
        await send(
            {
                "type": "http.response.start",
                "status": 404,
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": b"Not Found",
            }
        )


# 6. Mount 路由
# 定义路由挂载类
class Mount:
    """路由挂载"""

    # 初始化，保存路径和子应用
    def __init__(self, path, app):
        self.path = path  # 保存路径
        self.app = app  # 保存子应用


# 7. 使用示例（原始代码）
# 使用 asynccontextmanager 装饰器定义生命周期管理器
@asynccontextmanager
async def lifespan(app):
    """生命周期管理器伪代码"""
    # 创建异步退出栈，确保多个上下文正确进入和退出
    async with AsyncExitStack() as stack:
        # 启动两个会话管理器
        await stack.enter_async_context(
            echo_mcp.session_manager.run()
        )  # 进入 echo_mcp 会话管理器
        await stack.enter_async_context(
            math_mcp.session_manager.run()
        )  # 进入 math_mcp 会话管理器
        # 应用运行期间保持会话开启
        yield


# 创建 MCP 服务器实例
# 创建 EchoServer 实例
echo_mcp = FastMCP(name="EchoServer", stateless_http=True)
# 创建 MathServer 实例
math_mcp = FastMCP(name="MathServer", stateless_http=True)

# 创建 Starlette 应用
app = Starlette(
    routes=[
        # 挂载 echo_mcp 到 /echo 路径
        Mount("/echo", echo_mcp.streamable_http_app()),
        # 挂载 math_mcp 到 /math 路径
        Mount("/math", math_mcp.streamable_http_app()),
    ],
    # 指定生命周期管理器
    lifespan=lifespan,
)


# 8. ASGI 服务器启动流程
# 定义模拟服务器启动的异步函数
async def simulate_server_start():
    """服务器启动过程"""
    # 打印服务器启动序列
    print("=== Server Startup Sequence ===")

    # 创建模拟的 send 函数
    async def mock_send(message):
        print(f"ASGI Send: {message}")  # 打印 ASGI 发送消息

    # 1. ASGI 服务器发送 lifespan startup 事件
    lifespan_scope = {"type": "lifespan", "message": "startup"}

    # 2. Starlette 接收到 startup 事件，调用 lifespan 管理器
    await app(lifespan_scope, None, mock_send)  # 调用 lifespan 管理器

    # 3. lifespan 管理器创建 AsyncExitStack
    # 4. AsyncExitStack 进入两个 session_manager.run() 上下文
    # 5. 两个会话管理器启动完成
    # 6. yield 执行，应用进入运行状态

    # 打印应用运行中
    print("=== Application Running ===")

    # 7. HTTP 请求
    http_scope = {"type": "http", "method": "POST", "path": "/echo/mcp"}
    await app(http_scope, None, mock_send)  # 调用 HTTP 请求

    # 打印服务器关闭序列
    print("=== Server Shutdown Sequence ===")

    # 8. ASGI 服务器发送 lifespan shutdown 事件
    lifespan_scope = {"type": "lifespan", "message": "shutdown"}
    await app(lifespan_scope, None, mock_send)  # 调用 lifespan 管理器 关闭

    # 9. lifespan 管理器退出，AsyncExitStack 按相反顺序清理所有上下文
    # 10. 两个会话管理器停止


# 运行
# 导入 asyncio
import asyncio

# 运行模拟服务器启动流程
asyncio.run(simulate_server_start())