MCP Server API Reference¶

Production framework for building MCP-compliant tool servers. Every middleware, auth provider, guard, transform, and utility is documented here.

Core¶

MCPServer¶

`promptise.mcp.server.MCPServer` ¶

Production-grade MCP server with decorator-based tool registration.

Parameters:

Name	Type	Description	Default
`name`	`str`	Server name advertised to MCP clients.	`'promptise-server'`
`version`	`str`	Server version string.	`'0.1.0'`
`instructions`	`str \| None`	Optional instructions sent to clients on initialisation.	`None`
`require_auth`	`bool`	Force `auth=True` on every registered tool, resource and prompt.	`False`
`require_tenant`	`bool`	Make tenant identity a server-wide invariant: every tool, resource and prompt authenticates and carries a `RequireTenant` guard, so a client whose token lacks the tenant claim is denied on every call. Implies `require_auth`.	`False`
`hide_unauthorized_tools`	`bool`	Filter `tools/list`, `resources/list`, `resources/templates/list`, `prompts/list` (and the `docs://manifest` resource) per request, so each caller sees only what its identity may use: the request is authenticated with the server's `AuthMiddleware` and every definition's guards are evaluated against it. Off by default, in which case everything is listed to everyone and guards apply when a tool is called, a resource read or a prompt requested. Guards that depend on state set by other middleware (not `AuthMiddleware` or its `on_authenticate` hook) cannot be evaluated at list time and hide their definition.	`False`
`cancel_grace_period`	`float`	When a client cancels a call to a tool that takes a `CancellationToken`, the token is set and the handler gets this many seconds to stop on its own before its task is cancelled. The client is answered immediately either way.	`5.0`

`middleware` `property` ¶

Decorator to register a middleware function.

Example::

@server.middleware
async def log_calls(ctx, call_next):
    result = await call_next(ctx)
    return result

`add_middleware(middleware)` ¶

Add a middleware to the processing chain.

The chain runs for tool calls, resource reads and prompt requests; ctx.request_type says which.

`asgi_app(transport='http', *, allowed_hosts=None, allowed_origins=None, cors=None, stateless=False)` ¶

Return the server as an ASGI application, for uvicorn, gunicorn or hypercorn.

Serves the same routes as :meth:run (/mcp — or /sse and /messages/ — plus /health and /health/ready) with the same auth gate, token endpoint and CORS. The app's lifespan runs the server's startup and shutdown hooks and the MCP session manager, so the ASGI server must run lifespan events (uvicorn's default --lifespan auto does).

Example::

# app.py
app = server.asgi_app(allowed_hosts=["mcp.example.com"])

# uvicorn app:app --host 0.0.0.0 --port 8080
# gunicorn app:app -k uvicorn.workers.UvicornWorker -w 4

The bind address belongs to the ASGI server, so Host validation cannot follow it: the app always accepts the loopback names, plus every value in allowed_hosts. Name the public host your clients or proxy send, or every request through it is refused with 421.

Several workers or replicas. Each process keeps its own MCP sessions. A request whose mcp-session-id belongs to another process is answered 404; Promptise clients then open a new session, so routing that is not sticky turns into a new session on almost every call. Route each mcp-session-id to one process (sticky sessions), or pass stateless=True.

Parameters:

Name	Type	Description	Default
`transport`	`str`	`"http"` (Streamable HTTP) or `"sse"`.	`'http'`
`allowed_hosts`	`list[str] \| None`	`Host` header values to accept in addition to the loopback names, e.g. `["mcp.example.com"]`.	`None`
`allowed_origins`	`list[str] \| None`	`Origin` header values to accept for browser clients, in addition to the loopback origins.	`None`
`cors`	`Any`	Optional `CORSConfig`.	`None`
`stateless`	`bool`	Streamable HTTP only. Serve each request without a session, so any worker or replica can answer it. Stateless servers cannot send requests back to the client: MCP elicitation and sampling (including elicitation approval gates) and per-session state are unavailable.	`False`

Raises:

Type	Description
`ValueError`	An unknown or stdio transport, `stateless` with SSE, or an empty `allowed_hosts` list.

`enable_token_endpoint(jwt_auth, clients, *, path='/auth/token', default_expires_in=86400)` ¶

Enable a built-in token endpoint for development and testing.

This adds an /auth/token HTTP endpoint that issues JWT tokens using the OAuth2 client_credentials flow. Clients send {"client_id": "...", "client_secret": "..."} and receive a signed JWT back.

For production, use a proper Identity Provider (Auth0, Keycloak, Okta, etc.) instead.

Parameters:

Name	Type	Description	Default
`jwt_auth`	`Any`	The `JWTAuth` instance used to sign tokens (same one passed to `AuthMiddleware`).	required
`clients`	`dict[str, dict[str, Any]]`	Mapping of `client_id` → config dict. Each config must include `"secret"` and may include `"roles"` (list), `"expires_in"` (int, seconds), and `"claims"` (dict of extra JWT claims).	required
`path`	`str`	HTTP path for the endpoint (default `/auth/token`).	`'/auth/token'`
`default_expires_in`	`int`	Default token lifetime in seconds.	`86400`

jwt_auth = JWTAuth(secret="my-secret")
server.add_middleware(AuthMiddleware(jwt_auth))

server.enable_token_endpoint(
    jwt_auth=jwt_auth,
    clients={
        "agent-admin":  {"secret": "s3cret",  "roles": ["admin"]},
        "agent-viewer": {"secret": "v1ewer",  "roles": ["viewer"]},
    },
)

`exception_handler(exc_type)` ¶

Register a custom exception handler.

The handler receives (ctx, exc) and should return an MCPError instance.

Example::

@server.exception_handler(DatabaseError)
async def handle_db_error(ctx, exc):
    return ToolError("DB unavailable", code="DB_ERROR", retryable=True)

`include_prompts(*sources)` ¶

Register Promptise prompts as MCP prompt endpoints.

Accepts any combination of:

:class:~promptise.prompts.registry.PromptRegistry — exposes the latest version of every registered prompt.
:class:~promptise.prompts.core.Prompt — exposes a single prompt.
:class:~promptise.prompts.suite.PromptSuite — exposes all prompts in the suite.

Each prompt is converted to an MCP PromptDef with arguments extracted from the prompt's function signature. Each argument is described by the YAML file's arguments, an Annotated[..., Field(description=...)] hint, or else by the placeholder it fills, its type and its default. The MCP handler converts the string-valued arguments to the parameters' types and calls render_async(**arguments) to produce fully rendered prompt text with context providers, strategy, perspective, and constraints applied. Templates written as docstrings are dedented, so the code's indentation is not sent to the model.

Parameters:

Name	Type	Description	Default
`*sources`	`Any`	Prompt registries, individual prompts, or suites.	`()`

from promptise.prompts.registry import registry
server.include_prompts(registry)

`include_router(router, *, prefix='', tags=None)` ¶

Merge a router's registrations into this server.

Parameters:

Name	Type	Description	Default
`router`	`Any`	The `MCPRouter` to include.	required
`prefix`	`str`	Additional prefix (combined with router's own prefix).	`''`
`tags`	`list[str] \| None`	Additional tags (combined with router's own tags).	`None`

`notify_prompts_changed()` `async` ¶

Send notifications/prompts/list_changed to every connected client.

Sent automatically when a prompt is registered while the server is serving.

Returns:

Type	Description
`int`	How many sessions the notification reached.

`notify_resources_changed()` `async` ¶

Send notifications/resources/list_changed to every connected client.

Sent automatically when a resource or resource template is registered while the server is serving.

Returns:

Type	Description
`int`	How many sessions the notification reached.

`notify_tools_changed()` `async` ¶

Send notifications/tools/list_changed to every connected client.

Registering a tool while the server is serving sends this automatically; call it yourself after changing what list_tools returns some other way.

Returns:

Type	Description
`int`	How many sessions the notification reached.

`on_shutdown(func)` ¶

Register a shutdown hook.

`on_startup(func)` ¶

Register a startup hook.

`prompt(name=None, *, description=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

Register a function as an MCP prompt.

The handler may return a str (one user message), a PromptMessage, a list mixing those (or {"role", "content"} dicts), or a full GetPromptResult. MCP sends arguments as strings; they are coerced to the handler's type hints. Requests run through the server's middleware chain, like tool calls.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Prompt name (defaults to function name).	`None`
`description`	`str \| None`	Description (defaults to docstring).	`None`
`tags`	`list[str] \| None`	Optional tags for categorisation.	`None`
`auth`	`bool`	Require authentication to get this prompt.	`False`
`roles`	`list[str] \| None`	Required roles shorthand (implies `auth=True`).	`None`
`guards`	`list[Any] \| None`	Access control guards, checked before the handler.	`None`
`rate_limit`	`str \| None`	Rate limit string, e.g. `"100/min"`.	`None`
`timeout`	`float \| None`	Per-request timeout in seconds.	`None`

`resource(uri, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

Register a function as an MCP resource.

Reads run through the server's middleware chain, like tool calls.

Parameters:

Name	Type	Description	Default
`uri`	`str`	Static resource URI (e.g. `"config://app"`).	required
`name`	`str \| None`	Resource name (defaults to function name).	`None`
`description`	`str \| None`	Description (defaults to docstring).	`None`
`mime_type`	`str \| None`	MIME type of the resource content. Defaults to `application/json` for a handler annotated `-> dict` / `-> list` (or a Pydantic model), `application/octet-stream` for `-> bytes`, and `text/plain` otherwise.	`None`
`tags`	`list[str] \| None`	Optional tags for categorisation.	`None`
`auth`	`bool`	Require authentication to read this resource.	`False`
`roles`	`list[str] \| None`	Required roles shorthand (adds a `HasRole` guard and implies `auth=True`).	`None`
`guards`	`list[Any] \| None`	Access control guards, checked before the handler.	`None`
`rate_limit`	`str \| None`	Rate limit string, e.g. `"100/min"`.	`None`
`timeout`	`float \| None`	Per-read timeout in seconds (enforced by `TimeoutMiddleware`).	`None`

`resource_template(uri_template, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

Register a function as an MCP resource template.

Each {param} placeholder matches one path segment and is passed to the handler parameter of the same name, coerced to its type hint ({page} → int). {param*} (or {+param}) matches the rest of the URI including /, for hierarchical ids such as docs://pages/{path*}.

Parameters:

Name	Type	Description	Default
`uri_template`	`str`	URI template with `{param}` placeholders.	required
`name`	`str \| None`	Resource name.	`None`
`description`	`str \| None`	Description.	`None`
`mime_type`	`str \| None`	MIME type (see :meth:`resource`).	`None`
`tags`	`list[str] \| None`	Optional tags for categorisation.	`None`
`auth`	`bool`	Require authentication to read these resources.	`False`
`roles`	`list[str] \| None`	Required roles shorthand (implies `auth=True`).	`None`
`guards`	`list[Any] \| None`	Access control guards, checked before the handler.	`None`
`rate_limit`	`str \| None`	Rate limit string, e.g. `"100/min"`.	`None`
`timeout`	`float \| None`	Per-read timeout in seconds.	`None`

`run(transport='stdio', *, host='127.0.0.1', port=8080, dashboard=False, cors=None, allowed_hosts=None, allowed_origins=None)` ¶

Start the server (blocking).

Over HTTP/SSE the server also answers GET /health (liveness) and GET /health/ready (readiness from a registered :class:HealthCheck, 503 when a required check fails) for container and Kubernetes probes.

Parameters:

Name	Type	Description	Default
`transport`	`str`	`"stdio"`, `"http"`, or `"sse"`.	`'stdio'`
`host`	`str`	Bind host for HTTP/SSE transports. Defaults to `"127.0.0.1"` (reachable from this machine only, with Host/Origin validation on). Pass `"0.0.0.0"` to listen on every interface — in a container, for example — and name the public host in `allowed_hosts`; without it a non-loopback bind does not validate `Host` and logs a warning at startup. A non-loopback bind without `AuthMiddleware` also logs a warning: every tool is then callable by anyone who can reach the address.	`'127.0.0.1'`
`port`	`int`	Bind port for HTTP/SSE transports.	`8080`
`dashboard`	`bool`	Enable live terminal monitoring dashboard.	`False`
`cors`	`Any`	Optional `CORSConfig` for HTTP/SSE transports.	`None`
`allowed_hosts`	`list[str] \| None`	`Host` header values the HTTP/SSE transports accept (`"api.example.com"` or `"api.example.com:*"`). A loopback bind validates `Host` and `Origin` against the loopback names by default (DNS rebinding protection) and this list adds to them — name the public host a reverse proxy forwards. A non-loopback bind validates only when this is given, and then exactly these values.	`None`
`allowed_origins`	`list[str] \| None`	`Origin` header values to accept, for browser clients served from another origin. Requests without an `Origin` header always pass this check. Requires `allowed_hosts` on a non-loopback bind.	`None`

`run_async(transport='stdio', *, host='127.0.0.1', port=8080, dashboard=False, cors=None, allowed_hosts=None, allowed_origins=None)` `async` ¶

Start the server (async).

Parameters:

Name	Type	Description	Default
`transport`	`str`	`"stdio"`, `"http"`, or `"sse"`.	`'stdio'`
`host`	`str`	Bind host for HTTP/SSE transports.	`'127.0.0.1'`
`port`	`int`	Bind port for HTTP/SSE transports.	`8080`
`dashboard`	`bool`	Enable live terminal monitoring dashboard.	`False`
`cors`	`Any`	Optional `CORSConfig` for HTTP/SSE transports.	`None`
`allowed_hosts`	`list[str] \| None`	`Host` header values the HTTP/SSE transports accept; see :meth:`run`.	`None`
`allowed_origins`	`list[str] \| None`	`Origin` header values to accept; see :meth:`run`.	`None`

Raises:

Type	Description
`ValueError`	`allowed_origins` without `allowed_hosts` on a non-loopback bind, or an empty `allowed_hosts` list.

`tool(name=None, *, description=None, tags=None, auth=False, rate_limit=None, timeout=None, guards=None, roles=None, title=None, read_only_hint=None, destructive_hint=None, idempotent_hint=None, open_world_hint=None, max_concurrent=None, requires_approval=False, cache=True)` ¶

Register a function as an MCP tool.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Tool name (defaults to function name).	`None`
`description`	`str \| None`	Tool description (defaults to docstring first line).	`None`
`tags`	`list[str] \| None`	Optional tags for categorisation.	`None`
`auth`	`bool`	Require authentication for this tool.	`False`
`rate_limit`	`str \| None`	Rate limit string, e.g. `"100/min"`.	`None`
`timeout`	`float \| None`	Per-call timeout in seconds.	`None`
`guards`	`list[Any] \| None`	Access control guards (checked before handler). A built-in identity guard (`RequireAuth`, `HasRole`, `HasScope`, `RequireClientId`, tenant guards) implies `auth=True`: it can only pass once the caller is authenticated.	`None`
`roles`	`list[str] \| None`	Required roles shorthand (creates `HasRole` guard; implies `auth=True`).	`None`
`title`	`str \| None`	Human-readable title (MCP annotation hint).	`None`
`read_only_hint`	`bool \| None`	Tool does not modify state (MCP annotation hint).	`None`
`destructive_hint`	`bool \| None`	Tool may perform destructive operations (MCP annotation hint).	`None`
`idempotent_hint`	`bool \| None`	Repeated calls with same args produce same result (MCP annotation hint).	`None`
`open_world_hint`	`bool \| None`	Tool may interact with external systems (MCP annotation hint).	`None`
`max_concurrent`	`int \| None`	Maximum concurrent calls for this tool. When reached, additional calls receive a retryable error.	`None`
`requires_approval`	`bool`	Require a human approval decision before every call to this tool, enforced server-side for any MCP client. The server must install an `ApprovalGateMiddleware` — building a server with an ungated `requires_approval` tool raises at build time rather than silently not enforcing it.	`False`
`cache`	`bool`	`False` keeps a server-wide `CacheMiddleware` from caching this tool's results (use it for tools that change data or must always be fresh). Tools annotated `destructive_hint=True` are never cached either.	`True`

@server.tool()
async def search(query: str, limit: int = 10) -> list[dict]:
    """Search records."""
    return await db.search(query, limit)

MCPRouter¶

`promptise.mcp.server.MCPRouter` ¶

Modular grouping for tools, resources, and prompts.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Prepended to all tool names (e.g. `"db"` → `"db_search"`).	`''`
`tags`	`list[str] \| None`	Default tags merged with per-tool tags.	`None`
`auth`	`bool \| None`	If set, overrides the per-tool / per-resource / per-prompt auth flag.	`None`
`middleware`	`list[Any] \| None`	Router-level middleware (runs after server middleware) for the router's tools, resources and prompts.	`None`
`guards`	`list[Any] \| None`	Router-level guards applied to all tools, resources and prompts.	`None`

`include_prompts(*sources)` ¶

Register Promptise prompts as MCP prompt endpoints on this router.

Accepts :class:~promptise.prompts.registry.PromptRegistry, :class:~promptise.prompts.core.Prompt, or :class:~promptise.prompts.suite.PromptSuite instances.

Parameters:

Name	Type	Description	Default
`*sources`	`Any`	Prompt registries, individual prompts, or suites.	`()`

`include_router(router, *, prefix='', tags=None)` ¶

Nest a sub-router.

`prompt(name=None, *, description=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

Register a prompt (same signature as MCPServer.prompt()).

`resource(uri, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

Register a resource (same signature as MCPServer.resource()).

`resource_template(uri_template, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

Register a resource template (same signature as MCPServer.resource_template()).

`tool(name=None, *, description=None, tags=None, auth=False, rate_limit=None, timeout=None, guards=None, roles=None, title=None, read_only_hint=None, destructive_hint=None, idempotent_hint=None, open_world_hint=None, max_concurrent=None, requires_approval=False, cache=True)` ¶

Register a tool (same signature as MCPServer.tool()).

ServerSettings¶

`promptise.mcp.server.ServerSettings` ¶

Bases: BaseSettings

Base server settings with sensible defaults.

Reads from environment variables with PROMPTISE_ prefix. Subclass to add your own fields.

Attributes:

Name	Type	Description
`server_name`	`str`	MCP server name.
`log_level`	`str`	Logging level (`DEBUG`, `INFO`, `WARNING`, etc.).
`timeout_default`	`float`	Default tool timeout in seconds.

Depends¶

`promptise.mcp.server.Depends(dependency, *, use_cache=True)` ¶

Mark a parameter for dependency injection.

Parameters:

Name	Type	Description	Default
`dependency`	`Any`	A callable (sync or async) or generator that provides the dependency value.	required
`use_cache`	`bool`	If `True`, cache the resolved value per-request so the same dependency is only created once.	`True`

Returns:

Type	Description
`Any`	A marker that the framework detects at call time.

mount (server composition)¶

`promptise.mcp.server.mount(parent, child, *, prefix='', tags=None)` ¶

Mount a child server's tools, resources, and prompts into a parent.

All tool names are prefixed with {prefix}_ if a prefix is given. Tags from the parent are merged with child tags.

Parameters:

Name	Type	Description	Default
`parent`	`Any`	The parent `MCPServer` to mount into.	required
`child`	`Any`	The child `MCPServer` whose registrations will be copied.	required
`prefix`	`str`	Namespace prefix for tool names.	`''`
`tags`	`list[str] \| None`	Additional tags applied to all mounted tools.	`None`

Returns:

Type	Description
`int`	Number of tools mounted.

hot_reload¶

`promptise.mcp.server.hot_reload(server=None, *, transport='http', host='127.0.0.1', port=8080, watch_dirs=None, poll_interval=1.0, dashboard=False, allowed_hosts=None, allowed_origins=None)` ¶

Run an MCP server with automatic restart on file changes.

Watches *.py files in the specified directories (defaults to the current working directory). When a change is detected, the server process is restarted.

This function blocks and runs in the parent process while the server runs as a subprocess.

Parameters:

Name	Type	Description	Default
`server`	`Any`	Not used directly — the server is started via re-running the current script.	`None`
`transport`	`str`	Transport type.	`'http'`
`host`	`str`	Bind host.	`'127.0.0.1'`
`port`	`int`	Bind port.	`8080`
`watch_dirs`	`list[str] \| None`	Directories to watch (defaults to `["."]`).	`None`
`poll_interval`	`float`	Seconds between file change checks.	`1.0`
`dashboard`	`bool`	Enable dashboard in the child process.	`False`
`allowed_hosts`	`list[str] \| None`	`Host` header allow-list for the child server's HTTP/SSE transport (see `MCPServer.run`).	`None`
`allowed_origins`	`list[str] \| None`	`Origin` header allow-list for the child server's HTTP/SSE transport (see `MCPServer.run`).	`None`

if __name__ == "__main__":
    hot_reload(transport="http", port=8080)

Context¶

ClientContext¶

`promptise.mcp.server.ClientContext` `dataclass` ¶

Structured information about the authenticated client.

Populated by :class:AuthMiddleware after successful authentication. Handlers receive it via ctx.client — no need to dig into ctx.state["_jwt_payload"] or ctx.state["roles"].

Attributes:

Name	Type	Description
`client_id`	`str`	Unique client identifier (from JWT `sub` or API key mapping).
`tenant_id`	`str \| None`	Tenant / organisation the client belongs to. Populated by :class:`AuthMiddleware` from a configurable JWT claim (default `tenant_id`) or from the API-key config dict. A first-class field — rate limiting, audit entries, and tenant guards (`RequireTenant` / `HasTenant`) all key on it.
`roles`	`set[str]`	Set of role strings the client holds.
`scopes`	`set[str]`	Set of OAuth2 scope strings (from JWT `scope` claim).
`claims`	`dict[str, Any]`	Full JWT payload dict (empty for API key auth).
`issuer`	`str \| None`	JWT `iss` claim, or `None`.
`audience`	`str \| list[str] \| None`	JWT `aud` claim (string or list), or `None`.
`subject`	`str \| None`	JWT `sub` claim, or `None`.
`issued_at`	`float \| None`	JWT `iat` claim (Unix timestamp), or `None`.
`expires_at`	`float \| None`	JWT `exp` claim (Unix timestamp), or `None`.
`ip_address`	`str \| None`	Client IP address from the transport layer, or `None`.
`user_agent`	`str \| None`	`User-Agent` header value, or `None`.
`extra`	`dict[str, Any]`	Custom metadata populated by the server's `on_authenticate` enrichment hook. Any additional client info (org, plan tier, etc.) goes here.

@server.tool(auth=True)
async def my_tool(ctx: RequestContext) -> str:
    print(ctx.client.client_id)   # "agent-007"
    print(ctx.client.roles)       # {"admin", "analyst"}
    print(ctx.client.scopes)      # {"read", "write"}
    print(ctx.client.issuer)      # "https://auth.example.com"
    print(ctx.client.ip_address)  # "192.168.1.42"
    print(ctx.client.extra)       # {"org_id": "acme", "plan": "enterprise"}
    return "ok"

`has_all_roles(*roles)` ¶

Check if the client has all of the given roles.

`has_any_role(*roles)` ¶

Check if the client has at least one of the given roles.

`has_any_scope(*scopes)` ¶

Check if the client has at least one of the given scopes.

`has_role(role)` ¶

Check if the client has a specific role.

`has_scope(scope)` ¶

Check if the client has a specific OAuth2 scope.

RequestContext¶

`promptise.mcp.server.RequestContext` `dataclass` ¶

Per-request context available to handlers and middleware.

Automatically created for each tool call, resource read, or prompt request. Handlers can receive it by declaring a parameter typed as RequestContext, or by calling :func:get_context.

Attributes:

Name	Type	Description
`server_name`	`str`	Name of the MCPServer instance.
`tool_name`	`str`	Name of the tool/resource/prompt being invoked.
`request_id`	`str`	Unique identifier for this request. If the client sends an `X-Request-ID` header, that value is used; otherwise a random hex string is generated.
`client_id`	`str \| None`	Authenticated client identifier (set by auth middleware). Shortcut for `self.client.client_id`.
`client`	`ClientContext`	Structured client context with roles, scopes, claims, IP, and custom metadata. Populated by :class:`AuthMiddleware`.
`meta`	`dict[str, Any]`	Raw HTTP headers (lower-cased names) of the request that carries this call — per request, not per MCP session.
`state`	`dict[str, Any]`	Arbitrary per-request state (middleware can read/write). `state["tool_def"]` holds the definition being invoked (a `ToolDef`, `ResourceDef` or `PromptDef`); a resource read also sets `state["resource_uri"]`.
`request_type`	`str`	What is being invoked: `"tool"`, `"resource"` or `"prompt"`. Middleware runs for all three; branch on this when a middleware only makes sense for one of them.
`logger`	`Logger`	Pre-configured logger scoped to this request.

ToolResponse¶

`promptise.mcp.server.ToolResponse` `dataclass` ¶

Response wrapper that lets handlers return content with metadata.

Handlers can return a ToolResponse instead of a plain value to attach metadata that is logged, recorded in audit trails, and available to middleware. The content is serialised normally; metadata is stored on ctx.state["response_metadata"] for downstream use (e.g. by audit or webhook middleware).

Attributes:

Name	Type	Description
`content`	`Any`	The actual tool result (str, dict, list, etc.).
`metadata`	`dict[str, Any]`	Key-value metadata for observability and audit. Common keys: `"cache"` (`"hit"`/`"miss"`), `"source"` (where the data came from), `"version"`.

@server.tool()
async def search(query: str, ctx: RequestContext) -> ToolResponse:
    results = await db.search(query)
    return ToolResponse(
        content=results,
        metadata={"source": "primary_db", "result_count": len(results)},
    )

get_context¶

`promptise.mcp.server.get_context()` ¶

Return the current request context.

Raises:

Type	Description
`RuntimeError`	If called outside a request lifecycle.

get_request_headers¶

`promptise.mcp.server.get_request_headers()` ¶

Return HTTP request headers for the current async context.

Inside a tool handler on the HTTP/SSE transports these are the headers of the request that carries the current call (not of the request that opened the session). Returns an empty dict when called outside an HTTP request (e.g. stdio transport, tests).

set_request_headers¶

`promptise.mcp.server.set_request_headers(headers)` ¶

Store HTTP request headers for the current async context.

Called by the transport layer (ASGI middleware) so that call_tool can populate RequestContext.meta with headers like Authorization.

clear_request_headers¶

`promptise.mcp.server.clear_request_headers()` ¶

Clear HTTP request headers after the request completes.

Authentication¶

JWTAuth¶

`promptise.mcp.server.JWTAuth` ¶

JWT-based authentication provider.

Validates JWT tokens from the request metadata using HMAC-SHA256. Verified tokens are cached in an LRU to avoid repeated crypto operations on the hot path.

Parameters:

Name	Type	Description	Default
`secret`	`str`	Shared secret for HS256 signature verification.	required
`audience`	`str \| Sequence[str] \| None`	Expected `aud` claim — a string, or a list of accepted values. When set, a token is accepted only if its `aud` (a string or a list) contains one of them; a token without `aud` is rejected. Set it whenever the secret is shared by more than one service, so a token minted for one is refused by the others.	`None`
`issuer`	`str \| None`	Expected `iss` claim. When set, a token from any other issuer, or without `iss`, is rejected.	`None`
`meta_key`	`str`	Key in `ctx.meta` where the token is expected.	`'authorization'`
`cache_size`	`int`	Max number of verified tokens to cache (0 to disable).	`256`

auth = JWTAuth(
    secret=os.environ["JWT_SECRET"],
    audience="crm-mcp",
    issuer="https://auth.example.com",
)

`create_token(payload, *, expires_in=3600)` ¶

Create a signed JWT token (utility for testing).

When this provider has an audience or issuer and payload does not set aud / iss, the token gets the (first) expected audience and the issuer, so it verifies against this provider.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	Claims to include in the token.	required
`expires_in`	`int`	Token lifetime in seconds.	`3600`

`verify_token(token)` ¶

Check if a token is valid without requiring a request context.

Useful for transport-level auth gating.

AsymmetricJWTAuth¶

`promptise.mcp.server.AsymmetricJWTAuth` ¶

JWT authentication using asymmetric algorithms (RS256, ES256).

Validates JWT tokens signed with RSA or ECDSA keys. Requires the PyJWT and cryptography packages (optional dependencies).

Parameters:

Name	Type	Description	Default
`public_key`	`str`	PEM-encoded public key string, or path to a PEM file. Used for signature verification.	required
`algorithm`	`str`	JWT algorithm (`"RS256"` or `"ES256"`).	`'RS256'`
`audience`	`str \| Sequence[str] \| None`	Expected `aud` claim (a string or list of accepted values). When unset, tokens that carry an `aud` claim are rejected (PyJWT's default), so set it for IdP-issued tokens.	`None`
`issuer`	`str \| None`	Expected `iss` claim. When set, tokens from any other issuer are rejected.	`None`
`meta_key`	`str`	Key in `ctx.meta` where the token is expected.	`'authorization'`
`cache_size`	`int`	Max cached tokens (0 to disable).	`256`

auth = AsymmetricJWTAuth(
    public_key=open("public.pem").read(),
    algorithm="RS256",
)
server.add_middleware(AuthMiddleware(auth))

`authenticate(ctx)` `async` ¶

Authenticate using an asymmetric JWT token.

`verify_token(token)` ¶

Check if a token is valid without requiring a request context.

JwksAuth¶

`promptise.mcp.server.JwksAuth` ¶

JWT authentication against an identity provider's JWKS endpoint.

Verifies tokens issued by an external identity provider — Microsoft Entra, an OIDC provider, an internal IdP — by fetching the provider's public keys from its JWKS URL and selecting the one matching the token's kid. Keys are fetched on demand and cached, so issuer key rotation is handled without reconfiguration.

This is the server-side counterpart to promptise.identity: an agent presents an IdP-issued identity credential, this provider verifies it, and the validated sub / claims are surfaced on ctx.client so guards (RequireClientId, HasRole) and audit logs can see which agent called.

Requires the PyJWT and cryptography packages.

Parameters:

Name	Type	Description	Default
`jwks_url`	`str`	The IdP's JWKS endpoint (for Entra, e.g. `https://login.microsoftonline.com/<tenant>/discovery/v2.0/keys`).	required
`audience`	`str`	Expected `aud` claim — required. Set it to the resource the agents target (e.g. the App ID URI of this server). It is verified on every token, which is what prevents an agent from replaying a token it was issued for a different resource of the same IdP (token substitution).	required
`issuer`	`str \| None`	Expected `iss` claim. Strongly recommended: when set, tokens from any other issuer are rejected (defence in depth on top of the JWKS key set and audience check).	`None`
`algorithms`	`tuple[str, ...]`	Accepted signing algorithms. Asymmetric only, so a token claiming `HS256` (algorithm-confusion) is rejected.	`('RS256', 'ES256')`
`meta_key`	`str`	Key in `ctx.meta` holding the bearer token.	`'authorization'`
`leeway`	`float`	Seconds of clock skew tolerated when checking `exp`, `nbf` and `iat`.	`60.0`

Keys are cached for five minutes. A token signed with a key that is not in the cache (the IdP rotated its keys) makes the server re-fetch the JWKS immediately, at most once per key id per minute, so a rotated key is accepted on its first use.

Example::

auth = JwksAuth(
    jwks_url="https://idp.example.com/.well-known/jwks.json",
    audience="api://my-mcp-server",
    issuer="https://idp.example.com",
)
server.add_middleware(AuthMiddleware(auth))

Or let the JWKS endpoint be discovered from the issuer's OIDC .well-known/openid-configuration (see :meth:from_discovery)::

auth = JwksAuth.from_discovery(
    issuer="https://login.microsoftonline.com/<tenant>/v2.0",
    audience="api://my-mcp-server",
)

`authenticate(ctx)` `async` ¶

Authenticate by verifying a JWT against the IdP's JWKS.

`from_discovery(*, issuer, audience, algorithms=('RS256', 'ES256'), meta_key='authorization', request_timeout=5.0, leeway=60.0)` `classmethod` ¶

Build a :class:JwksAuth that discovers its JWKS from the issuer.

Resolves the JWKS endpoint from the IdP's OIDC discovery document at {issuer}/.well-known/openid-configuration instead of taking a jwks_url directly. The document is fetched once (lazily, on the first verification) and its issuer is checked against the configured issuer to prevent a spoofed discovery host. Tokens are then verified with their iss pinned to issuer (and aud to audience, which is required).

Parameters:

Name	Type	Description	Default
`issuer`	`str`	The OIDC issuer URL (its `.well-known` is read).	required
`audience`	`str`	Expected `aud` claim — required.	required
`algorithms`	`tuple[str, ...]`	Accepted signing algorithms.	`('RS256', 'ES256')`
`meta_key`	`str`	Key in `ctx.meta` holding the bearer token.	`'authorization'`
`request_timeout`	`float`	Seconds to wait for the discovery and JWKS fetches.	`5.0`
`leeway`	`float`	Seconds of clock skew tolerated when checking `exp`, `nbf` and `iat` (as for the constructor).	`60.0`

`verify_token(token)` ¶

Check if a token is valid without requiring a request context.

APIKeyAuth¶

`promptise.mcp.server.APIKeyAuth` ¶

API key-based authentication provider.

Supports two key formats:

Simple — {api_key: client_id}::

APIKeyAuth(keys={"sk-abc": "agent-1", "sk-xyz": "agent-2"})

Rich — {api_key: {"client_id": ..., "roles": [...]}}::

APIKeyAuth(keys={
    "sk-abc": {"client_id": "agent-1", "roles": ["admin", "write"]},
    "sk-xyz": {"client_id": "agent-2", "roles": ["read"]},
})

Rich keys populate ctx.state["roles"] so that role-based guards (HasRole, HasAllRoles) work out of the box.

Parameters:

Name	Type	Description	Default
`keys`	`dict[str, str \| dict[str, Any]]`	Mapping of `{api_key: client_id_or_config}`.	required
`header`	`str`	HTTP header name used to transmit the key.	`'x-api-key'`

`authenticate(ctx)` `async` ¶

Authenticate using an API key from request metadata.

On success, sets ctx.state["roles"] if the key config includes roles.

Returns:

Type	Description
`str`	The `client_id` associated with the key.

Raises:

Type	Description
`AuthenticationError`	If the key is missing or invalid.

`verify_token(key)` ¶

Check if an API key is valid without requiring a context.

AuthProvider¶

`promptise.mcp.server.AuthProvider` ¶

Bases: Protocol

Protocol for authentication providers.

A custom provider verifies the request's credential (ctx.meta holds the request headers) and returns the client id. To give the caller roles, scopes and a tenant, store the verified claims as a dict in ctx.state["jwt_payload"]: AuthMiddleware builds ctx.client from them as it does for a JWT (roles, scope, the tenant claim, iss / aud / sub / iat / exp).

`authenticate(ctx)` `async` ¶

Authenticate the request.

Returns:

Type	Description
`str`	Client identifier string on success.

Raises:

Type	Description
`AuthenticationError`	On authentication failure.

AuthMiddleware¶

`promptise.mcp.server.AuthMiddleware` ¶

Middleware that enforces authentication on tools with auth=True.

After successful authentication, populates ctx.client with a fully structured :class:ClientContext containing roles, scopes, standard JWT claims, client IP, and user-agent.

Optionally accepts an on_authenticate hook for custom enrichment (e.g. loading org, tenant, or plan info from a database).

Parameters:

Name	Type	Description	Default
`provider`	`Any`	An `AuthProvider` implementation.	required
`on_authenticate`	`OnAuthenticateHook \| None`	Optional async or sync callable that receives `(client_ctx, request_ctx)` and can mutate `client_ctx` (e.g. populate `client_ctx.extra`).	`None`
`tenant_claim`	`str`	JWT claim name carrying the client's tenant / organisation id (default `"tenant_id"`). The value lands on `ClientContext.tenant_id`, where rate limiting, audit entries, and tenant guards read it. For `APIKeyAuth`, the tenant comes from the key's config dict instead.	`'tenant_id'`

async def enrich_client(client: ClientContext, ctx: RequestContext):
    org = await db.get_org_for_client(client.client_id)
    client.extra["org_id"] = org.id
    client.extra["plan"] = org.plan

server.add_middleware(AuthMiddleware(auth, on_authenticate=enrich_client))

`authenticate(ctx)` `async` ¶

Verify the request's credentials and populate ctx.client.

Runs the provider and the on_authenticate hook. Used for every tool call to an auth=True tool, and for tools/list when the server hides tools the caller may not call.

Raises:

Type	Description
`AuthenticationError`	Missing or invalid credentials.

OnAuthenticateHook¶

`promptise.mcp.server.OnAuthenticateHook = Callable[[ClientContext, RequestContext], Awaitable[None] | None]` `module-attribute` ¶

TokenEndpointConfig¶

`promptise.mcp.server.TokenEndpointConfig` ¶

Configuration for the built-in OAuth 2.0 token issuer.

Parameters:

Name	Type	Description	Default
`jwt_auth`	`Any`	The `JWTAuth` instance used to sign tokens.	required
`clients`	`dict[str, dict[str, Any]]`	Mapping of `client_id` to client configuration. Each config must have `"secret"` and optionally `"roles"`, `"expires_in"`, and `"claims"`.	required
`path`	`str`	HTTP path for the endpoint (default `/auth/token`).	`'/auth/token'`
`default_expires_in`	`int`	Default access token lifetime in seconds.	`3600`
`refresh_token_ttl`	`int`	Refresh token lifetime in seconds (default 30 days). Set to `0` to disable refresh tokens.	`30 * 86400`

Guards¶

Guard¶

`promptise.mcp.server.Guard` ¶

Bases: Protocol

Protocol for tool access guards.

`check(ctx)` `async` ¶

Check whether the request is allowed.

Parameters:

Name	Type	Description	Default
`ctx`	`RequestContext`	The current request context.	required

Returns:

Type	Description
`bool`	`True` if allowed, `False` to deny.

`describe_denial(ctx)` ¶

Return a human-readable explanation of why access was denied.

Guards should override this to provide helpful error messages. The default returns a generic message with the guard class name.

RequireAuth¶

`promptise.mcp.server.RequireAuth` ¶

Bases: Guard

Guard that requires an authenticated client_id on the context.

HasRole¶

`promptise.mcp.server.HasRole` ¶

Bases: Guard

Guard that requires any of the given roles.

Reads roles from ctx.client.roles (or ctx.state["roles"]), populated by AuthMiddleware from the JWT payload.

Parameters:

Name	Type	Description	Default
`roles`	`str`	One or more role names. Access granted if the client has at least one of these roles.	`()`

HasAllRoles¶

`promptise.mcp.server.HasAllRoles` ¶

Bases: Guard

Guard that requires all of the given roles.

Parameters:

Name	Type	Description	Default
`roles`	`str`	All of these roles must be present on the client.	`()`

HasScope¶

`promptise.mcp.server.HasScope` ¶

Bases: Guard

Guard that requires any of the given OAuth2 scopes.

Reads scopes from ctx.client.scopes, populated by AuthMiddleware from the JWT scope claim (space-separated string per RFC 8693).

.. note::

Scopes are a JWT concept. When using ``APIKeyAuth`` without
JWT, ``ctx.client.scopes`` will be empty and scope guards
will always deny.  Use role-based guards (``HasRole``) for
API key auth.

Parameters:

Name	Type	Description	Default
`scopes`	`str`	One or more scope strings. Access granted if the client has at least one of these scopes.	`()`

@server.tool(auth=True, guards=[HasScope("read", "admin")])
async def get_data() -> str:
    return "data"

HasAllScopes¶

`promptise.mcp.server.HasAllScopes` ¶

Bases: Guard

Guard that requires all of the given OAuth2 scopes.

Parameters:

Name	Type	Description	Default
`scopes`	`str`	All of these scopes must be present on the client.	`()`

RequireClientId¶

`promptise.mcp.server.RequireClientId` ¶

Bases: Guard

Guard that requires a specific client_id.

Parameters:

Name	Type	Description	Default
`allowed_ids`	`str`	Set of allowed client identifiers.	`()`

Middleware¶

Composable middleware pipeline. Order matters: outermost first.

Middleware¶

`promptise.mcp.server.Middleware` ¶

Bases: Protocol

Protocol for middleware callables.

MiddlewareChain¶

`promptise.mcp.server.MiddlewareChain` ¶

Executes a stack of middleware around a handler.

Middleware runs in registration order (first added = outermost). The innermost function is the actual handler.

`add(middleware)` ¶

Append a middleware to the chain.

`run(ctx, handler, handler_kwargs)` `async` ¶

Execute the middleware chain, finishing with handler.

LoggingMiddleware¶

`promptise.mcp.server.LoggingMiddleware` ¶

Log every tool call with timing.

Parameters:

Name	Type	Description	Default
`log_level`	`int`	Logging level (default `logging.INFO`).	`INFO`

StructuredLoggingMiddleware¶

`promptise.mcp.server.StructuredLoggingMiddleware` ¶

Emit structured JSON log entries for every tool call.

Each log entry includes: - event: "tool_call_start" or "tool_call_end" - tool: Tool name - request_id: Unique request identifier - client_id: Authenticated client (if any) - duration_ms: Execution time (on completion) - status: "ok" or "error" - error: Error message (on failure)

Parameters:

Name	Type	Description	Default
`log_level`	`int`	Python logging level (default `logging.INFO`).	`INFO`
`include_args`	`bool`	Include tool arguments in logs (default `False` to avoid leaking sensitive data).	`False`

TimeoutMiddleware¶

`promptise.mcp.server.TimeoutMiddleware` ¶

Enforce per-call timeout.

Uses the tool's timeout setting if present, otherwise falls back to default_timeout.

Parameters:

Name	Type	Description	Default
`default_timeout`	`float`	Default timeout in seconds.	`30.0`

RateLimitMiddleware¶

`promptise.mcp.server.RateLimitMiddleware` ¶

Middleware that enforces rate limits.

Buckets are per client (and per tenant) only when ctx.client_id is set by the time this middleware runs, which takes an AuthMiddleware added before it and an authenticated tool (auth=True, or MCPServer(require_auth=True)). Otherwise every caller shares one "global" bucket; the server logs a warning at build time when an AuthMiddleware is installed but cannot identify the caller for this middleware. Pass key_func to choose the key yourself.

Parameters:

Name	Type	Description	Default
`limiter`	`Any \| None`	A `RateLimitStrategy` implementation. If `None`, a `TokenBucketLimiter` is created with the given params.	`None`
`per_tool`	`bool`	If `True`, rate limit per tool name (in addition to per client).	`False`
`key_func`	`Callable[[RequestContext], str] \| None`	Custom function to extract the rate limit key from context. Overrides `per_tool` behaviour.	`None`
`rate_per_minute`	`int`	Default rate if creating a new limiter.	`60`
`burst`	`int \| None`	Default burst if creating a new limiter.	`None`

TokenBucketLimiter¶

`promptise.mcp.server.TokenBucketLimiter` ¶

Token bucket rate limiter.

Parameters:

Name	Type	Description	Default
`rate_per_minute`	`float`	Sustained rate (tokens refilled per minute).	`60`
`burst`	`int \| None`	Maximum burst size (bucket capacity).	`None`

`consume(key)` ¶

Try to consume one token from the bucket for key.

CircuitBreakerMiddleware¶

`promptise.mcp.server.CircuitBreakerMiddleware` ¶

Circuit breaker middleware.

Tracks consecutive failures per tool. When failure_threshold consecutive failures occur, the circuit opens and subsequent calls are rejected immediately with :class:CircuitOpenError (a retryable CIRCUIT_OPEN error). After recovery_timeout seconds the circuit enters half-open state and lets exactly one probe call through; other calls are rejected until the probe finishes. A successful probe closes the circuit, a failed one re-opens it.

Only upstream and unexpected errors count as failures (see :func:is_upstream_failure): a tool rejecting bad input, an access denial, or a rate-limit / capacity refusal never pauses the tool for everyone. Errors that don't count leave the failure streak as it is.

Parameters:

Name	Type	Description	Default
`failure_threshold`	`int`	Consecutive failures before opening the circuit (default `5`).	`5`
`recovery_timeout`	`float`	Seconds to wait before probing recovery (default `60.0`).	`60.0`
`excluded_tools`	`set[str] \| None`	Tool names exempt from circuit breaking.	`None`
`is_failure`	`Callable[[BaseException], bool] \| None`	`(exc) -> bool` deciding whether an exception counts towards opening the circuit. Defaults to :func:`is_upstream_failure`.	`None`

`get_state(tool)` ¶

Get the current circuit state for a tool.

`reset(tool=None)` ¶

Reset circuit state for a tool (or all tools).

CircuitOpenError¶

`promptise.mcp.server.CircuitOpenError` ¶

Bases: MCPError

Raised when a tool call is rejected by an open circuit.

Reaches the client as a retryable CIRCUIT_OPEN error with details.retry_after_seconds, so an agent knows to wait (or use another tool) instead of treating the tool as broken. Register @server.exception_handler(CircuitOpenError) to customise it.

Attributes:

Name	Type	Description
`tool`		The tool whose circuit is open.
`retry_after`		Seconds until the breaker lets a probe call through.

is_upstream_failure¶

`promptise.mcp.server.is_upstream_failure(exc)` ¶

Default failure classifier for :class:CircuitBreakerMiddleware.

Counts:

any exception that is not an MCPError — a crash, a connection error, an upstream SDK raising: the tool is unhealthy;
a retryable MCPError other than the caller-side ones below — e.g. a TIMEOUT, or ToolError("upstream unavailable", retryable=True).

Does not count:

a non-retryable ToolError — the tool worked and rejected this input (ToolError("No product BAD"));
authentication / access denials, validation errors, rate-limit and concurrency refusals, approval denials, and CircuitOpenError.

CircuitState¶

`promptise.mcp.server.CircuitState` ¶

Bases: Enum

Circuit breaker states.

AuditMiddleware¶

`promptise.mcp.server.AuditMiddleware` ¶

HMAC-chained audit log middleware.

Writes one JSON line per tool call to log_path with optional argument/result capture and HMAC chain integrity.

Each signed entry carries seq (its position in the chain), prev_hash (the previous entry's hmac, or 64 zeros for the first entry) and hmac. When log_path already holds entries, the first entry this instance writes continues the chain from the file's last entry and is marked "resumed": true. Run one writer per file: two processes appending to the same file fork the chain.

A failed call has "status": "error", error ("ClassName: message") and, when it raised an :class:MCPError, error_code: the code the client received (ACCESS_DENIED for a role/scope/guard denial, AUTHENTICATION_ERROR for a missing or rejected credential, VALIDATION_ERROR, RATE_LIMIT_EXCEEDED, ...). Filter on error_code, not the class name: guard denials are raised as AuthenticationError too. Entries written before 1.3.5 have no error_code and still verify.

The middleware only sees requests that reach the MCP layer. With require_auth=True on an HTTP transport, a request whose token fails verification is rejected with 401 before that, so it is in the HTTP server's access log, not in this audit log.

Parameters:

Name	Type	Description	Default
`log_path`	`str \| None`	File path for audit log (JSONL format). If `None`, entries are only kept in memory and emitted via Python logging.	`None`
`signed`	`bool`	Enable HMAC chain (default `True`).	`True`
`hmac_secret`	`str \| None`	Secret key for HMAC chain. Resolved from (in order): 1. This parameter 2. `PROMPTISE_AUDIT_SECRET` env var 3. Without `log_path` only: an auto-generated random secret (logged as warning). A signed log file needs a key you keep, or nobody could ever verify it, so `log_path` with `signed=True` and no key raises `ValueError`.	`None`
`include_args`	`bool`	Log tool arguments (default `False` — may contain PII).	`False`
`include_result`	`bool`	Log tool results (default `False`).	`False`
`max_memory_entries`	`int \| None`	How many entries :attr:`entries` keeps in memory (default 10 000; `None` keeps all). The file keeps everything.	`10000`

Raises:

Type	Description
`ValueError`	`log_path` is set with `signed=True` and no key is configured.

`entries` `property` ¶

The most recent audit entries kept in memory (useful for testing).

At most max_memory_entries; the log file keeps every entry.

`verify_chain()` ¶

Verify the HMAC chain of the entries kept in memory.

Checks the entries this instance wrote (the most recent max_memory_entries). To verify a log file — every entry, across restarts, with the reason and line of the first break — use :func:~promptise.mcp.server.verify_audit_log.

Returns:

Type	Description
`bool`	`True` if the chain is valid, `False` if tampered.

WebhookMiddleware¶

`promptise.mcp.server.WebhookMiddleware` ¶

Fire webhooks on tool call events.

Supports tool.success, tool.error, and tool.call events. Webhooks are sent asynchronously and never block tool execution.

Parameters:

Name	Type	Description	Default
`url`	`str`	Webhook endpoint URL.	required
`events`	`set[str] \| None`	Set of event types to fire on (default: all).	`None`
`headers`	`dict[str, str] \| None`	Extra HTTP headers for the webhook request.	`None`
`timeout`	`float`	HTTP timeout in seconds (default `5.0`).	`5.0`

`sent` `property` ¶

Access sent webhook payloads (useful for testing).

ConcurrencyLimiter¶

`promptise.mcp.server.ConcurrencyLimiter` ¶

Middleware that limits concurrent in-flight tool executions.

When the limit is reached, incoming requests receive a retryable ConcurrencyLimitError (code CONCURRENCY_LIMIT_EXCEEDED) instead of queueing indefinitely.

Parameters:

Name	Type	Description	Default
`max_concurrent`	`int`	Maximum number of concurrent tool calls. `0` means unlimited (tracking only).	`100`

`active_requests` `property` ¶

Number of currently in-flight requests.

`peak_concurrent` `property` ¶

Highest concurrent request count observed.

`total_requests` `property` ¶

Total requests processed since server start.

`stats()` ¶

Return concurrency statistics.

PerToolConcurrencyLimiter¶

`promptise.mcp.server.PerToolConcurrencyLimiter` ¶

Middleware that enforces per-tool concurrency limits.

Reads max_concurrent from the ToolDef stored in ctx.state["tool_def"]. When a tool has max_concurrent set and the limit is reached, additional calls receive a retryable ConcurrencyLimitError (code CONCURRENCY_LIMIT_EXCEEDED).

Added automatically — by the server and by TestClient — when any tool declares max_concurrent. It is placed just outside the first CircuitBreakerMiddleware, so a call refused for capacity never reaches the breaker.

Semaphores are lazily created per tool name on first use.

BackgroundTasks¶

`promptise.mcp.server.BackgroundTasks` ¶

Collect and execute background tasks after the handler returns.

Tasks are added during handler execution and run sequentially after the response is sent. Errors are logged, never raised.

Can be injected via Depends(BackgroundTasks) in tool handlers.

`pending` `property` ¶

Number of tasks waiting to be executed.

`add(func, *args, **kwargs)` ¶

Schedule a task to run after the handler returns.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any]`	Sync or async callable.	required
`args`	`Any`	Positional arguments.	`()`
`kwargs`	`Any`	Keyword arguments.	`{}`

`execute()` `async` ¶

Run all queued tasks sequentially.

Errors are logged and swallowed — they never reach the client.

Observability¶

HealthCheck¶

`promptise.mcp.server.HealthCheck` ¶

Health and readiness probe manager.

Tracks named health checks and exposes liveness/readiness as JSON resources.

`add_check(name, check, *, required_for_ready=True)` ¶

Register a health check.

Parameters:

Name	Type	Description	Default
`name`	`str`	Check name.	required
`check`	`Callable[..., Any]`	Callable returning `True` (healthy) or `False`. Can be sync or async.	required
`required_for_ready`	`bool`	If `True`, failing this check makes readiness return unhealthy.	`True`

`liveness()` `async` ¶

Return liveness probe result as JSON string.

`liveness_report()` ¶

Liveness probe result as a dict (status is always "alive").

`readiness()` `async` ¶

Return readiness probe result as JSON string.

`readiness_report(*, include_errors=True)` `async` ¶

Run every check and return the readiness result as a dict.

Parameters:

Name	Type	Description	Default
`include_errors`	`bool`	Include the message of a check that raised. The unauthenticated HTTP probe route passes `False` so exception text never leaves the process.	`True`

Returns:

Type	Description
`dict[str, Any]`	`{"status": "ready" \| "not_ready", "checks": {name: {"healthy": bool, ...}}}`

`register_resources(server)` ¶

Register health resources on an MCPServer.

Also makes this check set back the server's GET /health and GET /health/ready HTTP routes.

Parameters:

Name	Type	Description	Default
`server`	`Any`	The `MCPServer` instance.	required

MetricsCollector¶

`promptise.mcp.server.MetricsCollector` ¶

Collects per-tool metrics: call counts, latency, errors.

`record_call(tool_name, latency, *, error=False)` ¶

Record a tool call.

`register_resource(server)` ¶

Register a metrics://server resource on server.

`snapshot()` ¶

Return a snapshot of all metrics.

MetricsMiddleware¶

`promptise.mcp.server.MetricsMiddleware` ¶

Middleware that records per-tool call metrics.

Parameters:

Name	Type	Description	Default
`collector`	`MetricsCollector \| None`	A `MetricsCollector` instance. Created automatically if not provided.	`None`

PrometheusMiddleware¶

`promptise.mcp.server.PrometheusMiddleware` ¶

Prometheus metrics middleware for MCP tool calls.

Records: - mcp_tool_calls_total — Counter of tool calls (labels: tool, status) - mcp_tool_duration_seconds — Histogram of call duration (labels: tool) - mcp_tool_in_flight — Gauge of in-flight calls (labels: tool)

Raises ImportError if prometheus-client is not installed (pip install "promptise[all]").

Parameters:

Name	Type	Description	Default
`namespace`	`str`	Metric namespace prefix (default `"mcp"`).	`'mcp'`
`registry`	`Any`	Custom Prometheus `CollectorRegistry`. Uses the default global registry if not provided.	`None`

`get_metrics_text()` ¶

Generate Prometheus text exposition format.

Returns:

Type	Description
`str`	Metrics in text/plain format for scraping.

OTelMiddleware¶

`promptise.mcp.server.OTelMiddleware` ¶

OpenTelemetry tracing middleware.

Creates a span for each tool call with attributes for tool name, request ID, client ID, and error status. Also records a histogram metric for tool call duration.

Raises ImportError if opentelemetry-api is not installed (pip install "promptise[all]"). It takes no endpoint: configure the exporter on the tracer / meter provider.

Parameters:

Name	Type	Description	Default
`service_name`	`str`	Service name for the tracer (default `"promptise-mcp-server"`).	`'promptise-mcp-server'`
`tracer_provider`	`Any`	Optional custom `TracerProvider`. If not given, uses the global provider.	`None`
`meter_provider`	`Any`	Optional custom `MeterProvider`. If not given, uses the global provider.	`None`

Dashboard¶

`promptise.mcp.server.Dashboard` ¶

Full-screen tabbed terminal dashboard with keyboard navigation.

Features:

6 tabs -- Overview, Tools, Agents, Logs, Metrics, Raw Logs
Arrow-key navigation -- left/right to switch tabs
Number keys -- 1-6 to jump directly
Tab key -- cycle to next tab
Live refresh -- every ~250ms in a daemon thread
Raw log capture -- Python logger output displayed in tab 6

Runs its own refresh + keyboard threads (both daemon), safe alongside uvicorn.

`start()` ¶

Start the live dashboard (takes over the terminal).

`stop()` ¶

Stop the dashboard and restore the terminal.

DashboardMiddleware¶

`promptise.mcp.server.DashboardMiddleware` ¶

Outermost middleware that records request metrics for the dashboard.

Must be inserted as the first middleware (outermost position) to capture all errors including authentication failures.

DashboardState¶

`promptise.mcp.server.DashboardState` `dataclass` ¶

Thread-safe shared state between middleware and dashboard renderer.

Simple counters and deque are GIL-protected for basic operations, making this safe without explicit locks for our single-writer (middleware) / single-reader (dashboard thread) pattern.

`record_request(log)` ¶

Record a completed request (called from middleware).

Caching¶

CacheBackend¶

`promptise.mcp.server.CacheBackend` ¶

Bases: Protocol

Protocol for cache backends.

`clear()` `async` ¶

Clear all cached values.

`delete(key)` `async` ¶

Delete a cached value.

`get(key)` `async` ¶

Get a cached value, or None if not found / expired.

`set(key, value, ttl)` `async` ¶

Store a value with a TTL in seconds.

InMemoryCache¶

`promptise.mcp.server.InMemoryCache` ¶

In-process cache with TTL-based expiry and background cleanup.

Thread-safe for asyncio (single-threaded event loop).

Parameters:

Name	Type	Description	Default
`max_size`	`int`	Maximum number of entries. When the limit is reached the least recently used entry (the one read or written longest ago) is evicted. `0` means unlimited.	`0`
`cleanup_interval`	`float`	Seconds between background sweeps of expired entries. `0` disables background cleanup (expired entries are still removed on access).	`60.0`

`evicted_count` `property` ¶

Total number of entries removed by background cleanup.

`size` `property` ¶

Current number of entries (including expired).

`stop_cleanup()` `async` ¶

Cancel the background cleanup task (for graceful shutdown).

RedisCache¶

`promptise.mcp.server.RedisCache` ¶

Redis-backed cache implementing the CacheBackend protocol.

Uses redis.asyncio for async Redis operations. Values are JSON-serialised for storage.

If redis is not installed, all operations raise ImportError on first use.

Parameters:

Name	Type	Description	Default
`url`	`str`	Redis connection URL (e.g. `"redis://localhost:6379/0"`).	`'redis://localhost:6379/0'`
`prefix`	`str`	Key prefix to namespace cache entries (default `"promptise:"`).	`'promptise:'`
`client`	`Any`	Pre-configured `redis.asyncio.Redis` client. If provided, `url` is ignored.	`None`

`clear()` `async` ¶

Clear all cached values with this prefix.

`close()` `async` ¶

Close the Redis connection (if we own it).

`delete(key)` `async` ¶

Delete a cached value.

`get(key)` `async` ¶

Get a cached value by key.

`set(key, value, ttl)` `async` ¶

Set a cached value with TTL (millisecond precision).

CacheMiddleware¶

`promptise.mcp.server.CacheMiddleware` ¶

Server-wide caching middleware.

Caches tool results, resource reads and prompt results, keyed on the server, request type, name, resource URI, arguments and — by default — the authenticated caller, so one client never receives a result computed for another. Note that it caches every tool that doesn't opt out: mark tools that change data or must always be fresh with @server.tool(cache=False) (tools annotated destructive_hint=True are skipped automatically), or use :func:cached on just the handlers you want cached.

A cache hit still runs the definition's guards (HasRole, HasTenant, ...), so a cached result is never returned to a caller the tool, resource or prompt would refuse. When it requires authentication but the caller has not been identified yet (CacheMiddleware added before AuthMiddleware), the request bypasses the cache — whatever the scope — so a hit can never skip authentication: add AuthMiddleware first.

Parameters:

Name	Type	Description	Default
`backend`	`CacheBackend \| None`	Cache backend.	`None`
`ttl`	`float`	Default TTL in seconds.	`60.0`
`scope`	`CacheScope`	Who may share an entry: `"client"` (default), `"tenant"` or `"shared"` — see :data:`CacheScope`.	`'client'`

cached (decorator)¶

`promptise.mcp.server.cached(ttl=60.0, *, key_func=None, backend=None, scope='client', coalesce=True)` ¶

Decorator that caches tool handler results.

Entries are scoped to the calling principal by default, so a result computed for one client is never returned to another — even when the handler reads the caller with :func:get_context rather than taking a ctx parameter. Injected parameters (ctx: RequestContext, Depends(...) values, BackgroundTasks ...) are not part of the key.

Identical calls that arrive while the first one is still computing wait for its result instead of each calling the upstream service (request coalescing, on by default), so a burst against a cold or just-expired entry costs one upstream call, not one per caller. Coalescing is per process; a RedisCache shared by several replicas still sees one upstream call per replica.

Parameters:

Name	Type	Description	Default
`ttl`	`float`	Time-to-live in seconds.	`60.0`
`key_func`	`Callable[..., str] \| None`	Custom key function `(func_name, kwargs) -> str`. Receives every keyword argument the handler gets (injected ones included). Its key is still prefixed with the scope, so a custom key cannot widen sharing beyond it.	`None`
`backend`	`CacheBackend \| None`	Cache backend. Defaults to the module-level `InMemoryCache` singleton.	`None`
`scope`	`CacheScope`	Who may share an entry: `"client"` (default), `"tenant"` or `"shared"` — see :data:`CacheScope`. Widen it only for data that does not depend on the caller.	`'client'`
`coalesce`	`bool`	Share one in-flight computation between identical concurrent calls (default `True`). Concurrent callers also share its error, if it raises.	`True`

@server.tool()
@cached(ttl=300)
async def my_open_tickets(ctx: RequestContext) -> list[dict]:
    return await db.tickets(owner=ctx.client.client_id)

@server.tool()
@cached(ttl=300, scope="shared")
async def exchange_rate(currency: str) -> float:
    return await fx.rate(currency)

Job Queue¶

Background task queue with priority, retry, progress, and cancellation.

MCPQueue¶

`promptise.mcp.server.MCPQueue` ¶

MCP-native job queue for asynchronous background work.

Allows server authors to define long-running job types that agents submit and poll for results, rather than blocking on synchronous tool calls.

When attached to an MCPServer, the queue auto-registers 5 MCP tools (queue_submit, queue_status, queue_result, queue_cancel, queue_list) and hooks into the server lifecycle for worker management.

Jobs are owned by the client and tenant that submitted them. The status, result, cancel and list tools only expose a caller's own jobs; a caller holding admin_role sees every job of its own tenant. Ownership relies on authenticated tool calls — on a server without require_auth=True, pass auth=True so the queue tools authenticate; unauthenticated callers all share one anonymous owner.

Parameters:

Name	Type	Description	Default
`server`	`Any \| None`	The MCPServer to attach to. When provided, tools and lifecycle hooks are registered immediately.	`None`
`backend`	`QueueBackend \| None`	Queue storage backend (default: InMemoryQueueBackend, which is process-local: jobs are lost on restart and not shared across replicas).	`None`
`max_workers`	`int`	Maximum concurrent job workers.	`4`
`default_timeout`	`float`	Default per-job timeout in seconds.	`300.0`
`result_ttl`	`float`	How long to keep completed job results before auto-cleanup (seconds).	`3600.0`
`cleanup_interval`	`float`	Seconds between cleanup sweeps.	`60.0`
`tool_prefix`	`str`	Prefix for auto-registered tool names.	`'queue'`
`auth`	`bool`	Require authentication on the queue tools (in addition to the server's `require_auth`).	`False`
`admin_role`	`str \| None`	Role that may see and cancel other clients' jobs within its own tenant. `None` disables the override.	`'admin'`
`cancel_grace_period`	`float`	Seconds a cancelled running job gets to stop on its own (its `CancellationToken` is set immediately) before its task is cancelled. Default `0`: the task is cancelled right away.	`0.0`

server = MCPServer(name="analytics")
queue = MCPQueue(server, max_workers=4)

@queue.job(name="generate_report", timeout=60)
async def generate_report(department: str) -> dict:
    await asyncio.sleep(10)
    return {"department": department, "rows": 500}

server.run(transport="http", port=8080)

`backend` `property` ¶

The queue storage backend.

`job_types` `property` ¶

List registered job type names.

`cancel(job_id, *, caller=None)` `async` ¶

Cancel a job.

A pending job is never started. A running job's CancellationToken is set and its handler task is cancelled (after cancel_grace_period, if set), so even a handler that never calls cancel.check() stops; anything it returns afterwards is discarded and the job stays cancelled.

Parameters:

Name	Type	Description	Default
`job_id`	`str`	Job identifier.	required
`caller`	`QueueCaller \| None`	Restrict to jobs this caller may access (`None` = unrestricted).	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with cancellation result.

Raises:

Type	Description
`ToolError`	If the job is not found (or not visible to caller).

`get_result(job_id, *, caller=None)` `async` ¶

Get job result.

If the job is still in progress, returns current status instead of a result.

Parameters:

Name	Type	Description	Default
`job_id`	`str`	Job identifier.	required
`caller`	`QueueCaller \| None`	Restrict to jobs this caller may access (`None` = unrestricted).	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with job result or status.

Raises:

Type	Description
`ToolError`	If the job is not found (or not visible to caller).

`job(name=None, *, timeout=None, max_retries=0, backoff_base=1.0)` ¶

Register an async function as a queue job type.

The decorated function runs in background workers, not inline with the tool call. It receives its arguments as keyword args, and may optionally accept ProgressReporter or CancellationToken parameters (detected by type annotation or a Depends(...) default). Arguments are validated against the handler signature when the job is submitted, and the resulting schema is listed in the queue_submit tool description.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Job type name (defaults to function name).	`None`
`timeout`	`float \| None`	Per-job timeout (overrides queue default).	`None`
`max_retries`	`int`	Max retry attempts on failure.	`0`
`backoff_base`	`float`	Exponential backoff base in seconds.	`1.0`

@queue.job(name="generate_report", timeout=60)
async def generate_report(department: str) -> dict:
    return {"department": department, "rows": 500}

`list_jobs(status=None, limit=20, *, offset=0, caller=None)` `async` ¶

List jobs, newest first.

Parameters:

Name	Type	Description	Default
`status`	`JobStatus \| None`	Optional status filter.	`None`
`limit`	`int`	Maximum number of jobs to return.	`20`
`offset`	`int`	Number of jobs to skip (pagination).	`0`
`caller`	`QueueCaller \| None`	Only list jobs this caller may access (`None` = every job).	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with the page of jobs and the total count of matching
`dict[str, Any]`	jobs visible to caller.

`register(server)` ¶

Register queue tools and lifecycle hooks on an MCPServer.

Called automatically when server is passed to the constructor. Call manually when constructing the queue separately.

Parameters:

Name	Type	Description	Default
`server`	`Any`	The MCPServer instance.	required

`register_health(health)` ¶

Add queue health checks to a HealthCheck instance.

Parameters:

Name	Type	Description	Default
`health`	`Any`	The HealthCheck to register on.	required

`start()` `async` ¶

Start the worker loop and cleanup task.

`status(job_id, *, caller=None)` `async` ¶

Get job status.

Parameters:

Name	Type	Description	Default
`job_id`	`str`	Job identifier.	required
`caller`	`QueueCaller \| None`	Restrict to jobs this caller may access (`None` = unrestricted).	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with job status information.

Raises:

Type	Description
`ToolError`	If the job is not found (or not visible to caller).

`stop()` `async` ¶

Gracefully stop all workers.

Jobs still running are cancelled. Jobs waiting out a retry backoff are put back on the queue, so a later :meth:start with the same backend picks them up.

`submit(job_type, args, *, priority=JobPriority.NORMAL, caller=None)` `async` ¶

Submit a job for background execution.

Parameters:

Name	Type	Description	Default
`job_type`	`str`	Registered job type name.	required
`args`	`dict[str, Any]`	Arguments for the job handler.	required
`priority`	`JobPriority`	Scheduling priority.	`NORMAL`
`caller`	`QueueCaller \| None`	Identity that will own the job (`None` for jobs submitted from server code).	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with job_id, status, and job_type.

Raises:

Type	Description
`ToolError`	If the job type is not registered (`UNKNOWN_JOB_TYPE`) or the arguments do not match the handler signature (`INVALID_JOB_ARGUMENTS`). Neither is retryable.

QueueBackend¶

`promptise.mcp.server.QueueBackend` ¶

Bases: Protocol

Protocol for pluggable queue storage backends.

The default implementation is InMemoryQueueBackend.

`count(status=None)` `async` ¶

Count jobs, optionally filtered by status.

`dequeue()` `async` ¶

Remove and return the highest-priority pending job, or None.

`enqueue(job)` `async` ¶

Add a job to the queue.

`get(job_id)` `async` ¶

Get a job by ID.

`list_jobs(status=None, limit=50)` `async` ¶

List jobs, optionally filtered by status.

`remove(job_id)` `async` ¶

Remove a job record. Returns True if found.

`update(job)` `async` ¶

Update a job's state.

InMemoryQueueBackend¶

`promptise.mcp.server.InMemoryQueueBackend` ¶

In-process queue backend using asyncio primitives.

Uses asyncio.PriorityQueue for the pending queue and a dict for job storage. Suitable for single-process deployments and testing: jobs live in this process only, so they are lost when it restarts and are not visible to other replicas of the server.

Parameters:

Name	Type	Description	Default
`max_size`	`int`	Maximum number of pending jobs (0 = unlimited).	`0`

`count(status=None)` `async` ¶

Count jobs, optionally filtered by status.

`dequeue()` `async` ¶

Remove and return the highest-priority pending job.

`enqueue(job)` `async` ¶

Add a job to the queue.

`get(job_id)` `async` ¶

Get a job by ID.

`list_jobs(status=None, limit=50)` `async` ¶

List jobs, optionally filtered by status.

`remove(job_id)` `async` ¶

Remove a job record.

`update(job)` `async` ¶

Update a job's state.

QueueCaller¶

`promptise.mcp.server.QueueCaller` `dataclass` ¶

Identity a queue operation is performed for.

The queue's MCP tools build one from the request context, so each caller only sees and controls its own jobs. Python code calling :class:MCPQueue methods directly passes caller=None (the default) and is not restricted.

Attributes:

Name	Type	Description
`client_id`	`str \| None`	Authenticated `client_id`, or `None` when the request was not authenticated.
`tenant_id`	`str \| None`	The caller's tenant, if any.
`is_admin`	`bool`	Holds the queue's admin role — may see and cancel every job in its own tenant, not just its own jobs.

`can_access(job)` ¶

Whether this caller may see or cancel job.

Tenants never cross: the admin role widens access to every client of the caller's own tenant, not to other tenants.

Streaming & Progress¶

StreamingResult¶

`promptise.mcp.server.StreamingResult` ¶

Collects partial results for streaming-style tool responses.

Tools that produce results incrementally can use this class to accumulate items. The framework serialises the collected items as a JSON array in a single TextContent response.

For true server-push streaming, the MCP protocol would need streaming result support. This class provides the collection pattern for when that lands.

Example::

result = StreamingResult()
result.add({"title": "Result 1", "score": 0.95})
result.add({"title": "Result 2", "score": 0.87})
return result  # → TextContent with JSON array

`items` `property` ¶

Access collected items.

`metadata` `property` ¶

Access result metadata.

`add(item)` ¶

Add a partial result item.

`add_many(items)` ¶

Add multiple items at once.

`set_metadata(key, value)` ¶

Attach metadata to the streaming result.

`to_dict()` ¶

Serialize to a dict for JSON encoding.

ProgressReporter¶

`promptise.mcp.server.ProgressReporter` ¶

Report progress during long-running tool calls.

Sends MCP notifications/progress messages to the connected client. If no progress token was provided by the client or no session is available (e.g. in tests), calls are silently ignored.

Use via dependency injection::

@server.tool()
async def etl(progress: ProgressReporter = Depends(ProgressReporter)) -> str:
    await progress.report(1, total=10, message="Step 1")
    ...

`report(progress, *, total=None, message=None)` `async` ¶

Send a progress notification to the client.

Parameters:

Name	Type	Description	Default
`progress`	`float`	Current progress value (e.g. items processed).	required
`total`	`float \| None`	Total expected value (e.g. total items). If provided, clients can calculate a percentage.	`None`
`message`	`str \| None`	Optional human-readable status message.	`None`

CancellationToken¶

`promptise.mcp.server.CancellationToken` ¶

Token for checking and responding to client cancellation requests.

The framework sets the token to cancelled state when the MCP client sends a notifications/cancelled notification. Tool handlers can poll this token during long-running operations.

Use via dependency injection::

@server.tool()
async def process(cancel: CancellationToken = Depends(CancellationToken)):
    for item in items:
        cancel.check()  # raises CancelledError if cancelled
        await do_work(item)

`is_cancelled` `property` ¶

Whether cancellation has been requested.

`reason` `property` ¶

The cancellation reason, if any.

`cancel(reason=None)` ¶

Mark this token as cancelled.

Called by the framework when a client cancellation is received.

`check()` ¶

Check if cancelled and raise CancelledError if so.

This is the primary API for handlers — call it at safe points in long-running loops.

`wait(timeout=None)` `async` ¶

Wait until cancelled or timeout expires.

Returns True if cancelled, False if timed out.

Session State¶

Per-client key-value storage that persists across tool calls.

SessionState¶

`promptise.mcp.server.SessionState` ¶

Key-value state scoped to the current MCP session.

Each client session gets its own isolated state. State persists across tool calls within the same session and is automatically cleaned up when the session ends.

Use via dependency injection::

@server.tool()
async def counter(session: SessionState = Depends(SessionState)) -> int:
    count = session.get("count", 0) + 1
    session.set("count", count)
    return count

`clear()` ¶

Clear all session state.

`delete(key)` ¶

Remove a key from session state.

`get(key, default=None)` ¶

Get a value from session state.

`keys()` ¶

Return all keys in session state.

`set(key, value)` ¶

Set a value in session state.

`to_dict()` ¶

Return a snapshot of all state.

SessionManager¶

`promptise.mcp.server.SessionManager` ¶

Manages per-session state stores.

Maps session IDs to SessionState instances. Thread-safe for use across async tasks.

The framework creates one SessionManager per MCPServer and looks up the current session by ID in the request context.

`active_sessions` `property` ¶

Number of active sessions.

`get_or_create(session_id)` ¶

Get or create a SessionState for the given session ID.

`remove(session_id)` ¶

Remove a session's state (called on disconnect).

Tool Versioning and Transforms¶

VersionedToolRegistry¶

`promptise.mcp.server.VersionedToolRegistry` ¶

Overlay registry that manages versioned tools.

Sits alongside the standard ToolRegistry and is consulted at list_tools time to expose versioned aliases.

Usage::

vr = VersionedToolRegistry()
vr.register("search", "1.0", tool_def_v1)
vr.register("search", "2.0", tool_def_v2)

# list_all() returns both pinned names and the latest alias

`get(name)` ¶

Look up by name or name@version.

`list_all()` ¶

Return all versioned tool defs (pinned names + latest alias).

`list_versions(base_name)` ¶

List available versions for a tool.

`register(base_name, version, tool_def)` ¶

Register a versioned tool definition.

ToolTransform¶

`promptise.mcp.server.ToolTransform` ¶

Bases: Protocol

Protocol for tool transforms.

A transform receives a list of ToolDef objects and returns a (potentially modified) list. Transforms are applied in order during list_tools — each sees the output of the previous one.

NamespaceTransform¶

`promptise.mcp.server.NamespaceTransform` ¶

Prefix all tool names with a namespace.

Example::

transform = NamespaceTransform(prefix="myapp")
# "search" → "myapp_search"

VisibilityTransform¶

`promptise.mcp.server.VisibilityTransform` ¶

Hide tools based on a predicate.

Hidden tools are removed from list_tools results but remain callable (for backwards compatibility with cached tool lists).

Example::

transform = VisibilityTransform(
    hidden={"admin_delete": lambda ctx: "admin" not in ctx.state.get("roles", set())}
)

TagFilterTransform¶

`promptise.mcp.server.TagFilterTransform` ¶

Only expose tools that have at least one of the required tags.

Example::

transform = TagFilterTransform(required_tags={"public"})
# Only tools tagged "public" are listed

Elicitation and Sampling¶

Request structured input or LLM completions from the client mid-execution.

Elicitor¶

`promptise.mcp.server.Elicitor` ¶

Request structured input from the user mid-execution.

Bound to the MCP session by the framework's DI wiring. Sends an elicitation/create request through ServerSession.elicit and resolves the client's :class:mcp.types.ElicitResult.

ask() returns None — never raises — when the answer is not a usable acceptance: the client declined or cancelled, did not declare the elicitation capability (the request is rejected by the client), the request timed out, or the transport failed. Every one of these is logged at WARNING/INFO level so an operator can tell why a gated call was denied instead of guessing.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Default timeout in seconds for elicitation requests.	`60.0`

`ask(message, schema=None, *, timeout=None)` `async` ¶

Ask the user for structured input.

Parameters:

Name	Type	Description	Default
`message`	`str`	Human-readable prompt.	required
`schema`	`dict[str, Any] \| None`	JSON Schema for the expected response. Defaults to an empty object schema (a bare confirmation).	`None`
`timeout`	`float \| None`	Override the default timeout, in seconds.	`None`

Returns:

Type	Description
`dict[str, Any] \| None`	The submitted form values when the user accepted, or `None`
`dict[str, Any] \| None`	when the elicitor is unbound, the client declined or cancelled,
`dict[str, Any] \| None`	the client does not support elicitation, the request timed
`dict[str, Any] \| None`	out, or the transport failed.

Sampler¶

`promptise.mcp.server.Sampler` ¶

Request LLM completions from the client.

Bound to the MCP session by the framework's DI wiring. If the client does not support sampling, methods return None silently.

`create_message(messages, *, max_tokens=1024, model=None, system=None, temperature=None, stop_sequences=None)` `async` ¶

Request an LLM completion from the client.

Parameters:

Name	Type	Description	Default
`messages`	`list[dict[str, str]]`	List of message dicts with `role` and `content`.	required
`max_tokens`	`int`	Maximum tokens to generate.	`1024`
`model`	`str \| None`	Model hint (client may ignore).	`None`
`system`	`str \| None`	System prompt.	`None`
`temperature`	`float \| None`	Sampling temperature.	`None`
`stop_sequences`	`list[str] \| None`	Stop sequences.	`None`

Returns:

Type	Description
`str \| None`	Generated text, or `None` if unavailable.

OpenAPI¶

Generate MCP tools from OpenAPI specs automatically.

OpenAPIProvider¶

`promptise.mcp.server.OpenAPIProvider` ¶

Generate MCP tools from an OpenAPI specification.

Parses an OpenAPI 3.x (or Swagger 2.x) spec and registers one MCP tool per operation. Each tool makes an HTTP request to the target API when called.

Parameters:

Name	Type	Description	Default
`spec`	`str \| dict[str, Any]`	Either a URL to fetch the spec from, a file path, or a pre-parsed dict.	required
`base_url`	`str \| None`	Override the server base URL from the spec.	`None`
`prefix`	`str`	Prefix for generated tool names (e.g. `"petstore_"`).	`''`
`include`	`set[str] \| None`	Only include operations matching these operation IDs.	`None`
`exclude`	`set[str] \| None`	Exclude operations matching these operation IDs.	`None`
`auth_header`	`tuple[str, str] \| None`	Default auth header for requests (e.g. `("Authorization", "Bearer tok-123")`).	`None`
`tags`	`list[str] \| None`	Tags to apply to all generated tools.	`None`

`parse()` ¶

Parse the spec and return operations (does not register tools).

Returns:

Type	Description
`list[dict[str, Any]]`	List of operation dicts with `operation_id`, `method`,
`list[dict[str, Any]]`	`path`, `summary`, and `input_schema`.

`register(server)` ¶

Parse the spec and register tools on the server.

Returns:

Type	Description
`int`	Number of tools registered.

Exception Handlers¶

ExceptionHandlerRegistry¶

`promptise.mcp.server.ExceptionHandlerRegistry` ¶

Registry mapping exception types to handler functions.

Handlers are matched using the exception's MRO (most specific first).

A handler receives (ctx, exc) and must return an MCPError instance, or None to fall through to the generic handler.

`find_handler(exc)` ¶

Find the best handler for exc by walking its MRO.

Returns None if no handler is registered for any type in the exception's MRO.

An MCPError already serialises to a structured error, so for one only handlers registered for an MCPError subclass (e.g. CircuitOpenError, RateLimitError) apply — a catch-all Exception handler does not swallow it.

`handle(ctx, exc)` `async` ¶

Attempt to handle exc.

Returns the MCPError produced by the matched handler, or None if no handler matched.

`register(exc_type, handler)` ¶

Register a handler for exc_type.

Parameters:

Name	Type	Description	Default
`exc_type`	`type[Exception]`	The exception class to handle.	required
`handler`	`Callable[..., Any]`	Async callable `(ctx, exc) -> MCPError \| None`.	required

Errors¶

MCPError¶

`promptise.mcp.server.MCPError` ¶

Bases: Exception

Base class for structured MCP server errors.

Attributes:

Name	Type	Description
`code`		Machine-readable error code (e.g. `"VALIDATION_ERROR"`).
`message`		Human-readable message.
`suggestion`		What the caller should try instead.
`retryable`		Whether re-trying the same call might succeed.
`details`		Additional structured context.

`to_text()` ¶

Serialise to a JSON text block suitable for MCP TextContent.

ToolError¶

`promptise.mcp.server.ToolError` ¶

Bases: MCPError

Error raised inside a tool handler.

ResourceError¶

`promptise.mcp.server.ResourceError` ¶

Bases: MCPError

Error raised inside a resource handler.

PromptError¶

`promptise.mcp.server.PromptError` ¶

Bases: MCPError

Error raised inside a prompt handler.

AuthenticationError¶

`promptise.mcp.server.AuthenticationError` ¶

Bases: MCPError

Raised when authentication fails.

RateLimitError¶

`promptise.mcp.server.RateLimitError` ¶

Bases: MCPError

Raised when rate limit is exceeded.

ConcurrencyLimitError¶

`promptise.mcp.server.ConcurrencyLimitError` ¶

Bases: RateLimitError

Raised when a tool (or the server) is already running its maximum number of concurrent calls.

A subclass of :class:RateLimitError so existing except RateLimitError code keeps working, but with its own code, CONCURRENCY_LIMIT_EXCEEDED: the caller is not over a quota, the tool is busy, and a retry shortly afterwards is likely to succeed.

ValidationError¶

`promptise.mcp.server.ValidationError` ¶

Bases: MCPError

Raised when input validation fails.

CancelledError¶

`promptise.mcp.server.CancelledError` ¶

Bases: Exception

Raised when a tool call is cancelled by the client.

Tool Types¶

ToolDef¶

`promptise.mcp.server.ToolDef` `dataclass` ¶

Internal definition of a registered tool.

ResourceDef¶

`promptise.mcp.server.ResourceDef` `dataclass` ¶

Internal definition of a registered resource or resource template.

The access-control fields mirror :class:ToolDef: a read runs through the server middleware chain with this definition in ctx.state["tool_def"], so AuthMiddleware, guards, rate limits and timeouts apply to resources exactly as they do to tools.

PromptDef¶

`promptise.mcp.server.PromptDef` `dataclass` ¶

Internal definition of a registered prompt.

The access-control fields mirror :class:ToolDef (see :class:ResourceDef).

ToolAnnotations¶

`promptise.mcp.server.ToolAnnotations` `dataclass` ¶

MCP tool annotations describing tool behaviour hints.

All fields are optional hints — they are not enforced by the framework but communicated to MCP clients so they can make informed decisions (e.g. skipping confirmation for read-only tools).

Attributes:

Name	Type	Description
`title`	`str \| None`	Human-readable title for the tool.
`read_only_hint`	`bool \| None`	Tool does not modify any state.
`destructive_hint`	`bool \| None`	Tool may perform destructive operations.
`idempotent_hint`	`bool \| None`	Calling the tool multiple times with the same arguments produces the same result.
`open_world_hint`	`bool \| None`	Tool may interact with external systems beyond the server's control.

Content¶

`promptise.mcp.server.Content = MCPTextContent | MCPImageContent | MCPEmbeddedResource | ImageContent` `module-attribute` ¶

ImageContent¶

`promptise.mcp.server.ImageContent` ¶

Helper for returning image content from tool handlers.

Wraps binary image data for automatic serialization to MCP ImageContent (base64-encoded).

Parameters:

Name	Type	Description	Default
`data`	`bytes`	Raw image bytes.	required
`mime_type`	`str`	MIME type (e.g. `"image/png"`, `"image/jpeg"`).	`'image/png'`

@server.tool()
async def chart(values: list[float]) -> ImageContent:
    png = render_chart(values)
    return ImageContent(data=png, mime_type="image/png")

`to_mcp()` ¶

Convert to an MCP ImageContent message.

TransportType¶

`promptise.mcp.server.TransportType` ¶

Bases: str, Enum

Supported MCP transport protocols.

CORSConfig¶

`promptise.mcp.server.CORSConfig` `dataclass` ¶

CORS configuration for HTTP and SSE transports.

The defaults cover a browser MCP client: the preflight admits the MCP request headers (:data:MCP_CORS_ALLOW_HEADERS) and the session id is readable from the initialize response (:data:MCP_CORS_EXPOSE_HEADERS). No origin is allowed until you name one.

CORS only decides what a browser lets the page read. The server's Origin validation is separate: pass the same origins to allowed_origins on run(), or the server answers the browser's requests 403.

Parameters:

Name	Type	Description	Default
`allow_origins`	`list[str]`	Allowed origin URLs. Use `["*"]` to allow all. Default: none.	`list()`
`allow_methods`	`list[str]`	Allowed HTTP methods.	`(lambda: ['GET', 'POST', 'DELETE', 'OPTIONS'])()`
`allow_headers`	`list[str]`	Allowed request headers. When you replace the default, keep `mcp-session-id` and `mcp-protocol-version` or the browser refuses every request after `initialize`.	`(lambda: list(MCP_CORS_ALLOW_HEADERS))()`
`expose_headers`	`list[str]`	Response headers the page may read. Keep `mcp-session-id`: without it a browser client cannot continue the session it opened.	`(lambda: list(MCP_CORS_EXPOSE_HEADERS))()`
`allow_credentials`	`bool`	Whether to allow credentials (cookies, auth). Cannot be combined with `allow_origins=[""]`: Starlette's CORS layer would then echo every* origin back with `Access-Control-Allow-Credentials: true`, letting any site make credentialed requests and read the answers. Name the origins instead.	`False`
`max_age`	`int`	Max seconds browsers may cache preflight responses.	`600`

Raises:

Type	Description
`ValueError`	`"*"` in `allow_origins` with `allow_credentials=True`.

server.run(
    transport="http",
    port=8080,
    allowed_origins=["https://app.example.com"],
    cors=CORSConfig(allow_origins=["https://app.example.com"]),
)

Testing¶

TestClient¶

`promptise.mcp.server.TestClient` ¶

`test = False` `class-attribute` `instance-attribute` ¶

In-process test client for :class:MCPServer.

Exercises the full call pipeline — validation, dependency injection, guard checks, middleware chain, handler invocation, and error serialisation — without starting a transport.

Parameters:

Name	Type	Description	Default
`server`	`Any`	The `MCPServer` instance to test.	required
`meta`	`dict[str, Any] \| None`	Simulated MCP request metadata (e.g. `{"authorization": "Bearer xxx"}`). Copied into every `RequestContext.meta` the client creates.	required

client = TestClient(server, meta={"authorization": "Bearer tok"})
result = await client.call_tool("search", {"query": "revenue"})

`call_tool(name, arguments=None, *, headers=None)` `async` ¶

Call a tool through the full middleware pipeline.

Returns the same content list as the real MCP server (may include TextContent, ImageContent, or EmbeddedResource). MCPError sub-classes are serialised into structured error JSON — they are not raised.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registered tool name.	required
`arguments`	`dict[str, Any] \| None`	Tool arguments (validated against the input model).	`None`
`headers`	`dict[str, str] \| None`	Simulated HTTP headers (e.g. `{"x-api-key": "..."}`). Merged with client-level meta (headers take precedence).	`None`

`get_prompt(name, arguments=None, *, headers=None)` `async` ¶

Get a prompt result through the full pipeline.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registered prompt name.	required
`arguments`	`dict[str, Any] \| None`	Prompt arguments. MCP clients send strings; they are coerced to the handler's type hints, as on the live server.	`None`
`headers`	`dict[str, str] \| None`	Simulated HTTP headers, merged with client-level meta.	`None`

Raises:

Type	Description
`ValueError`	If the prompt is not found.
`MCPError`	What the pipeline raised (`ValidationError` for a missing or malformed argument, `AuthenticationError` ...).

`list_prompts(*, headers=None)` `async` ¶

List the registered prompts (filtered like :meth:list_resources).

`list_resource_templates(*, headers=None)` `async` ¶

List the registered resource templates (filtered like :meth:list_resources).

`list_resources(*, headers=None)` `async` ¶

List the registered static resources (including docs://manifest).

With MCPServer(hide_unauthorized_tools=True) only the resources the client's credentials may read are listed, as on the live server.

`list_tools(*, headers=None)` `async` ¶

List the registered tools (including annotations).

With MCPServer(hide_unauthorized_tools=True) only the tools the client's credentials may call are listed, as on the live server.

Parameters:

Name	Type	Description	Default
`headers`	`dict[str, str] \| None`	Simulated HTTP headers, merged over the client meta.	`None`

`read_resource(uri, *, headers=None)` `async` ¶

Read a resource by URI.

Supports both static resources and URI templates, and runs the full pipeline (see :meth:read_resource_contents).

Parameters:

Name	Type	Description	Default
`uri`	`str`	The resource URI (e.g. `"config://app"`).	required
`headers`	`dict[str, str] \| None`	Simulated HTTP headers, merged with client-level meta.	`None`

Returns:

Type	Description
`str \| bytes`	The first content item: its text (a dict / list result arrives
`str \| bytes`	as JSON text), or the raw `bytes` for a binary resource.

Raises:

Type	Description
`ValueError`	If the resource is not found.
`MCPError`	What the pipeline raised (see :meth:`read_resource_contents`).

`read_resource_contents(uri, *, headers=None)` `async` ¶

Read a resource and return its MCP contents, as a client receives them.

Runs the full pipeline (middleware, auth, guards, coercion of template parameters). Each item carries mimeType; text arrives as TextResourceContents.text, bytes base64-encoded in BlobResourceContents.blob.

Parameters:

Name	Type	Description	Default
`uri`	`str`	The resource URI (e.g. `"config://app"`).	required
`headers`	`dict[str, str] \| None`	Simulated HTTP headers, merged with client-level meta.	`None`

Raises:

Type	Description
`ValueError`	If the resource is not found.
`MCPError`	What the pipeline raised (`AuthenticationError` for a missing credential or a denied guard, `ValidationError`, `RateLimitError`, a handler's `ResourceError` ...).

Logging¶

ServerLogger¶

`promptise.mcp.server.ServerLogger` ¶

Send log messages from the server to the connected MCP client.

Messages appear on the client side as notifications/message events. If no session is available (e.g. in tests), calls are silently ignored.

Use via dependency injection::

@server.tool()
async def run(log: ServerLogger = Depends(ServerLogger)) -> str:
    await log.info("starting")
    ...

`alert(data, *, logger=None)` `async` ¶

Send an alert-level log message.

`critical(data, *, logger=None)` `async` ¶

Send a critical-level log message.

`debug(data, *, logger=None)` `async` ¶

Send a debug-level log message.

`emergency(data, *, logger=None)` `async` ¶

Send an emergency-level log message.

`error(data, *, logger=None)` `async` ¶

Send an error-level log message.

`info(data, *, logger=None)` `async` ¶

Send an info-level log message.

`notice(data, *, logger=None)` `async` ¶

Send a notice-level log message.

`warning(data, *, logger=None)` `async` ¶

Send a warning-level log message.

Multi-Tenancy¶

RequireTenant¶

`promptise.mcp.server.RequireTenant` ¶

Bases: Guard

Guard that requires the client to belong to some tenant.

Reads ctx.client.tenant_id, populated by AuthMiddleware from the configured JWT tenant claim (default tenant_id) or from the API-key config dict. Fails closed: no authentication → no tenant → denied. Auto-applied to every tool when the server is built with MCPServer(require_tenant=True).

Example::

@server.tool(auth=True, guards=[RequireTenant()])
async def list_records() -> list:
    ...

HasTenant¶

`promptise.mcp.server.HasTenant` ¶

Bases: Guard

Guard that requires the client to belong to one of the given tenants.

Parameters:

Name	Type	Description	Default
`tenant_ids`	`str`	One or more allowed tenant identifiers.	`()`

@server.tool(auth=True, guards=[HasTenant("acme", "globex")])
async def partner_tool() -> str:
    ...

Approval Gates¶

ApprovalGateMiddleware¶

`promptise.mcp.server.ApprovalGateMiddleware` ¶

Enforces human approval on tools declared with requires_approval=True.

Runs in the middleware chain (install it after AuthMiddleware so the approval request carries the verified caller identity). For gated tools it builds an :class:ApprovalRequest, awaits the configured handler's decision, and either lets the call proceed or raises :class:ApprovalDeniedError. Non-gated tools pass through untouched.

Fail-closed semantics throughout: a timeout denies by default (on_timeout="deny"), a handler crash denies, and a decision carrying modified_arguments denies — the server-side gate cannot rewrite the call's arguments (they are already bound), and silently ignoring a reviewer's modification would execute something the reviewer did not approve.

Parameters:

Name	Type	Description	Default
`handler`	`Any`	An :class:`~promptise.approval.ApprovalHandler` (anything with `async request_approval(request) -> ApprovalDecision`), or a bare callable (sync/async, returning a decision or bool) — wrapped in a `CallbackApprovalHandler` automatically. Handlers may additionally implement `request_approval_ctx(request, ctx)` to receive the server :class:`RequestContext` (used by :class:`ElicitationApprover`).	required
`timeout`	`float`	Seconds to wait for a decision before applying `on_timeout` (default 300).	`300.0`
`on_timeout`	`Literal['deny', 'allow']`	`"deny"` (default) or `"allow"`.	`'deny'`
`include_arguments`	`bool`	Include the validated tool arguments in the approval request (default `True`). Disable for tools whose arguments are too sensitive to surface to reviewers.	`True`

PendingApprover¶

`promptise.mcp.server.PendingApprover` ¶

Blocks gated calls in a pending store until a human decides.

Independent four-eyes review: the gated call waits while a different human — anyone holding approver_role — lists pending requests and approves or denies them through two auto-registered, role-guarded admin tools:

approvals_list() — pending requests with tool, arguments, caller identity, and age.
approvals_decide(request_id, approve, reason) — release or deny; the reviewer's client_id is recorded on the decision.

The store is process-local (like the in-memory job queue); calls that time out before a decision are denied by the gate's default.

Parameters:

Name	Type	Description	Default
`server`	`Any`	Server to auto-register the admin tools on. Pass `None` and call :meth:`register_tools` later to register on a different server than the one being gated.	`None`
`approver_role`	`str`	Role required to list/decide (default `"approver"`).	`'approver'`
`max_pending`	`int`	Deny immediately beyond this many waiting requests (default 100) — a full queue must not become an unbounded pile of blocked calls.	`100`
`max_pending_per_client`	`int \| None`	Deny immediately once a single caller has this many requests waiting (default `None` — no per-client cap, only the global one). Without it the store is one shared queue: a single client or tenant that fills `max_pending` makes every other caller's gated call fail closed. A caller is identified by the request's `caller_user_id` (the verified `client_id` on the server-side gate), falling back to the `client_id` / `tenant_id` metadata and then `agent_id`; requests carrying no identity at all share one anonymous bucket — an unauthenticated flood cannot escape the cap by being anonymous. Must not exceed `max_pending` (nor `max_pending_per_tenant` when that is set).	`None`
`max_pending_per_tenant`	`int \| None`	Deny immediately once one tenant has this many requests waiting, across all of its clients (default `None` — no per-tenant cap). The per-client cap counts by client id, so a tenant holding several API keys could still take the whole store; this cap counts by the `tenant_id` metadata the server-side gate records (requests without one share one bucket). Must not exceed `max_pending`. The caps nest: client ≤ tenant ≤ store.	`None`

`caller_of(request_id)` ¶

The client_id that triggered a pending request, or None.

`client_key(request)` `staticmethod` ¶

The identity a request is counted under for max_pending_per_client.

Prefers the verified caller (caller_user_id), then the client_id and tenant_id metadata the server-side gate records, then the agent that raised the request. Returns "" for a request carrying no identity — all such requests share one bucket.

`decide(request_id, approved, *, reviewer_id=None, reason=None)` ¶

Resolve a pending request. Returns False if it is unknown (already decided, timed out, or never existed).

`pending()` ¶

Summaries of requests currently awaiting a decision.

`pending_for(client_key)` ¶

Number of waiting requests counted under client_key.

`pending_for_tenant(tenant_key)` ¶

Number of waiting requests counted under tenant_key.

`register_tools(server)` ¶

Register the role-guarded approvals_list / approvals_decide admin tools on server.

`tenant_key(request)` `staticmethod` ¶

The tenant a request is counted under for max_pending_per_tenant.

The tenant_id the server-side gate records in the request metadata; "" for a request without one — all such requests share one bucket.

ElicitationApprover¶

`promptise.mcp.server.ElicitationApprover` ¶

Approves by asking the human behind the calling MCP client.

Uses MCP elicitation: the connected client is asked to confirm the call ({"approve": bool, "reason": str}). This is confirm-your-own-action dual control — right for destructive-but-personal operations ("really delete this?"). For independent four-eyes review, use :class:PendingApprover instead.

Fail-closed: if the transport has no live MCP session (TestClient, stdio clients without elicitation support), the request is denied with a clear reason — never silently allowed.

ApprovalDeniedError¶

`promptise.mcp.server.ApprovalDeniedError` ¶

Bases: MCPError

Raised when a server-side approval gate denies (or times out on) a call.

Emitted by ApprovalGateMiddleware for tools declared with requires_approval=True. Not retryable by default: retrying would re-request the same approval a human just denied.

Rate Limiting¶

parse_rate_limit¶

`promptise.mcp.server.parse_rate_limit(spec)` ¶

Parse a declared rate-limit string like "100/min".

Accepted formats: "<count>/<unit>" where unit is one of s|sec|second, m|min|minute, h|hr|hour.

Parameters:

Name	Type	Description	Default
`spec`	`str`	The declared limit, e.g. `"100/min"`, `"10/sec"`, `"500/hour"`.	required

Returns:

Type	Description
`float`	`(rate_per_minute, burst)` — the sustained refill rate normalised
`int`	to tokens-per-minute, and the bucket capacity (the declared count).

Raises:

Type	Description
`ValueError`	If the spec is malformed, so a typo fails at tool registration instead of silently never limiting.

DeclaredRateLimitMiddleware¶

`promptise.mcp.server.DeclaredRateLimitMiddleware` ¶

Enforces per-tool rate limits declared via @server.tool(rate_limit=...).

Also enforces rate_limit= on @server.resource / @server.resource_template / @server.prompt.

Auto-inserted into the middleware chain at build time when any registered tool declares a rate_limit — no manual wiring required (mirrors PerToolConcurrencyLimiter). Each declaring tool gets its own token bucket sized from its spec; buckets are keyed per client when authentication populates ctx.client_id, falling back to a single shared bucket for unauthenticated tools.

Coexists with a user-installed :class:RateLimitMiddleware: that one enforces a server-wide policy, this one enforces each tool's declared contract.

MCP Server API Reference¶

Core¶

MCPServer¶

promptise.mcp.server.MCPServer ¶

middleware property ¶

add_middleware(middleware) ¶

asgi_app(transport='http', *, allowed_hosts=None, allowed_origins=None, cors=None, stateless=False) ¶

enable_token_endpoint(jwt_auth, clients, *, path='/auth/token', default_expires_in=86400) ¶

exception_handler(exc_type) ¶

include_prompts(*sources) ¶

include_router(router, *, prefix='', tags=None) ¶

notify_prompts_changed() async ¶

notify_resources_changed() async ¶

notify_tools_changed() async ¶

on_shutdown(func) ¶

on_startup(func) ¶

prompt(name=None, *, description=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None) ¶

resource(uri, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None) ¶

resource_template(uri_template, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None) ¶

run(transport='stdio', *, host='127.0.0.1', port=8080, dashboard=False, cors=None, allowed_hosts=None, allowed_origins=None) ¶

run_async(transport='stdio', *, host='127.0.0.1', port=8080, dashboard=False, cors=None, allowed_hosts=None, allowed_origins=None) async ¶

tool(name=None, *, description=None, tags=None, auth=False, rate_limit=None, timeout=None, guards=None, roles=None, title=None, read_only_hint=None, destructive_hint=None, idempotent_hint=None, open_world_hint=None, max_concurrent=None, requires_approval=False, cache=True) ¶

MCPRouter¶

promptise.mcp.server.MCPRouter ¶

include_prompts(*sources) ¶

include_router(router, *, prefix='', tags=None) ¶

prompt(name=None, *, description=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None) ¶

resource(uri, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None) ¶

resource_template(uri_template, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None) ¶

tool(name=None, *, description=None, tags=None, auth=False, rate_limit=None, timeout=None, guards=None, roles=None, title=None, read_only_hint=None, destructive_hint=None, idempotent_hint=None, open_world_hint=None, max_concurrent=None, requires_approval=False, cache=True) ¶

ServerSettings¶

promptise.mcp.server.ServerSettings ¶

Depends¶

promptise.mcp.server.Depends(dependency, *, use_cache=True) ¶

mount (server composition)¶

promptise.mcp.server.mount(parent, child, *, prefix='', tags=None) ¶

hot_reload¶

promptise.mcp.server.hot_reload(server=None, *, transport='http', host='127.0.0.1', port=8080, watch_dirs=None, poll_interval=1.0, dashboard=False, allowed_hosts=None, allowed_origins=None) ¶

Context¶

ClientContext¶

promptise.mcp.server.ClientContext dataclass ¶

has_all_roles(*roles) ¶

has_any_role(*roles) ¶

has_any_scope(*scopes) ¶

has_role(role) ¶

has_scope(scope) ¶

RequestContext¶

promptise.mcp.server.RequestContext dataclass ¶

ToolResponse¶

promptise.mcp.server.ToolResponse dataclass ¶

get_context¶

promptise.mcp.server.get_context() ¶

get_request_headers¶

promptise.mcp.server.get_request_headers() ¶

set_request_headers¶

promptise.mcp.server.set_request_headers(headers) ¶

clear_request_headers¶

promptise.mcp.server.clear_request_headers() ¶

Authentication¶

JWTAuth¶

promptise.mcp.server.JWTAuth ¶

create_token(payload, *, expires_in=3600) ¶

verify_token(token) ¶

AsymmetricJWTAuth¶

promptise.mcp.server.AsymmetricJWTAuth ¶

authenticate(ctx) async ¶

verify_token(token) ¶

JwksAuth¶

promptise.mcp.server.JwksAuth ¶

authenticate(ctx) async ¶

from_discovery(*, issuer, audience, algorithms=('RS256', 'ES256'), meta_key='authorization', request_timeout=5.0, leeway=60.0) classmethod ¶

verify_token(token) ¶

APIKeyAuth¶

promptise.mcp.server.APIKeyAuth ¶

authenticate(ctx) async ¶

verify_token(key) ¶

AuthProvider¶

promptise.mcp.server.AuthProvider ¶

authenticate(ctx) async ¶

AuthMiddleware¶

`promptise.mcp.server.MCPServer` ¶

`middleware` `property` ¶

`add_middleware(middleware)` ¶

`asgi_app(transport='http', *, allowed_hosts=None, allowed_origins=None, cors=None, stateless=False)` ¶

`enable_token_endpoint(jwt_auth, clients, *, path='/auth/token', default_expires_in=86400)` ¶

`exception_handler(exc_type)` ¶

`include_prompts(*sources)` ¶

`include_router(router, *, prefix='', tags=None)` ¶

`notify_prompts_changed()` `async` ¶

`notify_resources_changed()` `async` ¶

`notify_tools_changed()` `async` ¶

`on_shutdown(func)` ¶

`on_startup(func)` ¶

`prompt(name=None, *, description=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

`resource(uri, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

`resource_template(uri_template, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

`run(transport='stdio', *, host='127.0.0.1', port=8080, dashboard=False, cors=None, allowed_hosts=None, allowed_origins=None)` ¶

`run_async(transport='stdio', *, host='127.0.0.1', port=8080, dashboard=False, cors=None, allowed_hosts=None, allowed_origins=None)` `async` ¶

`tool(name=None, *, description=None, tags=None, auth=False, rate_limit=None, timeout=None, guards=None, roles=None, title=None, read_only_hint=None, destructive_hint=None, idempotent_hint=None, open_world_hint=None, max_concurrent=None, requires_approval=False, cache=True)` ¶

`promptise.mcp.server.MCPRouter` ¶

`include_prompts(*sources)` ¶

`include_router(router, *, prefix='', tags=None)` ¶

`prompt(name=None, *, description=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

`resource(uri, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

`resource_template(uri_template, *, name=None, description=None, mime_type=None, tags=None, auth=False, roles=None, guards=None, rate_limit=None, timeout=None)` ¶

`tool(name=None, *, description=None, tags=None, auth=False, rate_limit=None, timeout=None, guards=None, roles=None, title=None, read_only_hint=None, destructive_hint=None, idempotent_hint=None, open_world_hint=None, max_concurrent=None, requires_approval=False, cache=True)` ¶

`promptise.mcp.server.ServerSettings` ¶

`promptise.mcp.server.Depends(dependency, *, use_cache=True)` ¶

`promptise.mcp.server.mount(parent, child, *, prefix='', tags=None)` ¶

`promptise.mcp.server.hot_reload(server=None, *, transport='http', host='127.0.0.1', port=8080, watch_dirs=None, poll_interval=1.0, dashboard=False, allowed_hosts=None, allowed_origins=None)` ¶

`promptise.mcp.server.ClientContext` `dataclass` ¶

`has_all_roles(*roles)` ¶

`has_any_role(*roles)` ¶

`has_any_scope(*scopes)` ¶

`has_role(role)` ¶

`has_scope(scope)` ¶

`promptise.mcp.server.RequestContext` `dataclass` ¶

`promptise.mcp.server.ToolResponse` `dataclass` ¶

`promptise.mcp.server.get_context()` ¶

`promptise.mcp.server.get_request_headers()` ¶

`promptise.mcp.server.set_request_headers(headers)` ¶

`promptise.mcp.server.clear_request_headers()` ¶

`promptise.mcp.server.JWTAuth` ¶

`create_token(payload, *, expires_in=3600)` ¶

`verify_token(token)` ¶

`promptise.mcp.server.AsymmetricJWTAuth` ¶

`authenticate(ctx)` `async` ¶

`verify_token(token)` ¶

`promptise.mcp.server.JwksAuth` ¶

`authenticate(ctx)` `async` ¶

`from_discovery(*, issuer, audience, algorithms=('RS256', 'ES256'), meta_key='authorization', request_timeout=5.0, leeway=60.0)` `classmethod` ¶

`verify_token(token)` ¶

`promptise.mcp.server.APIKeyAuth` ¶

`authenticate(ctx)` `async` ¶

`verify_token(key)` ¶

`promptise.mcp.server.AuthProvider` ¶

`authenticate(ctx)` `async` ¶

`promptise.mcp.server.AuthMiddleware` ¶

`authenticate(ctx)` `async` ¶

`promptise.mcp.server.OnAuthenticateHook = Callable[[ClientContext, RequestContext], Awaitable[None] | None]` `module-attribute` ¶

`promptise.mcp.server.TokenEndpointConfig` ¶

`promptise.mcp.server.Guard` ¶

`check(ctx)` `async` ¶

`describe_denial(ctx)` ¶

`promptise.mcp.server.RequireAuth` ¶

`promptise.mcp.server.HasRole` ¶

`promptise.mcp.server.HasAllRoles` ¶

`promptise.mcp.server.HasScope` ¶

`promptise.mcp.server.HasAllScopes` ¶

`promptise.mcp.server.RequireClientId` ¶

`promptise.mcp.server.Middleware` ¶

`promptise.mcp.server.MiddlewareChain` ¶

`add(middleware)` ¶

`run(ctx, handler, handler_kwargs)` `async` ¶

`promptise.mcp.server.LoggingMiddleware` ¶

`promptise.mcp.server.StructuredLoggingMiddleware` ¶

`promptise.mcp.server.TimeoutMiddleware` ¶

`promptise.mcp.server.RateLimitMiddleware` ¶

`promptise.mcp.server.TokenBucketLimiter` ¶

`consume(key)` ¶