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, page_size=None)[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:
page_size

Most entries per listing page. None, the default, answers with all of them.

property auth: Authentication | tuple[Authentication, ...] | None

Configured authentication policies. Iterables are consumed once.

property auth_policies: tuple[Authentication, ...]

Authentication alternatives in registration order.

extension(extension)[source]

Install an extension and its resources after checking all declarations.

Register dependency providers first. Duplicate identifiers, methods, resources, and attempts to replace base protocol methods are rejected.

Parameters:

extension (Extension)

Return type:

None

property broadcasts: frozenset[str]

Every notification method some installed extension may broadcast.

property topic_fields: dict[str, str | None]

Each declared broadcast with the params field that carries its topic, or None.

async broadcast(method, params=None, *, meta=None)[source]

Send one notification to every client listening in the current namespace.

method must be declared by an installed extension. A method declared with a topic field is a multicast: params must carry that field as a string, and a listener that named topics gets the event only when the value is one of them. The event goes through the hub, so a listener on another worker gets it too, and a legacy stream that reconnects with Last-Event-ID is replayed it. Returns the hub id the event was stored under.

Parameters:
Return type:

str

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]])

Extensions and skills

class aiohttp_tiny_mcp.Extension(name, *, capabilities=None, min_revision='2026-07-28', notifications=())[source]

Bundle methods, capability settings, and resources for registration.

Configure the extension before passing it to registry.extension(). Each registry takes a snapshot of the declarations.

Parameters:

  • name (str)

  • capabilities (Mapping[str, Any] | None)

  • min_revision (str)

  • notifications (Iterable[str] | Mapping[str, str | None])

method(name, fn=None)[source]

Register an async handler whose first argument is a Pydantic model.

Return a result mapping or a ResultModel subclass. Additional parameters use the registry’s dependency providers, including Exchange. Older revisions invoke the same handler through a namespaced resource. Parameters use a percent-encoded JSON query value. Reading a method resource can change state, depending on the handler.

Parameters:
resource(uri, fn=None, *, legacy_path=None, **kw)[source]

Bundle a resource with a legacy mcp-extensions://{name}/... URI.

By default, the legacy path is the URI after its scheme. Set legacy_path to choose a different relative path within the extension.

Parameters:
class aiohttp_tiny_mcp.extensions.skills.Skills(name, *, capabilities=None, min_revision='2026-07-28', notifications=())[source]

Serve immutable skill files and manifests without executing scripts.

Load files before starting the server. Recreate the extension and registry to publish changed files, so manifests and resource bytes remain consistent.

Parameters:

  • name (str)

  • capabilities (Mapping[str, Any] | None)

  • min_revision (str)

  • notifications (Iterable[str] | Mapping[str, str | None])

classmethod from_directory(directory, *, page_size=100)[source]

Load one skill directory or a tree containing skill directories.

Requires the skills extra. Symlinks and special files are rejected. Each skill is limited to 512 files and 16 MiB, including nested skills.

Parameters:
Return type:

Skills

See Extensions and skills for registration and legacy resource URIs. Loading directories requires aiohttp-tiny-mcp[skills].

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.

Metadata routes are included when a policy declares them. OAuth Bearer policies publish protected-resource metadata; Basic policies do not. Pass metadata=False when this application cannot serve those paths. 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, *, metadata=True)[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.

Policies that declare metadata include its routes. Use metadata=False when another endpoint serves it or when mounting under a subapplication. In a subapplication, add metadata_routes() to the root application.

Parameters:

  • sse_path (str | None)

  • message_path (str | None)

  • metadata (bool)

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.protocol.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.storage.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 server.sse. 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.storage.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.storage.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. touch() renews the TTL without a write to the data, so a session stays alive as long as its client keeps sending requests. 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 five 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 touch(session_id, *, ttl_seconds)[source]

Renew the TTL of a live record, keeping its data and version. Return False where it is missing or expired.

Parameters:

  • session_id (str)

  • ttl_seconds (int)

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.server.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.storage.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 five methods. Readers call subscribe; the four abstract methods are the storage it is built on.

async subscribe(topic, *, after=None, wait=1.0)[source]

Read after after, or after now. An unknown after raises ValueError.

Parameters:
Return type:

Subscription

abstractmethod async publish(topic, message)[source]

Append one message to topic and return the id it got.

Parameters:
Return type:

str

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 the events after cursor, in order. Wait up to timeout for one.

Raise ValueError for a cursor this hub did not issue.

Parameters:
Return type:

Sequence[Event]

abstractmethod async delete(topic)[source]

Delete a completed topic.

Parameters:

topic (str)

Return type:

None

class aiohttp_tiny_mcp.storage.hub.Event(id, message)[source]

One published message and its id. The id is the cursor after it.

Parameters:
class aiohttp_tiny_mcp.storage.hub.Subscription(hub, topic, cursor, *, wait, pending=())[source]

A reader of one topic that keeps its own place. Made by Hub.subscribe.

Parameters:
async poll(*, timeout=None)[source]

Events since the last call. Wait up to timeout, or wait when None.

Parameters:

timeout (float | None)

Return type:

Sequence[Event]

class aiohttp_tiny_mcp.MemoryHub(*, keep=1000)[source]

Single-process hub. A topic keeps its last keep events for replay.

Parameters:

keep (int)

aiohttp_tiny_mcp.storage.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.storage.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.storage.sqlite.SqliteSessionStore(storage)[source]

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

Parameters:

storage (SqliteStorage)

class aiohttp_tiny_mcp.storage.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.transport.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.transport.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.storage.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.storage.redis.RedisSessionStore(storage, *, prefix=None)[source]

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

Parameters:
class aiohttp_tiny_mcp.storage.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.storage.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.storage.postgres.PostgresSessionStore(storage)[source]

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

Parameters:

storage (PostgresStorage)

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

A cursor-based Hub backed by one PostgreSQL table.

Parameters:

Many tenants

See principal and storage namespaces for verified identity mapping, and multitenancy for storage behavior.

aiohttp_tiny_mcp.storage.namespaces.scoped(key)[source]

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

Parameters:

key (str)

Return type:

str

Authentication

Start with the authentication guide for configuration, subclass examples, multiple policies, client credentials, and permissions. The classes below provide the API reference.

class aiohttp_tiny_mcp.auth.Authentication(*, bind_sessions=True, namespace_from_token=True, required_scopes=())[source]

Subclass this policy and pass an instance to Registry(auth=…).

Implement authenticate and challenge. Return a verified Principal or None for invalid credentials. Raise Unauthorized for a specific refusal. HTTP endpoints enforce expiry, required scopes, session ownership, and namespaces after authentication. Policies publish no metadata by default.

Parameters:
abstractmethod async authenticate(request)[source]

Verify this HTTP request. The full request is available to the policy.

Parameters:

request (Request)

Return type:

Principal | None

abstractmethod challenge(refusal)[source]

Return the WWW-Authenticate value for this policy.

Parameters:

refusal (Unauthorized)

Return type:

str

check(principal)[source]

Reject invalid identities and missing server-wide scopes.

Parameters:

principal (Principal | None)

Return type:

Principal

property metadata_path: str | None

Return an optional metadata route, relative to the site root.

metadata()[source]

Return metadata when the policy declares a metadata path.

Return type:

dict[str, Any]

class aiohttp_tiny_mcp.auth.BasicAuth(*, realm='mcp', bind_sessions=True, namespace_from_token=True, required_scopes=())[source]

Subclass verify to authenticate Basic credentials against your user store.

Credentials use standard Base64 and UTF-8. Query credentials are not read. Use StaticBasicAuth for configured accounts with individual scopes.

Parameters:
abstractmethod async verify(username, password)[source]

Return the verified identity, or None when credentials do not match.

Parameters:

  • username (str)

  • password (str)

Return type:

Principal | None

async authenticate(request)[source]

Verify this HTTP request. The full request is available to the policy.

Parameters:

request (Request)

Return type:

Principal | None

challenge(refusal)[source]

Return the WWW-Authenticate value for this policy.

Parameters:

refusal (Unauthorized)

Return type:

str

class aiohttp_tiny_mcp.auth.StaticBasicAuth(*accounts, realm='mcp', bind_sessions=True, namespace_from_token=True, required_scopes=())[source]

Basic authentication for configured accounts with individual scopes.

Pass each account as (username, password, scopes), or (username, password) for an account without scopes. Usernames must be unique. Use HTTPS in production.

Parameters:
async verify(username, password)[source]

Return the verified identity, or None when credentials do not match.

Parameters:

  • username (str)

  • password (str)

Return type:

Principal | None

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

async authenticate(request)[source]

Verify the request’s Bearer header using the configured token verifier.

Parameters:

request (Request)

Return type:

Principal

class aiohttp_tiny_mcp.auth.Principal(subject=None, client_id='', issuer=None, scopes=frozenset({}), expires_at=None, claims=<factory>, namespace=None)[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.

class aiohttp_tiny_mcp.auth.StaticVerifier(mapping)[source]

Verify configured opaque tokens for development and tests.

Compare fixed-size digests for every configured token, including after a match. Authorization enforces the returned principal’s expiry and scopes.

Parameters:

mapping (Mapping[str, Principal])

aiohttp_tiny_mcp.auth.principal_from_claims(claims)[source]

Map verified claims to a principal. This function does not verify tokens.

Empty claims are valid. Malformed identity, scope, or expiry values raise ValueError. Application-specific namespaces require a custom mapper.

Parameters:

claims (Mapping[str, Any])

Return type:

Principal

class aiohttp_tiny_mcp.auth.jwt.JWTVerifier(key, algorithms, issuer='', audience='', principal=<function principal_from_claims>)[source]

Verify JWTs, then map verified claims with the synchronous principal hook.

Use HMACJWTVerifier or PublicKeyJWTVerifier to select algorithms from trusted key material. The mapper may raise ValueError to reject verified claims.

Parameters:
principal()

Map verified claims to a principal. This function does not verify tokens.

Empty claims are valid. Malformed identity, scope, or expiry values raise ValueError. Application-specific namespaces require a custom mapper.

Parameters:

claims (Mapping[str, Any])

Return type:

Principal

async verify(credentials)[source]

Check the signature, required expiry, and configured issuer and audience.

Parameters:

credentials (str)

Return type:

Principal | None

class aiohttp_tiny_mcp.auth.jwt.HMACJWTVerifier(pre_shared_key, *, issuer='', audience='', principal=<function principal_from_claims>)[source]

Verify HS256 JWTs with a configured pre-shared key.

Parameters:
class aiohttp_tiny_mcp.auth.jwt.PublicKeyJWTVerifier(public_key, *, issuer='', audience='', principal=<function principal_from_claims>)[source]

Verify RSA, EC, or EdDSA JWTs with a configured public key.

Parameters:
class aiohttp_tiny_mcp.oauth.OAuthServer(issuer, provider, *, resource, clients, scopes=(), identity=None, store=None, tokens=None, access_token_ttl=3600)[source]

Exchange upstream identity for short-lived MCP tokens.

This implementation supports authorization_code, PKCE S256, and explicitly registered public clients. It does not register clients dynamically or issue refresh tokens. Supply a shared SessionStore for multiple workers.

The tokens object owns issuance and verification; OpaqueTokens is the default. Provider credentials stay in private flow storage until code redemption or expiry. Issue runs after PKCE validation and atomic code redemption.

Parameters:
resource(*, required_scopes=())[source]

Return the Registry authentication policy bound to this resource and issuer.

Parameters:

required_scopes (Sequence[str])

Return type:

Authorization

routes()[source]

Mount at the origin root. Paths come from the configured issuer.

Return type:

list[RouteDef]

setup(app)[source]

Add OAuth routes and manage the upstream HTTP session with the application.

Parameters:

app (Application)

Return type:

Application

static is_public(request)[source]

Identify server routes for application authentication middleware.

These handlers enforce their own OAuth checks and must be reachable before the caller has an MCP token.

Parameters:

request (Request)

Return type:

bool

async revoke(token)[source]

Revoke an access token. Custom token objects can supply async revoke(token).

Parameters:

token (str)

Return type:

None

class aiohttp_tiny_mcp.oauth.OAuthClient(client_id, redirect_uris, client_name='MCP client', require_consent=False)[source]

Register a trusted public client with exact callback URLs.

Set require_consent to show a local confirmation before provider sign-in. Client secrets are not used.

Parameters:
class aiohttp_tiny_mcp.oauth.OpaqueTokens(*, issuer, resource, store=None)[source]

Store principals by token digest; discard provider credentials after issue.

Share the store, issuer and resource across instances. Each token has its principal’s expiry and can be revoked independently.

Parameters:
async issue(principal, upstream)[source]

Issue an opaque token. Provider credentials are not retained.

Parameters:
Return type:

str

async verify(token)[source]

Return the stored principal, or None for invalid or expired tokens.

Parameters:

token (str)

Return type:

Principal | None

async revoke(token)[source]

Delete one token without affecting other grants.

Parameters:

token (str)

Return type:

None

class aiohttp_tiny_mcp.oauth.AbstractCipher[source]

Encode dictionaries to authenticated bytes; reject invalid blobs with ValueError.

class aiohttp_tiny_mcp.oauth.KECCAKCipher(key, compression_level=9)[source]

Compress JSON with zlib, then encrypt with SHAKE-256 XOR and HMAC-SHA256.

Format: nonce (16 bytes) || ciphertext || MAC (32 bytes). Verify MAC first. Anyone with the key can read and create tokens.

Level 9 favors smaller tokens. On M1 Pro/Python 3.14, encode/decode medians were ~35/18 us for 2 KiB and ~283/105 us for 16 KiB of random URL-safe JSON. At 16 KiB, level 9 added ~20 us per encode versus level 1; decode was similar.

Background on SHAKE/Keccak authenticated encryption: https://eprint.iacr.org/2024/1618 (Shaking up authenticated encryption) https://keccak.team/files/SpongeDuplex.pdf (SpongeWrap) https://github.com/XKCP/XKCP (reference impl - ShakingUpAE)

Parameters:
class aiohttp_tiny_mcp.oauth.EncryptedTokens(cipher, *, issuer, resource)[source]

OAuth payloads encrypted by an application-supplied cipher, in URL-safe base64.

Pass this object as OAuthServer(tokens=…). Share the cipher configuration and the same issuer/resource across instances. No token database is needed. Individual revocation is unsupported; replacing the key invalidates all tokens. Provider credentials remain separate from Principal.

Parameters:
async issue(principal, upstream)[source]

Encode the principal and complete provider response. A future expiry is required.

Parameters:
Return type:

str

decode(token)[source]

Return the verified payload, including upstream credentials, or raise ValueError.

The cipher checks integrity; this method checks issuer, resource and expiry. Use the result only in server code; do not return it to an MCP client or log it.

Parameters:

token (str)

Return type:

dict[str, Any]

async verify(token)[source]

Return only the principal. Invalid or expired tokens return None.

Parameters:

token (str)

Return type:

Principal | None

class aiohttp_tiny_mcp.oauth.OAuthProvider(name, authorize_url, token_url, client_id, scopes, profile, client_secret='', extra_authorize_params=<factory>)[source]

Configure an OAuth authorization-code provider and its identity lookup.

Parameters:
class aiohttp_tiny_mcp.oauth.Identity(sub, provider, claims=<factory>)[source]

A stable provider identity. Claims must not contain credentials.

Parameters:
aiohttp_tiny_mcp.oauth.github.GitHub(client_id, client_secret, *, scopes=('read:user',))[source]

GitHub.com OAuth App sign-in. Register the server’s callback URL in GitHub.

Parameters:
Return type:

OAuthProvider

Clients

class aiohttp_tiny_mcp.Client(base_url, adapter, *, client_info=None, session=None, auth=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)

  • auth (OAuth | 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, methods=())

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

methods names the extension broadcasts to relay: a method name for every event of it, or MethodFilter(method, topics) (a mapping of the same shape will do) for the topics of a multicast. A legacy stream carries every broadcast the server declares, so there it is not sent.

Parameters:
Return type:

AsyncIterator[dict[str, Any]]

async request_method(method, params=None, *, name=None)

Call a named method, including extensions, with protocol metadata and headers.

Check server discovery for the extension before using its methods. Results are returned as wire dictionaries; RPC errors raise ClientError.

Parameters:
Return type:

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.client.stdio.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, limit=1048576)[source]

Launch cmd with protocol pipes and inherit stderr.

limit bounds response lines in bytes and defaults to 1 MiB.

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.protocol.adapter.Adapter[source]

One instance per revision, stateless and shared across requests.

decode(pre, registry=None)[source]

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

Parameters:

  • pre (Preamble)

  • registry (RegistryProtocol | None)

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]

describe_tool(spec)[source]

Return how this revision shows spec, or None where it hides it.

The projection is cached on the spec. A tool does not change after registration, so each revision simplifies its schemas once instead of once per tools/list and once per tools/call.

Parameters:

spec (ToolSpec)

Return type:

ToolDef | None

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.protocol.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'
EXTENSION = 'extension'
class aiohttp_tiny_mcp.protocol.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'
SESSION_NOT_FOUND = 'session_not_found'
INPUT_UNSUPPORTED = 'input_unsupported'
MISSING_REQUIRED_CAPABILITY = 'missing_required_capability'
INTERNAL = 'internal'
class aiohttp_tiny_mcp.protocol.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.protocol.core.Value(result: 'BaseModel')[source]
Parameters:

result (BaseModel)

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

Client identity, capabilities, and desired log level.

Parameters:

Client OAuth

See Client OAuth for registration, callback handling, credential storage, and refresh behavior.

class aiohttp_tiny_mcp.OAuth(endpoint, *, redirect_uri, issuer=None, client_id=None, client_secret=None, token_endpoint_auth_method='none', metadata_url=None, registration=None, scopes=None, load=None, save=None, allow_http=False, check_url=None, timeout=30, login_timeout=300)[source]

OAuth authorization for one MCP endpoint and one application account.

Use as an async context manager, then pass it to Client(auth=...). login explicitly starts user interaction; ordinary requests only refresh. load and save are application callbacks scoped to this account and connection. Shared credentials require coordination outside this instance.

HTTPS is required. allow_http permits loopback HTTP for development. check_url can enforce an application’s network policy before each fetch and before returning an authorization URL to the application.

Parameters:

  • endpoint (str)

  • redirect_uri (str)

  • issuer (str | None)

  • client_id (str | None)

  • client_secret (str | None)

  • token_endpoint_auth_method (str)

  • metadata_url (str | None)

  • registration (Mapping[str, Any] | None)

  • scopes (Sequence[str] | None)

  • load (LoadCredentials | None)

  • save (SaveCredentials | None)

  • allow_http (bool)

  • check_url (CheckURL | None)

  • timeout (float)

  • login_timeout (float)

async begin(*, scopes=None)[source]

Start a login, replacing any pending login in this context.

Parameters:

scopes (Sequence[str] | None)

Return type:

AuthorizationAttempt

cancel(attempt)[source]

Discard a pending login without changing existing credentials.

Parameters:

attempt (AuthorizationAttempt)

Return type:

None

async login(*, authorize, scopes=None)[source]

Run an explicit login through the application’s browser/callback handler.

Parameters:
Return type:

None

async complete(attempt, callback_url)[source]

Validate and consume a callback before sending its code to the token endpoint.

Parameters:

  • attempt (AuthorizationAttempt)

  • callback_url (str)

Return type:

None

async token(*, rejected=None)[source]

Return a usable token, refreshing once if needed. Never start a login.

Parameters:

rejected (str | None)

Return type:

str

class aiohttp_tiny_mcp.OAuthCallback(auth, *, success_html='<!doctype html>\n<html lang="en"><meta charset="utf-8"><title>Authorization complete</title>\n<h1>Authorization complete</h1><p>You can close this window.</p></html>')[source]

Mount a local or remote callback for one OAuth account context.

The application must authenticate the user before calling begin and keep each user’s context separate. The external HTTPS auth.redirect_uri can differ from the listener address behind a trusted reverse proxy.

Parameters:
routes()[source]

Return the GET route at the path of the configured redirect URI.

Return type:

list[RouteDef]

setup(app)[source]

Mount the callback on an existing application.

Parameters:

app (Application)

Return type:

None

async begin(*, scopes=None)[source]

Prepare the URL to which the application redirects its signed-in user.

Parameters:

scopes (Sequence[str] | None)

Return type:

AuthorizationAttempt

async wait()[source]

Wait for token acquisition and persistence, or raise their error.

Return type:

None

cancel()[source]

Cancel a pending callback without changing existing credentials.

Return type:

None

async handle(request)[source]

Exchange the callback code before sending the result page.

Parameters:

request (Request)

Return type:

Response

class aiohttp_tiny_mcp.LoopbackCallback(*, port=0, timeout=300, open_browser=None, success_html='<!doctype html>\n<html lang="en"><meta charset="utf-8"><title>Authorization complete</title>\n<h1>Authorization complete</h1><p>You can close this window.</p></html>')[source]

Run the aiohttp callback on 127.0.0.1 and optionally open a browser.

Enter this context before constructing OAuth, then call login(auth). Register redirect_uri with the authorization server, or use DCR. A fixed port can be supplied for pre-registered clients. By default, the operating system selects a free port. success_html is trusted application HTML, served unchanged only after the token exchange and save.

Parameters:

  • port (int)

  • timeout (float)

  • open_browser (Callable[[str], Awaitable[None]] | None)

  • success_html (str)

async login(auth, *, scopes=None)[source]

Open the browser and wait until the token exchange and save finish.

Parameters:
Return type:

None

class aiohttp_tiny_mcp.AuthorizationRequired(scopes=())[source]

The application must obtain authorization before it can continue.

Parameters:

scopes (Iterable[str])

class aiohttp_tiny_mcp.RegistrationRequired[source]

The application must supply client registration information.

class aiohttp_tiny_mcp.OAuthError[source]

An OAuth failure. Messages exclude response bodies and credentials.