aiohttp-mcp

在aiohttp之上构建Model Context Protocol (MCP)服务器的工具。

特性

• 与aiohttp Web应用程序轻松集成
• 支持Model Context Protocol (MCP)工具
• 异步优先设计
• 类型提示支持
• 开发调试模式
• 灵活的路由选项

安装

使用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

# 初始化MCP
mcp = AiohttpMCP()


# 定义一个工具
@mcp.tool()
def get_time(timezone: str) -> str:
    """获取指定时区的当前时间。"""
    tz = ZoneInfo(timezone)
    return datetime.datetime.now(tz).isoformat()


# 创建并运行应用
app = build_mcp_app(mcp, path="/mcp")
web.run_app(app)

作为子应用使用

你也可以在现有的aiohttp服务器中使用aiohttp-mcp作为子应用：

import datetime
from zoneinfo import ZoneInfo

from aiohttp import web

from aiohttp_mcp import AiohttpMCP, setup_mcp_subapp

mcp = AiohttpMCP()


# 定义一个工具
@mcp.tool()
def get_time(timezone: str) -> str:
    """获取指定时区的当前时间。"""
    tz = ZoneInfo(timezone)
    return datetime.datetime.now(tz).isoformat()


# 创建你的主应用
app = web.Application()

# 添加MCP作为子应用
setup_mcp_subapp(app, mcp, prefix="/mcp")

web.run_app(app)

使用流式HTTP传输

对于需要高级会话管理的生产部署，你可以使用流式HTTP传输模式：

import datetime
from zoneinfo import ZoneInfo

from aiohttp import web

from aiohttp_mcp import AiohttpMCP, TransportMode, build_mcp_app

# 初始化MCP
mcp = AiohttpMCP()


# 定义一个工具
@mcp.tool()
def get_time(timezone: str) -> str:
    """获取指定时区的当前时间。"""
    tz = ZoneInfo(timezone)
    return datetime.datetime.now(tz).isoformat()


# 创建具有流式传输的应用
app = build_mcp_app(mcp, path="/mcp", transport_mode=TransportMode.STREAMABLE_HTTP, stateless=True)
web.run_app(app)

客户端示例

这里是如何创建一个与MCP服务器交互的客户端：

import asyncio

from mcp import ClientSession
from mcp.client.sse import sse_client


async def main():
    # 连接到MCP服务器
    async with sse_client("http://localhost:8080/mcp") as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化会话
            await session.initialize()

            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具：", [tool.name for tool in tools.tools])

            # 调用一个工具
            result = await session.call_tool("get_time", {"timezone": "UTC"})
            print("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  # 在Windows上：.venv\Scripts\activate

3. 安装开发依赖：

uv sync --all-extras

运行测试

uv run pytest

要求

• Python 3.10或更高版本
• aiohttp >= 3.9.0, < 4.0.0
• aiohttp-sse >= 2.2.0, < 3.0.0
• anyio >= 4.9.0, < 5.0.0
• mcp >= 1.8.0, < 2.0.0

许可证

本项目根据MIT许可证发布 - 查看LICENSE文件以获取详情。

贡献

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