aiohttp-mcp

Tools for building Model Context Protocol (MCP) servers on top of aiohttp.

Implements the MCP protocol natively — no heavy SDK dependencies. Only 3 runtime dependencies: aiohttp, aiohttp-sse, pydantic.

Features

• Native MCP protocol implementation (supports specs 2025-11-25, 2025-06-18, 2025-03-26)
• Streamable HTTP transport with SSE streaming
• Easy integration with aiohttp web applications
• Tool, Resource, and Prompt support with decorator-based registration
• Shared state via ctx.app and per-request data via ctx.request
• Stateless by default, with optional stateful mode for server push and resumability
• Event store support for resumability
• Async-first design with full type hints
• JSON response mode for non-streaming deployments

Installation

With uv package manager:

uv add aiohttp-mcp

Or with pip:

pip install aiohttp-mcp

Quick Start

Basic Server Setup

Create a simple MCP server with a custom tool:

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)

Using as a Sub-Application

You can also use aiohttp-mcp as a sub-application in your existing aiohttp server:

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)

Stateful Mode & Resumability

By default, the server runs in stateless mode — each request creates a fresh transport, making it safe for load-balanced and multi-instance deployments. Tool notifications (ctx.info()) work inline via SSE POST responses.

For single-instance deployments that need server-initiated push (via GET SSE stream) or SSE resumability, enable stateful mode. Session state and events are stored in-process memory — this is not suitable for multi-instance deployments without sticky sessions.

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)

Note: InMemoryEventStore is in-process only. For multi-instance stateful deployments, implement

... View full README on GitHub