aiohttp-mcp

aiohttp-mcp

基于 aiohttp 构建 Model Context Protocol (MCP) 服务的工具。

原生实现 MCP 协议，无需依赖庞大的 SDK。仅有 3 个运行时依赖：aiohttp、aiohttp-sse、pydantic。

功能特性

• 原生实现 MCP 协议（支持规范 2025-11-25、2025-06-18、2025-03-26）
• 支持 SSE 流式传输的 Streamable HTTP 传输
• 可轻松集成到 aiohttp Web 应用
• 支持工具、资源和提示词，并通过装饰器注册
• 通过 ctx.app 共享状态，并通过 ctx.request 获取每个请求的数据
• 默认无状态，可选有状态模式，支持服务端推送和恢复
• 支持用于恢复的事件存储
• 异步优先设计，提供完整类型提示
• 支持非流式部署的 JSON 响应模式

安装

使用 uv 包管理器：

uv add aiohttp-mcp

或使用 pip：

pip install aiohttp-mcp

快速开始

基本服务端设置

创建一个带有自定义工具的简单 MCP 服务：

import datetime
from zoneinfo import ZoneInfo

from aiohttp import web

from aiohttp_mcp import AiohttpMCP, build_mcp_app

# Initialize MCP
mcp = AiohttpMCP()


# Define a tool
@mcp.tool()
def get_time(timezone: str) -> str:
    """Get the current time in the specified timezone."""
    tz = ZoneInfo(timezone)
    return datetime.datetime.now(tz).isoformat()


# Create and run the application
app = build_mcp_app(mcp, path="/mcp")
web.run_app(app)

作为子应用使用

也可以将 aiohttp-mcp 作为子应用用于现有的 aiohttp 服务：

import datetime
from zoneinfo import ZoneInfo

from aiohttp import web

from aiohttp_mcp import AiohttpMCP, setup_mcp_subapp

mcp = AiohttpMCP()


# Define a tool
@mcp.tool()
def get_time(timezone: str) -> str:
    """Get the current time in the specified timezone."""
    tz = ZoneInfo(timezone)
    return datetime.datetime.now(tz).isoformat()


# Create your main application
app = web.Application()

# Add MCP as a sub-application
setup_mcp_subapp(app, mcp, prefix="/mcp")

web.run_app(app)

有状态模式与恢复

默认情况下，服务以无状态模式运行——每个请求都会创建一个全新的传输，因此适用于负载均衡和多实例部署。工具通知（ctx.info()）通过 SSE POST 响应内联工作。

对于需要服务端主动推送（通过 GET SSE 流）或 SSE 恢复能力的单实例部署，可启用有状态模式。会话状态和事件存储在进程内存中——若未使用粘性会话，则不适用于多实例部署。

from aiohttp_mcp import AiohttpMCP, InMemoryEventStore, build_mcp_app

# Stateful with resumability (single instance only)
# If client disconnects, it can reconnect with Last-Event-ID to replay missed events
mcp = AiohttpMCP(event_store=InMemoryEventStore())
app = build_mcp_app(mcp, path="/mcp", stateless=False)

注意： InMemoryEventStore 仅限进程内使用。对于多实例有状态部署，请实现一个由共享存储（例如 Redis）支持的自定义 EventStore，并使用粘性会话。

访问上下文

在工具中访问 MCP 上下文有 3 种方式。它们都会返回同一个 Context 对象：

1. get_current_context() — 模块函数

from aiohttp_mcp import get_current_context

@mcp.tool()
async def my_tool(query: str) -> str:
    ctx = get_current_context()
    user_id = ctx.request.headers.get("X-User-ID", "anonymous")
    await ctx.info(f"Query by {user_id}")
    return f"Result for {user_id}"

2. mcp.get_context() — 实例方法

@mcp.tool()
async def my_tool(query: str) -> str:
    ctx = mcp.get_context()
    user_id = ctx.request.headers.get("X-User-ID", "anonymous")
    return f"Result for {user_id}"

3. ctx: Context — 参数注入

将 ctx: Context 声明为参数，即可自动注入，并从工具的输入架构中排除：

from aiohttp_mcp import Context

@mcp.tool()
async def my_tool(query: str, ctx: Context) -> str:
    user_id = ctx.request.headers.get("X-User-ID", "anonymous")
    return f"Result for {user_id}"

上下文功能：

• ctx.request — aiohttp Request（标头、Cookie、客户端 IP）
• ctx.app — aiohttp Application，用于共享状态（ctx.app["db_pool"]）
• ctx.request_id — JSON-RPC 请求 ID
• await ctx.info(msg) / debug() / warning() / error() — 向客户端发送日志
• await ctx.report_progress(progress, total) — 报告进度
• await ctx.read_resource(uri) — 读取已注册的资源

通过 ctx.app 共享状态：

from collections.abc import AsyncIterator

from aiohttp import web

from aiohttp_mcp import AiohttpMCP, build_mcp_app, get_current_context

mcp = AiohttpMCP()


@mcp.tool()
async def secure_query(sql: str) -> str:
    """Run a database query with auth validation."""
    ctx = get_current_context()
    db_pool = ctx.app["db_pool"]
    return await db_pool.query(sql)


async def startup(app: web.Application) -> AsyncIterator[None]:
    app["db_pool"] = await create_db_pool()
    yield
    await app["db_pool"].close()


app = build_mcp_app(mcp, path="/mcp")
app.cleanup_ctx.append(startup)

资源组合

工具可以在执行期间通过 ctx.read_resource(uri) 读取已注册的资源，避免逻辑重复：

from aiohttp_mcp import AiohttpMCP, get_current_context

mcp = AiohttpMCP()


@mcp.resource("config://{service}")
async def get_config(service: str) -> str:
    """Service configuration exposed as a resource."""
    return load_config(service)


@mcp.tool()
async def deploy(service: str) -> str:
    """Deploy a service using its registered config."""
    ctx = get_current_context()
    config = await ctx.read_resource(f"config://{service}")
    return f"Deployed {service} with {config}"

该调用会使用 MCP 客户端所用的同一 URI 回调资源注册表——工具只需 URI，无需直接引用资源函数。

结构化返回类型

工具可以返回 Pydantic BaseModel 或 dataclass 实例——这些实例会自动序列化为 JSON，并在 tools/list 响应中生成 outputSchema：

import dataclasses
from pydantic import BaseModel
from aiohttp_mcp import AiohttpMCP

mcp = AiohttpMCP()


@dataclasses.dataclass
class UserData:
    name: str
    email: str
    age: int = 25


class UserResult(BaseModel):
    id: str
    name: str
    email: str


@mcp.tool()
async def create_user(data: UserData) -> UserResult:
    """Create a new user."""
    # Input: dataclass validated from dict/JSON automatically
    # Output: BaseModel serialized to JSON, outputSchema auto-generated
    return UserResult(id="123", name=data.name, email=data.email)

普通类型（str、dict、list）仍按原样序列化。任何返回类型注解都会生成 outputSchema；对于 BaseModel 和 dataclass 返回值，还会通过 Pydantic 的 TypeAdapter 进行适当的 JSON 序列化，而不是使用 str()。

客户端示例

以下示例展示了如何使用 mcp 客户端库创建与 MCP 服务交互的客户端：

import asyncio

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


async def main():
    # Connect to the MCP server
    async with streamablehttp_client("http://localhost:8080/mcp") as (
        read_stream,
        write_stream,
        _,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the session
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print("Available tools:", [tool.name for tool in tools.tools])

            # Call a tool
            result = await session.call_tool("get_time", {"timezone": "UTC"})
            print("Current time in UTC:", result.content)


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

更多示例

如需了解更多示例，请查看 examples 目录。

开发

设置开发环境

1. 克隆仓库：

git clone https://github.com/kulapard/aiohttp-mcp.git
cd aiohttp-mcp

2. 创建并激活虚拟环境：

uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

3. 安装开发依赖项：

uv sync --all-extras

运行测试

uv run pytest

要求

• Python 3.11 或更高版本
• aiohttp >= 3.9.0，< 4.0.0
• aiohttp-sse >= 2.2.0，< 3.0.0
• pydantic >= 2.0.0，< 3.0.0

许可证

本项目采用 MIT 许可证授权——详情请参阅 LICENSE 文件。

贡献

欢迎贡献！欢迎随时提交 Pull Request。