Core Primitives & Observability API¶

API reference for protocol-agnostic core primitives (Tool, ToolCall, ToolResult, ToolArgumentError, @tool, ToolContext), OpenTelemetry GenAI observability (TelemetryConfig, SpanManager), JSON-RPC 2.0 wire types (JSONRPCRequest, JSONRPCError, JSONRPCErrorException), SSE SubscriptionManager, the litestar-mcp exception hierarchy, and internal ASGI scope executors (execute_tool, execute_handler_response, blank_http_scope).

For guides, see Core Architectural Concepts and OpenTelemetry & Observability.

Core Tool & Context Primitives (`litestar_mcp.core`)¶

Tool ¶

Executable tool with a JSON Schema, msgspec argument model, and universal DI resolution.

Attributes:

Name	Type	Description
`description`	`str`
`fn`	`Callable[..., Any]`	Default: `fn`
`bound_instance`	`Any \| None`	Default: `bound_instance`
`name`	`str`	Default: `str(name or getattr(fn, '__name__', 'tool'))`
`dependencies`	`dict[str, Any]`	Default: `dict(dependencies) if dependencies is not None else {}`
`annotations`	`dict[str, Any]`	Default: `merged_annotations`
`parameters`	`dict[str, Any]`	Default: `param_schema`
`stripped_parameters`	`set[str]`	Names of parameters injected from context or DI rather than passed by the caller.
`di_parameters`	`dict[str, DIParameterSpec]`	Return mapping of DI-injected parameter names to their specifications.
`input_schema`	`dict[str, Any]`	JSON Schema representing tool inputs.

register_dependencies ¶

register_dependencies(controller_dependencies)

Merge controller-level dependencies and rebuild the tool schema.

from_schema ¶

from_schema(
    name,
    description=None,
    parameters=None,
    call=None,
    *,
    annotations=None,
    read_only=None,
)

Construct a tool instance from an explicit JSON schema and callable.

to_json_schema ¶

to_json_schema()

Return the JSON schema representation exposed to LLMs.

execute ¶

execute(arguments=None, *, context=None)

Validate arguments, inject context and DI parameters, and execute the tool.

ToolCall ¶

Bases: Struct

A model-requested tool invocation.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registered name of the tool to invoke.	required
`call_id`	`str`	Unique correlation identifier for this tool invocation.	required
`arguments`	`dict[str, Any]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`
`provider_metadata`	`dict[str, dict[str, Any]]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`

Attributes:

Name	Type	Description
`name`	`str`	Registered name of the tool to invoke.
`call_id`	`str`	Unique correlation identifier for this tool invocation.
`arguments`	`dict[str, Any]`	Model-supplied keyword arguments for the tool.
`provider_metadata`	`dict[str, dict[str, Any]]`	Provider-specific metadata attached to the tool call.

ToolResult ¶

Bases: Struct

The JSON-compatible outcome of one tool invocation.

Parameters:

Name	Type	Description	Default
`call_id`	`str`	Correlation identifier matching the originating `ToolCall.call_id`.	required
`name`	`str`	Name of the tool that produced this result.	required
`content`	`Any`	JSON-serializable return payload or error dict from the tool.	required
`is_error`	`bool`	Whether the tool invocation raised an error.	`False`

Attributes:

Name	Type	Description
`call_id`	`str`	Correlation identifier matching the originating `ToolCall.call_id`.
`name`	`str`	Name of the tool that produced this result.
`content`	`Any`	JSON-serializable return payload or error dict from the tool.
`is_error`	`bool`	Whether the tool invocation raised an error.
`result`	`Any`	Alias for content.
`error`	`str \| None`	Error string if invocation failed, or None.

ToolArgumentError ¶

Bases: LitestarMCPError, ValueError

Raised when model-supplied arguments fail validation against the tool's argument model.

tool ¶

tool(
    name_or_fn=None,
    *,
    name=None,
    description=None,
    dependencies=None,
    annotations=None,
    read_only=None,
)

Decorate a callable as an executable tool.

ToolContext ¶

Identity, tenant, request metadata, and DI container context injected into tools.

Parameters:

Name	Type	Description	Default
`user`	`Any \| None`	Authenticated principal from `connection.user`, or `None`.	`None`
`auth`	`Any \| None`	Authentication value from `connection.auth`, or `None`.	`None`
`tenant_id`	`str \| None`	Tenant identifier resolved from the principal or `request.state`.	`None`
`roles`	`tuple[str, ...]`	Principal roles.	`()`
`scopes`	`tuple[str, ...]`	Principal scopes.	`()`
`session_id`	`str \| None`	Conversation session identifier.	`None`
`turn_id`	`str \| None`	Agent turn identifier.	`None`
`request`	`Any \| None`	Originating Litestar connection, or `None` outside HTTP.	`None`
`state`	`dict[str, Any]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`
`type_encoders`	`TypeEncodersMap \| None`	Encoders resolved from the route's ownership layers, used to serialize tool results.	`None`
`container`	`Any \| None`	Active DI container (Dishka, Wireup, svcs, dependency-injector, or custom) for resolving tool dependencies.	`None`
`dishka_container`	`Any \| None`	Alias for `container` when using Dishka.	`None`
`dependency_resolver`	`Callable[[str, Any, ToolContext], Any] \| None`	Optional callback `(param_name, type_hint, ctx) -> Any \| Awaitable[Any]` for custom DI frameworks.	`None`

Attributes:

Name	Type	Description
`user`	`Any \| None`	Authenticated principal from `connection.user`, or `None`.
`auth`	`Any \| None`	Authentication value from `connection.auth`, or `None`.
`tenant_id`	`str \| None`	Tenant identifier resolved from the principal or `request.state`.
`roles`	`tuple[str, ...]`	Principal roles.
`scopes`	`tuple[str, ...]`	Principal scopes.
`session_id`	`str \| None`	Conversation session identifier.
`turn_id`	`str \| None`	Agent turn identifier.
`request`	`Any \| None`	Originating Litestar connection, or `None` outside HTTP.
`state`	`dict[str, Any]`	Mutable per-invocation values shared between tools.
`type_encoders`	`TypeEncodersMap \| None`	Encoders resolved from the route's ownership layers, used to serialize tool results.
`container`	`Any \| None`	Active DI container (Dishka, Wireup, svcs, dependency-injector, or custom) for resolving tool dependencies.
`dishka_container`	`Any \| None`	Alias for `container` when using Dishka.
`dependency_resolver`	`Callable[[str, Any, ToolContext], Any] \| None`	Optional callback `(param_name, type_hint, ctx) -> Any \| Awaitable[Any]` for custom DI frameworks.

from_connection ¶

from_connection(connection, *, session_id=None, turn_id=None)

Build a context from a Litestar connection's user, auth, state, and route encoders.

owner_key ¶

owner_key()

Return "<tenant or default>:<user id>" for authenticated callers and None for anonymous ones.

OpenTelemetry GenAI Observability¶

TelemetryConfig ¶

Configuration for OpenTelemetry observability in litestar-mcp.

Parameters:

Name	Type	Description	Default
`enable_spans`	`bool`	Whether to emit OpenTelemetry spans for MCP and agent operations.	`False`
`provider_factory`	`Callable[[], Any] \| None`	Optional factory returning a custom OpenTelemetry `TracerProvider`.	`None`
`resource_attributes`	`dict[str, Any]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`
`tracer_name`	`str`	Instrumentation scope name passed to `get_tracer`.	`'litestar_mcp'`

Attributes:

Name	Type	Description
`enable_spans`	`bool`	Whether to emit OpenTelemetry spans for MCP and agent operations.
`provider_factory`	`Callable[[], Any] \| None`	Optional factory returning a custom OpenTelemetry `TracerProvider`.
`resource_attributes`	`dict[str, Any]`	Extra OpenTelemetry resource attributes attached to emitted spans.
`tracer_name`	`str`	Instrumentation scope name passed to `get_tracer`.

SpanManager ¶

Lazy OpenTelemetry span manager with graceful degradation.

Provides zero-overhead no-op execution when OpenTelemetry is not installed or when spans are disabled.

Attributes:

Name	Type	Description
`is_enabled`	`bool`	Return True once OpenTelemetry spans are available and active.

start_agent_span ¶

start_agent_span(
    agent_name="",
    action="invoke_agent",
    *,
    agent_group=None,
    session_id=None,
    turn_id=None,
    model=None,
    provider=None,
    attributes=None,
)

Start a span representing an agent turn with standard GenAI semantic conventions.

start_tool_span ¶

start_tool_span(tool_name='', call_id=None, *, attributes=None)

Start a span representing a single tool execution.

set_usage ¶

set_usage(span, prompt_tokens, completion_tokens)

Record input and output token usage on an agent turn span.

start_span ¶

start_span(name, attributes=None)

Start a generic span with custom attributes.

end_span ¶

end_span(span, error=None)

Close a span and record errors when provided.

span ¶

span(name, attributes=None)

Context manager for tracing blocks with automatic end and error capture.

use ¶

use(span)

Make span current in the OpenTelemetry context during execution.

JSON-RPC 2.0 & Subscription Streams¶

JSONRPCRequest ¶

A parsed JSON-RPC 2.0 request.

Parameters:

Name	Type	Description	Default
`jsonrpc`	`str`	JSON-RPC protocol version (`"2.0"`).	required
`method`	`str`	The RPC method name.	required
`id`	`Any \| None`	Request identifier (absent for notifications).	`None`
`params`	`dict[str, Any]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`

Attributes:

Name	Type	Description
`jsonrpc`	`str`	JSON-RPC protocol version (`"2.0"`).
`method`	`str`	The RPC method name.
`id`	`Any \| None`	Request identifier (absent for notifications).
`params`	`dict[str, Any]`	Method parameters.
`is_notification`	`bool`	Whether this is a notification (no id field).

JSONRPCError ¶

A JSON-RPC 2.0 error object.

Parameters:

Name	Type	Description	Default
`code`	`int`	Integer error code.	required
`message`	`str`	Short description of the error.	required
`data`	`Any \| None`	Optional additional error data.	`None`

Attributes:

Name	Type	Description
`code`	`int`	Integer error code.
`message`	`str`	Short description of the error.
`data`	`Any \| None`	Optional additional error data.

to_dict ¶

to_dict()

Convert to a JSON-serializable dict.

JSONRPCErrorException ¶

Bases: Exception

Raised by method handlers to signal a JSON-RPC error.

SubscriptionManager ¶

Manage stateless, filtered subscription response streams.

start ¶

start()

Start cross-worker fan-out when a ChannelsPlugin was supplied.

open ¶

open(subscription_id, notifications)

Open a filtered stream whose first item is its acknowledgement.

publish ¶

publish(method, params)

Publish a notification only to subscriptions whose filter matches.

disconnect ¶

disconnect(stream_id)

Remove one stream and wake its consumer.

close_all ¶

close_all()

Gracefully close every active stream.

Core Exceptions¶

LitestarMCPError ¶

Bases: Exception

Base exception for Litestar MCP.

BridgeConnectionError ¶

Bases: LitestarMCPError

Raised when the stdio bridge cannot connect to its target endpoint.

BridgeMessageTooLargeError ¶

Bases: LitestarMCPError, ValueError

Raised when a stdio JSON-RPC frame exceeds the bridge line limit.

MissingDependencyError ¶

Bases: LitestarMCPError, ImportError

Missing optional dependency.

This exception is raised when a module depends on a dependency that has not been installed.

Parameters:

Name	Type	Description	Default
`package`	`str`	Name of the missing package.	required
`install_package`	`str \| None`	Optional alternative package name to install directly.	`None`
`extra`	`str \| None`	Optional project extra that installs the missing package.	`None`

Internal ASGI Scope Execution (`litestar_mcp.mcp.executor`)¶

execute_tool ¶

execute_tool(
    handler, app, tool_args, *, request=None, config=None, tool_name=None
)

Execute an MCP tool handler through Litestar's full HTTP pipeline.

Runs, in order:

Guards from every ownership layer (app → router → controller → handler).
before_request (closest-wins). A truthy return short-circuits the handler; a falsy return (None / "" / 0 / [] / {}) falls through to normal dispatch.
create_kwargs_model → to_kwargs → resolve_dependencies → parse_values_from_connection_kwargs → handler.fn(**parsed) (skipped on short-circuit).
handler.to_response(...) + capture-send — this is where after_request, type_encoders, response-class selection, and media-type negotiation happen.
On any raised exception: app-level after_exception observers fire (observer failures are logged and swallowed), then handler.resolve_exception_handlers() is walked MRO-style. A matching handler's returned Response renders in-place; if no handler matches, the exception propagates to the route layer's JSON-RPC blanket catch.
after_response (closest-wins) fires in finally, exactly once, on every path — success, short-circuit, guard failure, exception.

Transport caveat: JSON-RPC has no HTTP semantics, so response headers and cookies set in after_request or by the response class are silently dropped. status_code >= 400 maps to isError: true.

Parameters:

Name	Type	Description	Default
`handler`	`BaseRouteHandler`	The MCP-tool-marked route handler.	required
`app`	`Litestar`	The running Litestar application.	required
`tool_args`	`dict[str, Any]`	Arguments from the MCP `tools/call` request; routed into path / query / body based on the handler's signature.	required
`request`	`Request[Any, Any, Any] \| None`	Inbound `litestar.Request` in HTTP mode, `None` for CLI / stdio invocations.	`None`
`config`	`MCPConfig \| None`	MCP plugin configuration. When supplied with `tool_name`, tool-call observability callbacks are invoked around dispatch.	`None`
`tool_name`	`str \| None`	MCP tool name for callback payloads. Omit for resources, prompts, CLI calls, and other non-`tools/call` dispatch.	`None`

Returns:

Type	Description
`Any`	The decoded response body for success paths.

Raises:

Type	Description
`MCPToolErrorResult`	When the handler pipeline or an exception handler renders `status_code >= 400`. The `content` attribute carries the already-rendered payload for the JSON-RPC route layer.

execute_handler_response ¶

execute_handler_response(handler, app, tool_args, *, request=None)

Execute a handler and return raw response bytes plus decoded content.

blank_http_scope ¶

blank_http_scope(app)

Return a minimum ASGI HTTP scope for stdio-mode dispatch.

Keys mirror litestar.testing.RequestFactory._create_scope so Litestar's native dispatch pipeline finds everything it expects. Parity audited 2026-04-19 — the only delta is asgi.spec_version: RequestFactory uses "3.0" (stale), we use the current ASGI spec version "2.4". Every other key matches what RequestFactory emits (after _build_dispatch_scope adds the per-request method, path, raw_path, query_string, headers, path_params, route_handler, and path_template values).

Core Primitives & Observability API¶

Core Tool & Context Primitives (litestar_mcp.core)¶

Tool ¶

register_dependencies ¶

from_schema ¶

to_json_schema ¶

execute ¶

ToolCall ¶

ToolResult ¶

ToolArgumentError ¶

tool ¶

ToolContext ¶

from_connection ¶

owner_key ¶

OpenTelemetry GenAI Observability¶

TelemetryConfig ¶

SpanManager ¶

start_agent_span ¶

start_tool_span ¶

set_usage ¶

start_span ¶

end_span ¶

span ¶

use ¶

JSON-RPC 2.0 & Subscription Streams¶

JSONRPCRequest ¶

JSONRPCError ¶

to_dict ¶

JSONRPCErrorException ¶

SubscriptionManager ¶

start ¶

open ¶

publish ¶

disconnect ¶

close_all ¶

Core Exceptions¶

LitestarMCPError ¶

BridgeConnectionError ¶

BridgeMessageTooLargeError ¶

MissingDependencyError ¶

Internal ASGI Scope Execution (litestar_mcp.mcp.executor)¶

execute_tool ¶

execute_handler_response ¶

blank_http_scope ¶

Core Tool & Context Primitives (`litestar_mcp.core`)¶

Internal ASGI Scope Execution (`litestar_mcp.mcp.executor`)¶