Agent API#

Public classes and protocols for declaring agents, controllers, skills, and model execution.

PromptController#

PromptMessage#

SkillController#

Agent#

class litestar_mcp.agent.Agent[source]#

Bases: object

Agent definition with instructions, tools, skills, and an optional model binding.

__post_init__()[source]#

Instantiate skill classes once, normalize tools, and reject duplicate tool names.

Return type:: None

property tool_set: tuple[Tool, ...]#: Return the agent's own tools followed by skill tools.

property skill_instances: tuple[SkillController, ...]#: Return the instantiated skill controllers attached to this agent.

property combined_instructions: str#: Return agent instructions joined with skill instructions by blank lines.

__init__(name, description='', instructions='', tools=(), skills=(), model=None)#

AgentGroup#

class litestar_mcp.agent.AgentGroup[source]#

Bases: object

Coordinator and specialist agents with a group-owned transfer_to_agent tool.

__init__(coordinator, specialists=())[source]#

tools_for(agent)[source]#

Return tools available to an agent in this group.

Return type:: tuple[Tool, ...]

get_agent(name)[source]#

Retrieve an agent by name from coordinator or specialist roster.

Return type:: Agent

resolve_transfer(call)[source]#

Resolve a tool call into an agent transfer target if applicable.

Return type:: Agent | None

AgentRuntime#

class litestar_mcp.agent.AgentRuntime[source]#

Bases: object

Runs agent turns with budgets, owner-scoped sessions, and streamed frames.

__init__(target, *, default_model=None, budget=None, sessions=None, span_manager=None, compaction_hook=None, stream_buffer=64)[source]#: Initialize runtime with target agent or group and execution policies.

async get_history(context, session_id)[source]#

Return conversation history scoped to connection owner.

Return type:: list[AgentMessage] | None

async produce(turn, send)[source]#

Execute turn loop as a producer coroutine emitting frames to the send stream.

Return type:: TurnResponse

open_stream(turn)[source]#

Open a background task group streaming frames to an in-memory receive stream.

Return type:: AsyncIterator[MemoryObjectReceiveStream[AgentStreamFrame]]

async stream(turn)[source]#

Yield stream frames sequentially from open_stream.

Return type:: AsyncIterator[AgentStreamFrame]

async run_turn(turn)[source]#

Run turn to completion and return TurnResponse.

Return type:: TurnResponse

GoogleGenAIClient#

class litestar_mcp.agent.GoogleGenAIClient[source]#

Bases: object

First-party Google GenAI model client wrapping google.genai with httpx2 support.

__init__(model='gemini-3.8-flash', *, api_key=None, vertexai=None, project=None, location=None, thinking_budget=None, thinking_level=None, include_thoughts=None, temperature=None, http_client=None, http_options=None, client=None)[source]#: Initialize Google GenAI client with configuration.

property provider_name: str#: Return canonical OpenTelemetry provider name.

async stream_turn(*, system_instruction, messages, tools)[source]#

Stream an agent turn using Google GenAI SDK.

Return type:: AsyncIterator[ModelDelta]

MockModelClient#

class litestar_mcp.agent.MockModelClient[source]#

Bases: object

Deterministic scriptable model client for fast offline unit tests.

__init__(responses=None)[source]#: Initialize mock client with optional response queue.

async stream_turn(*, system_instruction, messages, tools)[source]#

Record turn inputs and yield queued scripted responses.

Return type:: AsyncIterator[ModelDelta]

ModelDelta#

class litestar_mcp.agent.ModelDelta[source]#

Bases: object

Incremental chunk emitted by a ModelClient during an agent turn.

__init__(event_type, text=None, thought=None, tool_name=None, call_id=None, arguments=None, prompt_tokens=None, completion_tokens=None, provider_metadata=<factory>, thought_tokens=None)#

Tool#

ToolContext#

AgentChatController#

class litestar_mcp.agent.AgentChatController[source]#

Bases: Controller

SSE chat, synchronous turns, and owner-scoped history over an injected AgentRuntime.

path: str#

A path fragment for the controller.

All route handlers under the controller will have the fragment appended to them. If not set it defaults to /.

after_request: AfterRequestHookHandler | None#

A sync or async function executed before a Request is passed to any route handler.

If this function returns a value, the request will not reach the route handler, and instead this value will be used.

after_response: AfterResponseHookHandler | None#

A sync or async function called after the response has been awaited.

It receives the Request instance and should not return any values.

before_request: BeforeRequestHookHandler | None#

A sync or async function called immediately before calling the route handler.

It receives the Request instance and any non-None return value is used for the response, bypassing the route handler.

cache_control: CacheControlHeader | None#

A CacheControlHeader header to add to route handlers of this controller.

Can be overridden by route handlers.

dependencies: Dependencies | None#: A string keyed dictionary of dependency Provider instances.

dto: type[AbstractDTO] | None | EmptyType#: AbstractDTO to use for (de)serializing and validation of request data.

etag: ETag | None#

An etag header of type ETag to add to route handlers of this controller.

Can be overridden by route handlers.

exception_handlers: ExceptionHandlersMap | None#: A map of handler functions to status codes and/or exception types.

guards: Sequence[Guard] | None#: A sequence of Guard callables.

include_in_schema: bool | EmptyType#: A boolean flag dictating whether the route handler should be documented in the OpenAPI schema

middleware: Sequence[Middleware] | None#: A sequence of Middleware.

opt: Mapping[str, Any] | None#: A string key mapping of arbitrary values that can be accessed in Guards or wherever you have access to Request or ASGI Scope.

owner: Router#

The Router or Litestar app that owns the controller.

This value is set internally by Litestar and it should not be set when subclassing the controller.

parameters: ParametersMap | None#: A mapping of Parameter definitions available to all application paths.

request_class: type[Request] | None#: A custom subclass of Request to be used as the default request for all route handlers under the controller.

request_max_body_size: int | None | EmptyType#: Maximum allowed size of the request body in bytes. If this size is exceeded, a '413 - Request Entity Too Large' error response is returned.

response_class: type[Response] | None#: A custom subclass of Response to be used as the default response for all route handlers under the controller.

response_cookies: ResponseCookies | None#: A list of Cookie instances.

response_headers: ResponseHeaders | None#: A string keyed dictionary mapping ResponseHeader instances.

return_dto: type[AbstractDTO] | None | EmptyType#: AbstractDTO to use for serializing outbound response data.

security: Sequence[SecurityRequirement] | None#: A sequence of dictionaries that to the schema of all route handlers under the controller.

signature_namespace: dict[str, Any]#: A mapping of names to types for use in forward reference resolution during signature modelling.

signature_types: Sequence[Any]#

A sequence of types for use in forward reference resolution during signature modelling.

These types will be added to the signature namespace using their __name__ attribute.

tags: Sequence[str] | None#: A sequence of string tags that will be appended to the schema of all route handlers under the controller.

type_decoders: TypeDecodersSequence | None#: A sequence of tuples, each composed of a predicate testing for type identity and a msgspec hook for deserialization.

type_encoders: TypeEncodersMap | None#: A mapping of types to callables that transform them into types supported for serialization.

websocket_class: type[WebSocket] | None#: A custom subclass of WebSocket to be used as the default websocket for all route handlers under the controller.

AgentStreamFrame#

class litestar_mcp.agent.AgentStreamFrame[source]#

Bases: Struct

Typed SSE frame emitted during agent execution.

to_sse_message(type_encoders=None)[source]#: Encode as an SSE message whose id is seq when non-zero and whose event is event_type.

classmethod from_sse(event)[source]#

Decode an SSE event, taking seq from the event id when the payload omits it.

Return type:: AgentStreamFrame

iter_agent_stream#

async litestar_mcp.agent.iter_agent_stream(response, *, max_event_size=1048576)[source]#

Consume a streaming httpx2.Response using httpx2.EventSource and yield AgentStreamFrame objects.

Return type:: AsyncIterator[AgentStreamFrame]

MCPHttpClient#

TurnBudget#

class litestar_mcp.agent.TurnBudget[source]#

Bases: object

Configurable execution limits for a single agent turn.

__init__(max_turns=15, max_tool_calls=50, timeout_seconds=300.0)#

BudgetExceededError#

class litestar_mcp.agent.BudgetExceededError[source]#

Bases: LitestarMCPError

Raised when an agent turn exceeds its configured turn, tool-call, or timeout budget.

SpanManager#

class litestar_mcp.core.observability.SpanManager[source]#

Bases: object

Lazy OpenTelemetry span manager with graceful degradation.

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

__init__(telemetry=None)[source]#: Initialize the SpanManager with configuration and resolve trace API if enabled.

property is_enabled: bool#: Return True once OpenTelemetry spans are available and active.

start_agent_span(agent_name='', action='invoke_agent', *, agent_group=None, session_id=None, turn_id=None, model=None, provider=None, attributes=None)[source]#

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

Return type:: Any

start_tool_span(tool_name='', call_id=None, *, attributes=None)[source]#

Start a span representing a single tool execution.

Return type:: Any

set_usage(span, prompt_tokens, completion_tokens)[source]#

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

Return type:: None

start_span(name, attributes=None)[source]#

Start a generic span with custom attributes.

Return type:: Any

end_span(span, error=None)[source]#

Close a span and record errors when provided.

Return type:: None

span(name, attributes=None)[source]#

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

Return type:: Iterator[typing.Any]

use(span)[source]#

Make span current in the OpenTelemetry context during execution.

Return type:: Iterator[typing.Any]

TelemetryConfig#

class litestar_mcp.core.observability.TelemetryConfig[source]#

Bases: object

Configuration for OpenTelemetry observability in litestar-mcp.

__init__(enable_spans=False, provider_factory=None, resource_attributes=<factory>, tracer_name='litestar_mcp')#