In-process stdio#

class litestar_mcp.mcp.stdio.ASGIStreamingTransport[source]#

Bases: AsyncBaseTransport

Run an ASGI application in-process and stream its response body to httpx.

httpx.ASGITransport buffers every http.response.body chunk until the application returns, so Server-Sent Events never reach the client while the stream is open. This transport returns the response as soon as http.response.start arrives, hands each body chunk to the response stream as it is sent, and cancels the application task when the response is closed, or when the caller is cancelled before http.response.start arrives, so an aborted request observes http.disconnect and task cancellation. The httpx read timeout bounds the wait for http.response.start and for each body chunk (httpx.ReadTimeout).

Parameters:

app¶ (Callable[[Union[HTTPScope, WebSocketScope], Callable[..., Awaitable[Union[HTTPRequestEvent, HTTPDisconnectEvent, WebSocketConnectEvent, WebSocketReceiveEvent, WebSocketDisconnectEvent]]], Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent]], Awaitable[None]]], Awaitable[None]]) -- The ASGI application to call for every request.
client¶ (tuple[str, int]) -- The scope["client"] tuple presented to the application.
root_path¶ (str) -- The scope["root_path"] presented to the application.
shutdown_timeout¶ (float) -- Positive finite seconds allowed for application cleanup. Expiry is logged; cancellation-resistant application code may outlive the response.

__init__(app, *, client=('mcp-stdio', 0), root_path='', shutdown_timeout=5.0)[source]#

async handle_async_request(request)[source]#

Dispatch request to the application and return a streaming response.

Return type:: Response

class litestar_mcp.mcp.stdio.MCPStdioContext[source]#

Bases: object

Runtime identity context for standalone MCP stdio transports.

__init__(client_id='stdio', owner_id=None, user=None, auth=None, session=None, state=None)#

async litestar_mcp.mcp.stdio.run_stdio_async(app, *, stdio_context=None, headers=None, token_provider=None, header_name='Authorization', token_prefix='Bearer ', sse_read_timeout=300.0, stdin=None, stdout=None, stderr=None, max_message_size=16777216, shutdown_timeout=5.0)[source]#

Serve a Litestar MCP endpoint to a local stdio client without a socket.

shutdown_timeout bounds in-process request cleanup and application shutdown after native lifespan entry succeeds. Bridge errors and body cancellation are preserved if that shutdown fails or times out.

Startup and its unwind follow Litestar's native semantics, including exception grouping and chaining. Application startup, lifespan, and shutdown hooks must bound and shield their own cleanup where needed.

litestar_mcp.mcp.stdio.run_stdio(app, *, stdio_context=None, headers=None, token_provider=None, header_name='Authorization', token_prefix='Bearer ', sse_read_timeout=300.0, stdin=None, stdout=None, stderr=None, max_message_size=16777216, shutdown_timeout=5.0)[source]#: Synchronously run run_stdio_async(); KeyboardInterrupt exits 0.