AI导哈

MCP服务器工具

aiohttp-mcp是一个基于aiohttp的Model Context Protocol (MCP)服务器构建工具，支持MCP协议的原生实现，提供流式HTTP传输、工具和资源注册以及状态管理功能。

  • by kulapard
  • Python
  • 本地部署
  • 8 星标
  • 更新于
在 GitHub 查看 kulapard/aiohttp-mcp

使用说明

aiohttp mcp

建筑工具 模型上下文协议（MCP） 服务器位于 意图tp.

本机实现MCP协议，无需大量依赖SDK。只有3个运行时依赖关系： aiohttp, aiohttp-sse, pydantic.

特性

  • 本地MCP协议实现（支持规范2025-11-25、2025-06-18、2025-03-26）
  • 使用SSE流式传输的可流式HTTP传输
  • 易于与aiohttp web应用程序集成
  • 基于装饰器注册的工具、资源和提示支持
  • 共享状态通过 ctx.app 根据请求，数据通过 ctx.request
  • 默认情况下为无状态，具有用于服务器推送和可恢复性的可选状态模式
  • 事件存储支持可恢复性
  • 使用完整类型提示异步第一设计
  • 非流部署的JSON响应模式

安装

随着 紫外线 包管理器：

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服务器中将aiohttp-mcp用作子应用程序：

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 仅在处理过程中。对于多实例有状态部署，实现自定义 EventStore 由共享存储（如Redis）支持，并使用粘性会话。

上下文访问

有三种方法可以访问工具内的MCP上下文。所有返回相同 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 — 意图 Request （标头、Cookie、客户端IP）
  • ctx.app — 意图 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并生成 outputSchema 在 tools/list 响应：

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获得正确的JSON序列化 TypeAdapter 而不是 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())

更多示例

有关更多示例，请查看 例子 目录。

发展

设置开发环境

  1. 克隆存储库：
git clone https://github.com/kulapard/aiohttp-mcp.git
cd aiohttp-mcp
  1. 创建并激活虚拟环境：
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
  1. 安装开发依赖项：
uv sync --all-extras

运行测试

uv run pytest

需求

  • Python 3.11或更高版本
  • aiohttp >= 3.9.0, < 4.0.0
  • aiohttp >= 2.2.0， < 3.0.0
  • 媒染剂>=2.0.0，<3.0.0

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

贡献

欢迎投稿！请随时提交拉取请求。

相关 MCP