nat.plugins.mcp.client.client_base#

Attributes#

logger

Classes#

`AuthAdapter`	httpx.Auth adapter for authentication providers.
`MCPBaseClient`	Base client for creating a MCP transport session and connecting to an MCP server
`MCPSSEClient`	Client for creating a session and connecting to an MCP server using SSE
`MCPStdioClient`	Client for creating a session and connecting to an MCP server using stdio.
`MCPStreamableHTTPClient`	Client for creating a session and connecting to an MCP server using streamable-http
`MCPToolClient`	Client wrapper used to call an MCP tool. This assumes that the MCP transport session

Module Contents#

logger#

class AuthAdapter( auth_provider: nat.authentication.interfaces.AuthProviderBase, user_id: str | None = None, )#

Bases: httpx.Auth

httpx.Auth adapter for authentication providers. Converts AuthProviderBase to httpx.Auth interface for dynamic token management.

auth_provider#

user_id = None#

_lock#

is_authenticating = False#

async async_auth_flow( request: httpx.Request, ) → collections.abc.AsyncGenerator[httpx.Request, httpx.Response]#: Add authentication headers to the request using NAT auth provider.

async _get_auth_headers( request: httpx.Request | None = None, response: httpx.Response | None = None, ) → dict[str, str]#: Get authentication headers from the NAT auth provider.

class MCPBaseClient( transport: str = 'streamable-http', auth_provider: nat.authentication.interfaces.AuthProviderBase | None = None, user_id: str | None = None, tool_call_timeout: datetime.timedelta = timedelta(seconds=60), auth_flow_timeout: datetime.timedelta = timedelta(seconds=300), reconnect_enabled: bool = True, reconnect_max_attempts: int = 2, reconnect_initial_backoff: float = 0.5, reconnect_max_backoff: float = 50.0, )#

Bases: abc.ABC

Base client for creating a MCP transport session and connecting to an MCP server

Args:: transport (str): The type of client to use (‘sse’, ‘stdio’, or ‘streamable-http’) auth_provider (AuthProviderBase | None): Optional authentication provider for Bearer token injection tool_call_timeout (timedelta): Timeout for tool calls when authentication is not required auth_flow_timeout (timedelta): Extended timeout for tool calls that may require interactive authentication reconnect_enabled (bool): Whether to automatically reconnect on connection failures reconnect_max_attempts (int): Maximum number of reconnection attempts reconnect_initial_backoff (float): Initial backoff delay in seconds for reconnection attempts reconnect_max_backoff (float): Maximum backoff delay in seconds for reconnection attempts

_tools = None#

_transport = ''#

_exit_stack: contextlib.AsyncExitStack | None = None#

_session: mcp.ClientSession | None = None#

_connection_established = False#

_initial_connection = False#

_lifecycle_task: asyncio.Task | None = None#

_lifecycle_commands: asyncio.Queue[tuple[str, asyncio.Future[None]]] | None = None#

_auth_provider = None#

_httpx_auth#

_tool_call_timeout#

_auth_flow_timeout#

_reconnect_enabled = True#

_reconnect_max_attempts = 2#

_reconnect_initial_backoff = 0.5#

_reconnect_max_backoff = 50.0#

_reconnect_lock: asyncio.Lock#

property auth_provider: nat.authentication.interfaces.AuthProviderBase | None#

property transport: str#

property is_connected: bool#: Whether the client has an active, initialized connection.

property server_name#: Provide server name for logging

abstractmethod connect_to_server() → collections.abc.AsyncGenerator[mcp.ClientSession, None]#

Async:

Establish a session with an MCP server within an async context

async _reconnect()#: Attempt to reconnect by tearing down and re-establishing the session.

_ensure_lifecycle_worker() → asyncio.Task#

Respawn the lifecycle worker if it died, so the client can recover (#2111).

A failing transport background child task cancels the worker that hosts the transport’s AnyIO task group. That cancellation bypasses the worker’s command loop, so it never clears connection state and the task is left done. Rather than bricking the client for the rest of the process, drop the orphaned transport references and start a fresh worker that the next command can drive to reconnect.

Returns the live worker task, so callers can watch it alongside their command.

async _run_lifecycle_command(command: str) → None#: Run a connection lifecycle command in the task that owns the transport stack.

async _lifecycle_worker() → None#: Own MCP transport context entry and exit to keep AnyIO cancel scopes task-local.

async _connect_connection() → None#: Enter the MCP transport context in the lifecycle task.

async _close_connection() → None#: Exit the MCP transport context in the same task that entered it.

async _with_reconnect(coro)#: Execute an awaited operation, reconnecting once on errors. Does not reconnect if the error occurs during an active authentication flow.

async _has_cached_auth_token() → bool#

Check if we have a cached, non-expired authentication token.

Returns:: bool: True if we have a valid cached token, False if authentication may be needed

async _get_tool_call_timeout() → datetime.timedelta#

Determine the appropriate timeout for a tool call based on authentication state.

Returns:: timedelta: auth_flow_timeout if authentication may be needed, tool_call_timeout otherwise

async get_tools() → dict[str, MCPToolClient]#: Retrieve a dictionary of all tools served by the MCP server. Uses unauthenticated session for discovery.

async get_tool(tool_name: str) → MCPToolClient#

Get an MCP Tool by name.

Args:: tool_name (str): Name of the tool to load.
Returns:: MCPToolClient for the configured tool.
Raises:: MCPToolNotFoundError: If no tool is available with that name.

set_user_auth_callback( auth_callback: collections.abc.Callable[[nat.authentication.interfaces.AuthFlowType], nat.authentication.interfaces.AuthenticatedContext], )#: Set the user authentication callback.

async call_tool(tool_name: str, tool_args: dict | None)#

class MCPSSEClient( url: str, tool_call_timeout: datetime.timedelta = timedelta(seconds=60), auth_flow_timeout: datetime.timedelta = timedelta(seconds=300), reconnect_enabled: bool = True, reconnect_max_attempts: int = 2, reconnect_initial_backoff: float = 0.5, reconnect_max_backoff: float = 50.0, )#

Bases: MCPBaseClient

Client for creating a session and connecting to an MCP server using SSE

Args:: url (str): The url of the MCP server

_url#

property url: str#

property server_name#: Provide server name for logging

async connect_to_server()#: Establish a session with an MCP SSE server within an async context

class MCPStdioClient( command: str, args: list[str] | None = None, env: dict[str, str] | None = None, tool_call_timeout: datetime.timedelta = timedelta(seconds=60), auth_flow_timeout: datetime.timedelta = timedelta(seconds=300), reconnect_enabled: bool = True, reconnect_max_attempts: int = 2, reconnect_initial_backoff: float = 0.5, reconnect_max_backoff: float = 50.0, )#

Bases: MCPBaseClient

Client for creating a session and connecting to an MCP server using stdio. This is a local transport that spawns the MCP server process and communicates with it over stdin/stdout.

Args:: command (str): The command to run args (list[str] | None): Additional arguments for the command env (dict[str, str] | None): Environment variables to set for the process

_command#

_args = None#

_env = None#

property command: str#

property server_name#: Provide server name for logging

property args: list[str] | None#

property env: dict[str, str] | None#

async connect_to_server()#: Establish a session with an MCP server via stdio within an async context

class MCPStreamableHTTPClient( url: str, auth_provider: nat.authentication.interfaces.AuthProviderBase | None = None, user_id: str | None = None, custom_headers: dict[str, str] | None = None, tool_call_timeout: datetime.timedelta = timedelta(seconds=60), auth_flow_timeout: datetime.timedelta = timedelta(seconds=300), reconnect_enabled: bool = True, reconnect_max_attempts: int = 2, reconnect_initial_backoff: float = 0.5, reconnect_max_backoff: float = 50.0, )#

Bases: MCPBaseClient

Client for creating a session and connecting to an MCP server using streamable-http

Args:: url (str): The url of the MCP server auth_provider (AuthProviderBase | None): Optional authentication provider for Bearer token injection user_id (str | None): Optional user ID for session isolation custom_headers (dict[str, str] | None): Optional custom HTTP headers to include in requests tool_call_timeout (timedelta): Timeout for tool calls auth_flow_timeout (timedelta): Extended timeout for interactive authentication reconnect_enabled (bool): Whether to automatically reconnect on connection failures reconnect_max_attempts (int): Maximum number of reconnection attempts reconnect_initial_backoff (float): Initial backoff delay in seconds reconnect_max_backoff (float): Maximum backoff delay in seconds

_url#

_custom_headers#

_get_mcp_session_id: collections.abc.Callable[[], str | None] | None = None#

property url: str#

property custom_headers: dict[str, str]#: Returns the custom headers configured for this client.

property mcp_session_id: str | None#

Returns the MCP transport-level session ID if available.

This is the session ID assigned by the MCP server (from the mcp-session-id header), which can be used for correlating backend sessions with MCP server sessions.

Returns:: The MCP session ID string, or None if not connected or not available.

property server_name#: Provide server name for logging

async connect_to_server()#: Establish a session with an MCP server via streamable-http within an async context

class MCPToolClient( session: mcp.ClientSession, parent_client: MCPBaseClient, tool_name: str, tool_description: str | None, tool_input_schema: dict | None = None, )#

Client wrapper used to call an MCP tool. This assumes that the MCP transport session has already been setup.

Args:: session (ClientSession): The MCP client session tool_name (str): The name of the tool to wrap tool_description (str): The description of the tool provided by the MCP server. tool_input_schema (dict): The input schema for the tool. parent_client (MCPBaseClient): The parent MCP client for auth management.

_session#

_tool_name#

_tool_description#

_input_schema = None#

_parent_client#

property name#: Returns the name of the tool.

property description#: Returns the tool’s description. If none was provided. Provides a simple description using the tool’s name

property input_schema#: Returns the tool’s input_schema.

set_description(description: str)#: Manually define the tool’s description using the provided string.

async acall(tool_args: dict) → str#

Call the MCP tool with the provided arguments. Session context is now handled at the client level, eliminating the need for metadata injection.

Args:: tool_args (dict[str, Any]): A dictionary of key value pairs to serve as inputs for the MCP tool.