aiohttp-mcp

aiohttp-mcp

aiohttp 上で Model Context Protocol (MCP) サーバーを構築するためのツールです。

MCP プロトコルをネイティブに実装しており、重量級の SDK 依存関係はありません。実行時の依存関係は aiohttp、aiohttp-sse、pydantic の 3 つだけです。

特長

• MCP プロトコルをネイティブに実装（仕様 2025-11-25、2025-06-18、2025-03-26 に対応）
• SSE ストリーミングに対応した Streamable HTTP トランスポート
• aiohttp ウェブアプリケーションとの簡単な連携
• デコレーターによる登録で、ツール、リソース、プロンプトに対応
• 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 サーバーで、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 はプロセス内でのみ使用できます。ステートフルな複数インスタンスのデプロイでは、共有ストレージ（例：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 の戻り値は、str() ではなく Pydantic の TypeAdapter を介して適切に JSON にシリアライズされます。

クライアントの例

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 の下でライセンスされています。詳細については LICENSE ファイルを参照してください。

貢献

貢献を歓迎します！お気軽にプルリクエストを送信してください。