RouterMCP
SDKsPython

Client API

Complete API reference for RouterMCPClient, SyncRouterMCPClient, and RouterMCPSession.

Client API

The Python SDK provides three main classes:

  • RouterMCPClient - Async client for high-concurrency applications
  • SyncRouterMCPClient - Sync client for scripts and CLI tools
  • RouterMCPSession - Session for multi-user scenarios

RouterMCPClient

Constructor

from routermcp import RouterMCPClient
from routermcp.types import OAuthConfig

client = RouterMCPClient(
    url: str | None = None,           # RouterMCP server URL
    api_key: str | None = None,       # API key for authentication
    user_id: str | None = None,       # User ID for OAuth token storage
    timeout: int | None = None,       # Request timeout in ms (default: 30000)
    headers: dict[str, str] | None = None,  # Custom HTTP headers
    oauth: OAuthConfig | None = None, # OAuth adapter configuration
    debug: bool | None = None,        # Enable debug logging
)

Context Manager

async with RouterMCPClient(url='...') as client:
    # Client is connected
    tools = await client.list_tools()
# Client is automatically disconnected

connect()

Initialize the connection and perform MCP handshake.

capabilities = await client.connect()
# Returns: InitializeResult with server capabilities

disconnect()

Close the connection and clean up resources.

await client.disconnect()

list_tools()

List all available tools from aggregated MCP servers.

tools = await client.list_tools()

for tool in tools:
    print(f'{tool.name}: {tool.description}')
    print(f'Input schema: {tool.inputSchema}')

Returns: list[Tool]

class Tool(BaseModel):
    name: str
    description: str | None
    inputSchema: ToolInputSchema

class ToolInputSchema(BaseModel):
    type: str = 'object'
    properties: dict[str, Any]
    required: list[str]

call_tool()

Execute a tool with the given arguments.

result = await client.call_tool('tool_name', {
    'arg1': 'value1',
    'arg2': 123,
})

# Access result content
for item in result.content:
    if item.get('type') == 'text':
        print(item.get('text'))

Returns: ToolResult

class ToolResult(BaseModel):
    content: list[dict[str, Any]]
    isError: bool = False

call_tool_typed()

Execute a tool and parse the result with a Pydantic model.

from pydantic import BaseModel

class User(BaseModel):
    id: str
    name: str
    email: str

user = await client.call_tool_typed('get_user', {'id': '123'}, User)
# user is typed as User

list_resources()

List available resources.

resources = await client.list_resources()

for resource in resources:
    print(f'{resource.uri}: {resource.name}')

read_resource()

Read a resource by URI.

content = await client.read_resource('file:///path/to/file.txt')

list_prompts()

List available prompts.

prompts = await client.list_prompts()

for prompt in prompts:
    print(f'{prompt.name}: {prompt.description}')

get_prompt()

Get a prompt with arguments.

result = await client.get_prompt('summarize', {
    'text': 'Long text to summarize...',
})

create_session()

Create a new session for a different user.

session = client.create_session(user_id='user-456')
await session.connect()
# Use session for user-456's requests
await session.disconnect()

SyncRouterMCPClient

Synchronous wrapper with the same API as RouterMCPClient.

from routermcp import RouterMCPClient

async with RouterMCPClient(url='...') as client:
    tools = await client.list_tools()
    result = await client.call_tool('tool_name', {'arg': 'value'})
from routermcp import SyncRouterMCPClient

with SyncRouterMCPClient(url='...') as client:
    tools = client.list_tools()
    result = client.call_tool('tool_name', {'arg': 'value'})

Methods

All methods are synchronous versions of RouterMCPClient:

  • connect() / disconnect()
  • list_tools() / call_tool()
  • list_resources() / read_resource()
  • list_prompts() / get_prompt()

RouterMCPSession

Sessions are created via client.create_session() and provide the same API as the client but with a different user context.

from routermcp import RouterMCPClient
from routermcp.oauth import cloud_adapter
from routermcp.types import OAuthConfig

client = RouterMCPClient(
    url='https://gateway.routermcp.com/v1/mcp/my-project',
    user_id='admin',
    oauth=OAuthConfig(adapter=cloud_adapter(api_key='key')),
)

await client.connect()

# Create sessions for different users
user1_session = client.create_session(user_id='user-1')
user2_session = client.create_session(user_id='user-2')

await user1_session.connect()
await user2_session.connect()

# Each session uses its own OAuth tokens
await user1_session.call_tool('gmail_send', {'to': 'a@example.com', 'body': 'Hi'})
await user2_session.call_tool('gmail_send', {'to': 'b@example.com', 'body': 'Hello'})

await user1_session.disconnect()
await user2_session.disconnect()
await client.disconnect()

Sessions share the transport configuration with the parent client but maintain separate MCP sessions and OAuth token contexts.

Error Handling

from routermcp import RouterMCPClient
from routermcp.types import MCPError, TransportError, ConfigError

try:
    async with RouterMCPClient(url='...') as client:
        await client.call_tool('unknown_tool', {})
except TransportError as e:
    print(f'Network error: {e} (code: {e.code})')
except MCPError as e:
    print(f'MCP error: {e}')

Error Types

ErrorDescription
MCPErrorBase error for all MCP operations
TransportErrorHTTP or network-level errors
ConfigErrorConfiguration validation errors
OAuthErrorOAuth flow or token errors

On this page