API

Declaring what a server offers

class aiohttp_tiny_mcp.Registry(name, version, *, hub=None, session_store=None, auth=None, instructions=None, session_ttl_seconds=3600, request_state_ttl_seconds=600, hub_poll_seconds=30.0, ask_timeout_seconds=120.0)[source]

Register handlers as decorators or by passing functions directly.

@registry.tool async def add(args: Add) -> int: …

registry.tool(add, name=”sum”) registry.resource(“config://app”, config)

Uses process-local memory backends by default. Supply shared hub and session_store implementations before running more than one worker.

Parameters:
provide(kind, source)[source]

Bind a type to where it comes from. Call before registering handlers – plans are checked at registration, not at first call.

source is an async callable taking the exchange, an async generator for anything that must be released afterwards, or a web.AppKey. For an object that already exists, use provide_instance instead.

Parameters:
Return type:

None

provide_instance(value, kind=None)[source]

Register an existing object under its type, or under an explicit kind such as a base class or protocol. Use provide for per-request construction or cleanup.

Parameters:
Return type:

None

completions(fn)[source]

(CompleteParams-shaped model, deps) -> list[str] | Completion.

Parameters:

fn (Callable[[...], Awaitable[Any]])

Serving it

class aiohttp_tiny_mcp.Endpoint(registry, *, adapters=None, allowed_origins=None, trust_proxy_origin_validation=False, compress=True)[source]

The MCP endpoint, in the spec’s sense: one path that accepts POST.

Mount it however you mount anything else in aiohttp:

app.add_routes(ep.routes("/mcp"))
app.router.add_view("/mcp", ep.view)
ep.setup(app, "/mcp")  # the same routes, added for you

Under add_subapp the metadata route needs the root application, because a prefix must not reach a well-known path:

section.add_routes(ep.routes("/mcp", metadata=False))
app.add_subapp("/api/", section)
app.add_routes(ep.metadata_routes())
Parameters:
property view: type[View][source]

GET opens the legacy notification stream; POST handles requests.

2026-07-28 uses subscriptions/listen instead of GET. DELETE returns 405; sessions end by expiration.

routes(path='/mcp', *, name='mcp', metadata=True)[source]

The endpoint, and where a client looks to find out how to reach it.

The metadata route is included whenever tokens are verified, because a client that has no token learns where to get one from there and nowhere else. Pass metadata=False where this application cannot serve that path – see metadata_routes.

Parameters:
Return type:

list[RouteDef]

class aiohttp_tiny_mcp.SseEndpoint(registry, *, adapters=None, allowed_origins=None, trust_proxy_origin_validation=False, compress=True, sse_path='/sse', message_path='/messages')[source]

Two endpoints: one to listen on, one to send to.

Parameters:
routes(sse_path=None, message_path=None)[source]

One route to listen on, one to post to.

Either path may be set here or on the constructor. Both are kept, because the stream names the posting path to the client.

Parameters:

  • sse_path (str | None)

  • message_path (str | None)

Return type:

list[RouteDef]

async aiohttp_tiny_mcp.run_stdio(registry, *, adapters=None, app_state=None)[source]

Serve registry over real stdin/stdout until stdin closes.

Parameters:
Return type:

None

Inside a handler

class aiohttp_tiny_mcp.Exchange(registry, request, adapter, call, session=None)[source]
Parameters:
session

Application values kept between calls, or None where this request reached no session.

property can_ask: bool

MRTR requires declared elicitation; the tool-argument fallback does not.

Type:

Whether input can be requested

property sessions: SessionAccess

Access sessions by explicit handles, including on revisions without protocol sessions.

answered(key)[source]

Check for a reply, including accepted forms with empty content.

Parameters:

key (str)

Return type:

bool

accepted(key)[source]

Check the action; accepted and declined replies can both have empty content.

Parameters:

key (str)

Return type:

bool

async ask(key, request, *, default=None)[source]

Ask for input and return the answer.

MRTR raises NeedInput and restarts the handler when the answer arrives, possibly on another node. Put irreversible work after the final ask; preceding work may run again.

For clients that cannot be asked, default accepts an elicit_accept/decline/cancel result. Without a default, the call fails.

Parameters:
Return type:

Answer

action(key)[source]

Return ACCEPT, DECLINE, CANCEL, or None if unanswered.

Parameters:

key (str)

Return type:

AnswerAction | None

logs(level)[source]

Whether a message of level would reach this client.

Parameters:

level (str)

Return type:

bool

async log(level, data, *, logger=None)[source]

Emit a message at the requested severity on the current request stream. Without a stream, emit nothing.

Parameters:
Return type:

None

async progress(progress, total=None, message=None)[source]

Emit on the current HTTP response or stdio request.

Parameters:
Return type:

None

aiohttp_tiny_mcp.elicit(message, schema=None)[source]

Ask the user something. Omit schema to ask only for agreement.

Parameters:
Return type:

Mapping[str, Any]

aiohttp_tiny_mcp.elicit_accept(content)[source]
Parameters:

content (Mapping[str, Any])

Return type:

dict[str, Any]

aiohttp_tiny_mcp.elicit_decline()[source]
Return type:

dict[str, Any]

aiohttp_tiny_mcp.elicit_cancel()[source]
Return type:

dict[str, Any]

class aiohttp_tiny_mcp.NeedInput(requests, state=None)[source]

Raise from a tool/prompt/resource handler to trigger MRTR.

Parameters:

  • requests (Mapping[str, InputRequest])

  • state (Any)

class aiohttp_tiny_mcp.core.Answer(action, content=<factory>)[source]

Unwrapped reply content and action. Empty content may still mean acceptance.

Parameters:

Sessions and state

class aiohttp_tiny_mcp.sessions.Session(store, session_id, record, ttl_seconds=3600)[source]

Application session values, read from the request snapshot and written with compare-and-set retries.

Parameters:
async pause(attempt)[source]

Use jitter between retries to avoid collisions; do not delay the first attempt.

Parameters:

attempt (int)

Return type:

None

async set(key, value)[source]

Store one value. It must survive a JSON round trip.

Parameters:
Return type:

None

async replace(values)[source]

Replace the entire application-owned mapping.

Parameters:

values (Mapping[str, Any])

Return type:

None

async remember_version(version)[source]

Store the revision this client negotiated.

Written where a handshake happens on a transport that keeps no header to carry it – see http_sse.py. The header path writes the same key when it opens a session.

Parameters:

version (str)

Return type:

None

async set_log_level(level)[source]

Persist the requested log severity.

Parameters:

level (str)

Return type:

None

async write_value(key, value)[source]

Store one top-level value under compare-and-set.

A plain value rather than a slot, so it is written the same way whichever worker handles the next request.

Parameters:
Return type:

None

async update(change)[source]

Apply change to the application’s values under compare-and-set.

Parameters:

change (Callable[[Mapping[str, Any]], Mapping[str, Any]])

Return type:

None

async update_slot(name, change)[source]

Apply change under compare-and-set, recomputing it from fresh values on every retry.

Parameters:
Return type:

None

class aiohttp_tiny_mcp.sessions.SessionAccess(store, ttl_seconds=3600)[source]

Sessions addressed by explicit handles on any revision.

Return a handle to the caller and accept it as an argument on subsequent requests. Handles and legacy session headers resolve to the same Session interface.

Parameters:
async open()[source]

Create an empty session and return the handle callers must send back.

Return type:

Session

async use(handle)[source]

Resolve a handle within the current namespace, or return None.

Parameters:

handle (str)

Return type:

Session | None

class aiohttp_tiny_mcp.sessions.SessionStore(*args, **kwargs)[source]

Application-provided storage, safe across workers.

create() atomically returns False for an existing id. save() atomically returns False when expected_version no longer matches. Store only JSON-serializable values, never requests, sockets, queues, or tasks.

Subclass it to have the methods checked and the missing ones refused, or supply any object with these four methods: this is a protocol, so a backend that inherits nothing is still a SessionStore.

abstractmethod async create(session_id, data, *, ttl_seconds)[source]

Create with a TTL. Return False where a live record already holds the id.

Parameters:
Return type:

bool

abstractmethod async get(session_id)[source]

The live record and its version, or None where it is missing or expired.

Parameters:

session_id (str)

Return type:

SessionRecord | None

abstractmethod async save(session_id, data, *, expected_version, ttl_seconds)[source]

Replace the data and renew the TTL, or return False where the version moved.

Parameters:
Return type:

bool

abstractmethod async delete(session_id)[source]

Forget it.

Parameters:

session_id (str)

Return type:

None

class aiohttp_tiny_mcp.MemorySessionStore(*, clock=<built-in function monotonic>)[source]

Process-local session store for development, tests, and single-worker deployments.

Parameters:

clock (Callable[[], float])

class aiohttp_tiny_mcp.request_state.RequestStates(store, ttl_seconds=600)[source]

State rows, in the store the deployment already supplies.

Parameters:
async open(call, payload)[source]

Store payload and return its opaque client-visible id.

Parameters:
Return type:

str

async read(call, state_id)[source]

Read the payload; raise KeyError for missing state or a different call binding.

Parameters:
Return type:

Any

Events

class aiohttp_tiny_mcp.hub.Hub(*args, **kwargs)[source]

Application-provided event storage.

Subclass it to have the methods checked and the missing ones refused, or supply any object with these four methods: this is a protocol, so a backend that inherits nothing is still a Hub.

abstractmethod async publish(topic, message)[source]

Append one message to topic.

Parameters:
Return type:

None

abstractmethod async position(topic)[source]

Capture the cursor before triggering a publish so replies preceding the first poll are included.

Parameters:

topic (str)

Return type:

str

abstractmethod async poll(topic, cursor, *, timeout)[source]

Return messages after cursor and the next cursor.

Wait up to timeout seconds for a message. Backends choose whether to poll or wait for notification; timeout is a deadline, not a polling interval.

Parameters:
Return type:

tuple[Sequence[Mapping[str, Any]], str]

abstractmethod async delete(topic)[source]

Delete a completed topic.

Parameters:

topic (str)

Return type:

None

class aiohttp_tiny_mcp.MemoryHub[source]

Single-process hub, waiting on a condition rather than sleeping.

aiohttp_tiny_mcp.hub.topic(kind, name=None)[source]

Prefix the topic with the current namespace to isolate callers.

Parameters:

  • kind (str)

  • name (str | None)

Return type:

str

On SQLite

Both backends on one file, for workers on one machine. Needs aiosqlite: install aiohttp-tiny-mcp[sqlite]. See Stores and hubs.

class aiohttp_tiny_mcp.sqlite.SqliteStorage(path, *, busy_timeout_ms=None, event_ttl_seconds=None, sweep_seconds=None)[source]

One SQLite connection shared by the store and hub.

Parameters:

  • path (str | Path)

  • busy_timeout_ms (int)

  • event_ttl_seconds (float)

  • sweep_seconds (float)

async sweep()[source]

Remove what has expired. Safe to call at any time, from any worker.

Return type:

None

async sweeping(*, every=None)[source]

Sweep until cancelled. A failed sweep is logged and tried again.

Parameters:

every (float | None)

Return type:

None

async cleanup_ctx(app)[source]

Manage the connection and sweeper for an aiohttp app.

Parameters:

app (Any)

Return type:

AsyncIterator[None]

class aiohttp_tiny_mcp.sqlite.SqliteSessionStore(storage)[source]

SessionStore with version-checked writes, so two workers cannot lose each other’s.

Parameters:

storage (SqliteStorage)

class aiohttp_tiny_mcp.sqlite.SqliteHub(storage, *, look_again=None)[source]

A cursor-based Hub backed by one SQLite table.

Parameters:

Streams

class aiohttp_tiny_mcp.SSEResponse(*, heartbeat=15.0, retry=None, max_queue=1024, compress=True, headers=None, **kwargs)[source]

A queued text/event-stream response.

Parameters:

  • heartbeat (float | None)

  • retry (int | None)

  • max_queue (int)

  • compress (bool)

  • headers (Mapping[str, str] | None)

  • kwargs (Any)

property closed: bool

Never opened, closed, or the write failed.

async close()[source]

Write what is queued, then stop.

Return type:

None

class aiohttp_tiny_mcp.sse.SSEEvent(data=None, event=None, id=None, retry=None, comment=None)[source]

One event-stream frame.

Parameters:

  • data (str | None)

  • event (str | None)

  • id (str | None)

  • retry (int | None)

  • comment (str | None)

async aiohttp_tiny_mcp.sse.read_sse(response, *, comments=False)[source]

Decode events from an SSE response.

Parameters:

  • response (ClientResponse)

  • comments (bool)

Return type:

AsyncIterator[SSEEvent]

On Redis

Both backends on a Redis server, for workers on more than one machine. Needs redis: install aiohttp-tiny-mcp[redis]. See Stores and hubs.

class aiohttp_tiny_mcp.redis.RedisStorage(client, *, prefix=None, owned=False)[source]

A Redis client shared by the store and hub.

Parameters:

  • client (Redis)

  • prefix (str)

  • owned (bool)

classmethod from_url(url, **settings)[source]

Build an owned Redis client from a URL.

Parameters:
Return type:

RedisStorage

async close()[source]

Close an owned client.

Return type:

None

async cleanup_ctx(app)[source]

Manage an owned client for an aiohttp app.

Parameters:

app (Any)

Return type:

AsyncIterator[None]

class aiohttp_tiny_mcp.redis.RedisSessionStore(storage, *, prefix=None)[source]

SessionStore on one key per session, with version-checked writes.

Parameters:
class aiohttp_tiny_mcp.redis.RedisHub(storage, *, prefix=None, keep=None, ttl_seconds=None)[source]

A Redis stream per topic.

Parameters:

On PostgreSQL

Both backends on a PostgreSQL server. Needs psycopg: install aiohttp-tiny-mcp[postgres]. See Stores and hubs.

class aiohttp_tiny_mcp.postgres.PostgresStorage(pool, *, prefix=None, event_ttl_seconds=None, sweep_seconds=None, create_tables=None, owned=False)[source]

A pool, schema, and deployment maintenance state.

Parameters:

  • pool (AsyncConnectionPool)

  • prefix (str)

  • event_ttl_seconds (int)

  • sweep_seconds (float)

  • create_tables (bool)

  • owned (bool)

classmethod from_url(conninfo, *, min_size=4, max_size=None, **settings)[source]

Build an owned pool from a connection string.

Parameters:

  • conninfo (str)

  • min_size (int)

  • max_size (int | None)

  • settings (Any)

Return type:

PostgresStorage

property state_table: Identifier

Whatever this deployment has to remember between runs, by key.

async open()[source]

Open the pool and create the tables, once.

Return type:

AsyncConnectionPool

async sweep()[source]

Remove expired rows now. Use sweep_if_due on multiple workers.

Return type:

None

async sweep_if_due(*, every=None)[source]

Sweep once per interval across all workers; return whether this one did.

Parameters:

every (float | None)

Return type:

bool

async due(every, connection=None)[source]

Whether the last sweep was longer than every seconds ago.

Parameters:
Return type:

bool

async remember(key, value, connection=None)[source]

Store one value under key, stamping updated_at.

Parameters:
Return type:

None

async recall(key)[source]

What was stored under key, or None.

Parameters:

key (str)

Return type:

Any | None

async sweeping(*, every=None)[source]

Sweep repeatedly; safe to run on every worker.

Parameters:

every (float | None)

Return type:

None

async cleanup_ctx(app)[source]

Open and close the pool for an aiohttp application’s lifetime.

Parameters:

app (Any)

Return type:

AsyncIterator[None]

class aiohttp_tiny_mcp.postgres.PostgresSessionStore(storage)[source]

SessionStore with version-checked writes, timed by the database.

Parameters:

storage (PostgresStorage)

class aiohttp_tiny_mcp.postgres.PostgresHub(storage, *, look_again=None)[source]

A cursor-based Hub backed by one PostgreSQL table.

Parameters:

Many tenants

aiohttp_tiny_mcp.namespaces.scoped(key)[source]

Prefix a key with the percent-encoded namespace to prevent separator collisions.

Parameters:

key (str)

Return type:

str

Authentication

class aiohttp_tiny_mcp.auth.Authorization(verifier, resource, authorization_servers=(), scopes_supported=None, required_scopes=(), resource_name=None, documentation=None, bind_sessions=True, namespace_from_token=True)[source]

Configuration for an OAuth protected resource server.

Parameters:
property metadata_path: str

Return this resource’s RFC 9728 metadata path.

metadata()[source]

Build this resource’s RFC 9728 metadata document.

Return type:

dict[str, Any]

challenge(refusal)[source]

Build a Bearer challenge with a metadata URL.

Parameters:

refusal (Unauthorized)

Return type:

str

async principal(authorization)[source]

Verify an Authorization header and return its principal.

Parameters:

authorization (str | None)

Return type:

Principal

class aiohttp_tiny_mcp.auth.Principal(subject=None, client_id='', issuer=None, scopes=frozenset({}), expires_at=None, claims=<factory>)[source]

Identity and claims returned by a token verifier.

Parameters:
holds(wanted)[source]

Return required scopes absent from this principal.

Parameters:

wanted (Iterable[str])

Return type:

frozenset[str]

property identity: str

Return an issuer-qualified subject or client identifier.

class aiohttp_tiny_mcp.auth.TokenVerifier(*args, **kwargs)[source]

Verify a bearer token for this resource.

Clients

class aiohttp_tiny_mcp.Client(base_url, adapter, *, client_info=None, session=None, on_ask=None, on_notification=None, log_level=None)[source]
Parameters:

  • base_url (str)

  • adapter (Adapter)

  • client_info (Implementation | None)

  • session (aiohttp.ClientSession | None)

  • on_ask (Elicitor | None)

  • on_notification (Callable[[Mapping[str, Any]], Any] | None)

  • log_level (str | None)

async listen(*, resources=(), tools_changed=False, prompts_changed=False, resources_changed=False)

Yield changes until cancelled, via a request stream or legacy resource subscriptions.

Parameters:
Return type:

AsyncIterator[dict[str, Any]]

async set_log_level(level)

Request this severity and above, via session state or per-request metadata.

Parameters:

level (str)

Return type:

None

class aiohttp_tiny_mcp.stdio_client.StdioClient(reader, writer, adapter, *, client_info=None, on_ask=None, on_notification=None, log_level=None)[source]
Parameters:

  • reader (asyncio.StreamReader)

  • writer (ByteWriter)

  • adapter (Adapter)

  • client_info (Implementation | None)

  • on_ask (Elicitor | None)

  • on_notification (Callable[[Mapping[str, Any]], Any] | None)

  • log_level (str | None)

classmethod spawn(*cmd, adapter, client_info=None, on_ask=None)[source]

Launch cmd with protocol pipes; inherit stderr so logging cannot corrupt stdout.

Parameters:
Return type:

AsyncIterator[StdioClient]

class aiohttp_tiny_mcp.ClientError(code, message, data=None)[source]
Parameters:

  • code (int)

  • message (str)

  • data (Any)

Revisions

class aiohttp_tiny_mcp.adapter.Adapter[source]

One instance per revision, stateless and shared across requests.

decode(pre)[source]

Decode messages independently; one invalid item does not abort the batch.

Parameters:

pre (Preamble)

Return type:

Sequence[Call | DecodeFailure]

encode(call, registry, outcome)[source]

Encode a final response, identically for JSON, SSE, and stdio.

Parameters:
Return type:

Mapping[str, Any]

property carries_state: bool

Client-held state must be sealed on output and verified on return.

capabilities(registry)[source]

Capabilities shared by all supported revisions.

Parameters:

registry (RegistryProtocol)

Return type:

Mapping[str, Any]

can_ask: ClassVar[bool] = False

return questions, then retry with answers.

Type:

MRTR

can_push_ask: ClassVar[bool] = False

Push questions on the active stream; receive answers in a separate POST.

asks_in_arguments: ClassVar[bool] = False

exchange questions and answers through tool calls.

Type:

Pre-elicitation fallback

has_handshake: ClassVar[bool] = False

Negotiate once via initialize; retain the revision and capabilities in a session.

class aiohttp_tiny_mcp.protocol.selection.AdapterSet(adapters, *, fallback=None)[source]
Parameters:

The normalized core

class aiohttp_tiny_mcp.core.Operation(*values)[source]
DESCRIBE = 'describe'
HANDSHAKE_COMPLETE = 'handshake_complete'
PING = 'ping'
LIST_TOOLS = 'list_tools'
CALL_TOOL = 'call_tool'
LIST_RESOURCES = 'list_resources'
LIST_RESOURCE_TEMPLATES = 'list_resource_templates'
READ_RESOURCE = 'read_resource'
LIST_PROMPTS = 'list_prompts'
GET_PROMPT = 'get_prompt'
COMPLETE = 'complete'
LISTEN = 'listen'
SUBSCRIBE = 'subscribe'
UNSUBSCRIBE = 'unsubscribe'
SET_LOG_LEVEL = 'set_log_level'
class aiohttp_tiny_mcp.core.FailureKind(*values)[source]
PARSE = 'parse'
MALFORMED = 'malformed'
UNKNOWN_METHOD = 'unknown_method'
UNKNOWN_TARGET = 'unknown_target'
RESOURCE_NOT_FOUND = 'resource_not_found'
INVALID_PARAMS = 'invalid_params'
INVALID_ARGUMENTS = 'invalid_arguments'
HEADER_MISMATCH = 'header_mismatch'
UNSUPPORTED_VERSION = 'unsupported_version'
ORIGIN_REJECTED = 'origin_rejected'
INPUT_UNSUPPORTED = 'input_unsupported'
MISSING_REQUIRED_CAPABILITY = 'missing_required_capability'
INTERNAL = 'internal'
class aiohttp_tiny_mcp.core.Call(operation: 'Operation', id: 'str | int | None', target: 'str | None', arguments: 'Mapping[str, Any]', params: 'Params', client: 'ClientInfo', progress_token: 'str | int | None' = None, log_level: 'str | None' = None, answers: 'Mapping[str, Any]'=<factory>, actions: 'Mapping[str, AnswerAction]'=<factory>, state: 'Any' = None, raw: 'Mapping[str, Any]'=<factory>, is_notification: 'bool' = False)[source]
Parameters:
class aiohttp_tiny_mcp.core.Value(result: 'BaseModel')[source]
Parameters:

result (BaseModel)

class aiohttp_tiny_mcp.core.NeedsInput(requests: 'Mapping[str, InputRequest]', state: 'Any' = None)[source]
Parameters:
class aiohttp_tiny_mcp.core.Failure(kind: 'FailureKind', message: 'str', data: 'Any' = None)[source]
Parameters:
class aiohttp_tiny_mcp.core.ClientProfile(info, capabilities=<factory>, log_level=None)[source]

Client identity, capabilities, and desired log level.

Parameters: