Skip to content

MCP transport API

Most applications configure MCPServerConfig values on AsyncAIClient. Read MCP connections before using these lower-level APIs. OAuth is supported only with Streamable HTTP. Authentication and hidden identity resolution for remote tools belong to their server.

Server and OAuth configuration

seqoria_agent.client.mcp_transport.MCPServerConfig dataclass

Configuration for an MCP server connection.

Attributes:

Name Type Description
url str

The server URL or command for stdio transport. - For HTTP: "http://localhost:8000/mcp" or "https://..." - For SSE: "sse://localhost:8000/events" - For stdio: "stdio://python -m my_server" or "stdio:///path/to/server"
transport TransportType

Transport type: "http", "sse", "stdio", or "auto" (default). When "auto", the transport is inferred from the URL scheme.
headers dict[str, str] | None

Optional HTTP headers for HTTP/SSE transports.
timeout float

Connection timeout in seconds (default: 30.0).
env dict[str, str] | None

Environment variables for stdio subprocess.
oauth OAuthConfig | bool | None

OAuth configuration for Streamable HTTP servers (not SSE or stdio). - None or False: No OAuth (use headers for auth if needed) - True: Enable OAuth with default settings - OAuthConfig: Enable OAuth with custom settings

seqoria_agent.client.mcp_transport.OAuthConfig dataclass

Configuration for OAuth 2.1 authentication.

Attributes:

Name Type Description
client_name str

Human-readable name for the OAuth client (shown during consent).
redirect_uri str

URI where the authorization server redirects after consent. Default is localhost for CLI/desktop apps.
scopes list[str] | None

OAuth scopes to request. If None, uses server defaults.
storage TokenStorage | None

Token storage implementation. Defaults to InMemoryTokenStorage.
redirect_handler callable | None

Custom handler for opening the auth URL. Defaults to webbrowser.open.
callback_handler callable | None

Custom handler for receiving the callback. Automatic redirects use a loopback listener; explicit redirect URIs default to a CLI prompt.

seqoria_agent.client.mcp_transport.InMemoryTokenStorage

Bases: TokenStorage

In-memory token storage implementation for OAuth.

Stores OAuth tokens and client registration info in memory. Tokens are lost when the process exits.

For persistent storage, subclass TokenStorage and implement file-based or database-backed storage.

get_client_info() -> OAuthClientInformationFull | None async

Get stored client information.

get_tokens() -> OAuthToken | None async

Get stored tokens.

set_client_info(client_info: OAuthClientInformationFull) -> None async

Store client information.

set_tokens(tokens: OAuthToken) -> None async

Store tokens.

In-memory OAuth storage is process-local and temporary. Production token persistence, encryption, tenancy, consent, and refresh behavior need application-specific review.

Connection ownership

Use the factory or convenience context manager for a connection. The returned MCPConnection is valid only inside that context. A supplied HTTP pool is caller-owned; an internally created pool is closed on exit. Connection pools own their sessions and should be entered and closed within one application event loop.

seqoria_agent.client.mcp_transport.connect_mcp(url: str, *, transport: TransportType = 'auto', client_pool: HTTPClientPool | None = None) -> AsyncIterator[MCPConnection] async

Convenience function to connect to an MCP server.

Example

async with connect_mcp("http://localhost:8000/mcp") as conn: ... tools = await conn.session.list_tools()

seqoria_agent.client.mcp_transport.MCPTransportFactory

Factory for creating MCP transport connections.

Supports automatic transport type detection based on URL scheme: - http://, https:// -> Streamable HTTP - sse:// -> Server-Sent Events - stdio:// -> Standard I/O subprocess

Example

async with MCPTransportFactory.connect("http://localhost:8000/mcp") as conn: ... tools = await conn.session.list_tools()

connect(config: str | MCPServerConfig, *, client_pool: HTTPClientPool | None = None) -> AsyncIterator[MCPConnection] async staticmethod

Create an MCP connection with automatic transport detection.

Yields:

Type Description
MCPConnection

The active MCP connection.

detect_transport(url: str) -> TransportType staticmethod

Detect the transport type from URL scheme.

seqoria_agent.client.mcp_transport.MCPConnection dataclass

Session details valid within the transport factory context manager.

Attributes:

Name Type Description
session ClientSession

The MCP ClientSession for this connection.
url str

The original URL/command used to establish the connection.
transport_type TransportType

The resolved transport type.

seqoria_agent.client.mcp_transport.HTTPClientPool

Shared pool of httpx.AsyncClient instances for connection reuse.

This pool maintains one AsyncClient per unique (host, port) combination, enabling HTTP connection reuse (keep-alive) across multiple MCP requests.

Use within one event loop; access is serialized with asyncio.Lock.

Example

pool = HTTPClientPool() client = await pool.get_client("https://api.example.com/mcp") response = await client.get("https://api.example.com/health") await pool.close() # Cleanup when done

__init__(*, default_timeout: float = 30.0, read_timeout: float = 300.0, max_connections: int = 100, max_keepalive_connections: int = 20) -> None

Initialize the HTTP client pool.

close() -> None async

Close all pooled clients and release resources.

get_client(url: str, *, headers: Dict[str, str] | None = None, timeout: float | None = None) -> httpx.AsyncClient async

Get or create an AsyncClient for the given URL's host.

seqoria_agent.client.ai_client.connection_pool.MCPConnectionPool

Manage MCP server connections with resilience features.

The pool supports multiple transport types (HTTP, SSE, stdio) and provides: - Lazy connection establishment - Circuit breaker pattern for fault tolerance - Retry with exponential backoff - Tool routing to appropriate servers

connections: List[_ActiveConnection] property

Return a copy of the current connection list (read-only).

has_servers: bool property

Check whether any remote server connection is alive.

server_configs: List[MCPServerConfig] property

Return a copy of configured server configs (read-only).

add_server(server: str | MCPServerConfig, *, connect: bool = True) -> int async

Convenience wrapper for adding a single server. Returns 1 if added, else 0.

add_servers(servers: Iterable[str | MCPServerConfig], *, connect: bool = True) -> int async

Add server configs to the pool and optionally connect immediately.

Returns the number of new server configs added (deduped by URL).

call_tool(name: str, arguments: Dict[str, Any], *, server_args: Optional[Dict[str, Any]] = None) -> List[Any] async

Invoke a tool on the appropriate server with resilience guards.

close() -> None async

Gracefully shut down all connections.

get_session_for_url(url: str) -> Optional[ClientSession]

Get the ClientSession for a specific server URL.

list_server_urls() -> List[str]

Return configured server URLs (read-only).

remove_server(server: str | MCPServerConfig, *, disconnect: bool = True) -> int async

Convenience wrapper for removing a single server. Returns 1 if removed, else 0.

remove_servers(servers: Iterable[str | MCPServerConfig], *, disconnect: bool = True) -> int async

Remove server configs from the pool and optionally disconnect.

Returns the number of server configs removed (deduped by URL).

start() -> None async

Establish connections to all configured servers (idempotent).