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`

`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.

`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`

Example::

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. The MCP handler calls render_async(**arguments) to produce fully rendered prompt text with context providers, strategy, perspective, and constraints applied.

Parameters:

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

Example::

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`

`on_shutdown(func)` ¶

Register a shutdown hook.

`on_startup(func)` ¶

Register a startup hook.

`prompt(name=None, *, description=None)` ¶

Register a function as an MCP prompt.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Prompt name (defaults to function name).	`None`
`description`	`str \| None`	Description (defaults to docstring).	`None`

`resource(uri, *, name=None, description=None, mime_type='text/plain')` ¶

Register a function as an MCP resource.

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`	MIME type of the resource content.	`'text/plain'`

`resource_template(uri_template, *, name=None, description=None, mime_type='text/plain')` ¶

Register a function as an MCP resource template.

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`	MIME type.	`'text/plain'`

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

Start the server (blocking).

Parameters:

Name	Type	Description	Default
`transport`	`str`	`"stdio"`, `"http"`, or `"sse"`.	`'stdio'`
`host`	`str`	Bind host for HTTP/SSE transports.	`'0.0.0.0'`
`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`

`run_async(transport='stdio', *, host='0.0.0.0', port=8080, dashboard=False, cors=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.	`'0.0.0.0'`
`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`

`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)` ¶

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).	`None`
`roles`	`list[str] \| None`	Required roles shorthand (creates `HasRole` guard).	`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`

Example::

@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 per-tool auth flag.	`None`
`middleware`	`list[Any] \| None`	Router-level middleware (runs after server middleware).	`None`
`guards`	`list[Any] \| None`	Router-level guards applied to all tools.	`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)` ¶

Register a prompt.

`resource(uri, *, name=None, description=None, mime_type='text/plain')` ¶

Register a resource.

`resource_template(uri_template, *, name=None, description=None, mime_type='text/plain')` ¶

Register a 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)` ¶

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)` ¶

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`

Example::

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).
`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, tenant, plan tier, etc.) goes here.

Example::

@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 from the transport layer.
`state`	`dict[str, Any]`	Arbitrary per-request state (middleware can read/write).
`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"`.

Example::

@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.

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
`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`

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

Create a signed JWT token (utility for testing).

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'`
`meta_key`	`str`	Key in `ctx.meta` where the token is expected.	`'authorization'`
`cache_size`	`int`	Max cached tokens (0 to disable).	`256`

Example::

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.

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.

`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`

Example::

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

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.	`()`

Example::

@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.

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`	`int`	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. After recovery_timeout seconds the circuit enters half-open state and allows one probe call through.

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`

`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: Exception

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

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.

Parameters:

Name	Type	Description	Default
`log_path`	`str \| None`	File path for audit log (JSONL format). If `None`, entries are only 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. Auto-generated random secret (logged as warning)	`None`
`include_args`	`bool`	Log tool arguments (default `False` — may contain PII).	`False`
`include_result`	`bool`	Log tool results (default `False`).	`False`

`entries` `property` ¶

Access in-memory audit entries (useful for testing).

`verify_chain()` ¶

Verify the HMAC chain integrity of logged entries.

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 RateLimitError 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 RateLimitError.

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.

`readiness()` `async` ¶

Return readiness probe result as JSON string.

`register_resources(server)` ¶

Register health resources on an MCPServer.

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.

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.

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. Oldest entries are evicted when the limit is reached. `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.

CacheMiddleware¶

`promptise.mcp.server.CacheMiddleware` ¶

Server-wide caching middleware.

Caches all tool results based on tool name + arguments. Opt-out individual tools by setting tdef.cache = False in state.

Parameters:

Name	Type	Description	Default
`backend`	`CacheBackend \| None`	Cache backend.	`None`
`ttl`	`float`	Default TTL in seconds.	`60.0`

cached (decorator)¶

`promptise.mcp.server.cached(ttl=60.0, *, key_func=None, backend=None)` ¶

Decorator that caches tool handler results.

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`. Defaults to hashing the function name + JSON-serialised kwargs.	`None`
`backend`	`CacheBackend \| None`	Cache backend. Defaults to the module-level `InMemoryCache` singleton.	`None`

Example::

@server.tool()
@cached(ttl=300)
async def expensive_query(query: str) -> dict:
    return await db.search(query)

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.

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).	`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'`

Example::

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)` `async` ¶

Cancel a job.

Parameters:

Name	Type	Description	Default
`job_id`	`str`	Job identifier.	required

Returns:

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

Raises:

Type	Description
`ToolError`	If the job is not found.

`get_result(job_id)` `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

Returns:

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

Raises:

Type	Description
`ToolError`	If the job is not found.

`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 _JobProgressReporter or CancellationToken parameters (detected by type annotation).

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`

Example::

@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)` `async` ¶

List jobs.

Parameters:

Name	Type	Description	Default
`status`	`JobStatus \| None`	Optional status filter.	`None`
`limit`	`int`	Maximum number of jobs to return.	`20`

Returns:

Type	Description
`dict[str, Any]`	Dict with jobs list and total count.

`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)` `async` ¶

Get job status.

Parameters:

Name	Type	Description	Default
`job_id`	`str`	Job identifier.	required

Returns:

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

Raises:

Type	Description
`ToolError`	If the job is not found.

`stop()` `async` ¶

Gracefully stop all workers.

`submit(job_type, args, *, priority=JobPriority.NORMAL)` `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`

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.

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.

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.

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. If the client does not support elicitation, ask() returns None silently.

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.	`None`
`timeout`	`float \| None`	Override default timeout.	`None`

Returns:

Type	Description
`dict[str, Any] \| None`	Parsed response dict, or `None` if unavailable/declined.

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.

`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.

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.

PromptDef¶

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

Internal definition of a registered prompt.

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'`

Example::

@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.

Parameters:

Name	Type	Description	Default
`allow_origins`	`list[str]`	Allowed origin URLs. Use `["*"]` to allow all.	`list()`
`allow_methods`	`list[str]`	Allowed HTTP methods.	`(lambda: ['GET', 'POST', 'DELETE', 'OPTIONS'])()`
`allow_headers`	`list[str]`	Allowed request headers.	`(lambda: ['Content-Type', 'Authorization', 'x-api-key'])()`
`allow_credentials`	`bool`	Whether to allow credentials (cookies, auth).	`False`
`max_age`	`int`	Max seconds browsers may cache preflight responses.	`600`

Example::

server.run(
    transport="http",
    port=8080,
    cors=CORSConfig(
        allow_origins=["https://app.example.com"],
        allow_headers=["Authorization", "x-api-key"],
    ),
)

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

Example::

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)` `async` ¶

Get a prompt result.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registered prompt name.	required
`arguments`	`dict[str, str] \| None`	Prompt arguments (string-valued).	`None`

Raises:

Type	Description
`ValueError`	If the prompt is not found.

`list_prompts()` `async` ¶

List all registered prompts.

`list_resource_templates()` `async` ¶

List all registered resource templates.

`list_resources()` `async` ¶

List all registered static resources.

`list_tools()` `async` ¶

List all registered tools (including annotations).

`read_resource(uri)` `async` ¶

Read a resource by URI.

Supports both static resources and URI templates.

Parameters:

Name	Type	Description	Default
`uri`	`str`	The resource URI (e.g. `"config://app"`).	required

Raises:

Type	Description
`ValueError`	If the resource is not found.

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.

MCP Server API Reference¶

Core¶

MCPServer¶

promptise.mcp.server.MCPServer ¶

middleware property ¶

add_middleware(middleware) ¶

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

on_shutdown(func) ¶

on_startup(func) ¶

prompt(name=None, *, description=None) ¶

resource(uri, *, name=None, description=None, mime_type='text/plain') ¶

resource_template(uri_template, *, name=None, description=None, mime_type='text/plain') ¶

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

run_async(transport='stdio', *, host='0.0.0.0', port=8080, dashboard=False, cors=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) ¶

MCPRouter¶

promptise.mcp.server.MCPRouter ¶

include_prompts(*sources) ¶

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

prompt(name=None, *, description=None) ¶

resource(uri, *, name=None, description=None, mime_type='text/plain') ¶

resource_template(uri_template, *, name=None, description=None, mime_type='text/plain') ¶

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

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

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

APIKeyAuth¶

promptise.mcp.server.APIKeyAuth ¶

authenticate(ctx) async ¶

verify_token(key) ¶

AuthProvider¶

promptise.mcp.server.AuthProvider ¶

authenticate(ctx) async ¶

AuthMiddleware¶

promptise.mcp.server.AuthMiddleware ¶

OnAuthenticateHook¶

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

TokenEndpointConfig¶

promptise.mcp.server.TokenEndpointConfig ¶

Guards¶

Guard¶

promptise.mcp.server.Guard ¶

check(ctx) async ¶

`promptise.mcp.server.MCPServer` ¶

`middleware` `property` ¶

`add_middleware(middleware)` ¶

`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)` ¶

`on_shutdown(func)` ¶

`on_startup(func)` ¶

`prompt(name=None, *, description=None)` ¶

`resource(uri, *, name=None, description=None, mime_type='text/plain')` ¶

`resource_template(uri_template, *, name=None, description=None, mime_type='text/plain')` ¶

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

`run_async(transport='stdio', *, host='0.0.0.0', port=8080, dashboard=False, cors=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)` ¶

`promptise.mcp.server.MCPRouter` ¶

`include_prompts(*sources)` ¶

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

`prompt(name=None, *, description=None)` ¶

`resource(uri, *, name=None, description=None, mime_type='text/plain')` ¶

`resource_template(uri_template, *, name=None, description=None, mime_type='text/plain')` ¶

`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)` ¶

`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)` ¶

`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.APIKeyAuth` ¶

`authenticate(ctx)` `async` ¶

`verify_token(key)` ¶

`promptise.mcp.server.AuthProvider` ¶

`authenticate(ctx)` `async` ¶

`promptise.mcp.server.AuthMiddleware` ¶

`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)` ¶

`promptise.mcp.server.CircuitBreakerMiddleware` ¶

`get_state(tool)` ¶

`reset(tool=None)` ¶

`promptise.mcp.server.CircuitOpenError` ¶

`promptise.mcp.server.CircuitState` ¶

`promptise.mcp.server.AuditMiddleware` ¶

`entries` `property` ¶

`verify_chain()` ¶

`promptise.mcp.server.WebhookMiddleware` ¶