OAuth Adapters
Configure token storage with memory, file, or cloud adapters.
OAuth Adapters
OAuth adapters handle storage and retrieval of OAuth tokens for upstream MCP servers that require authentication (e.g., GitHub, Gmail, Slack).
Available Adapters
| Adapter | Use Case | Persistence | Multi-Process |
|---|---|---|---|
memory_adapter | Testing, short-lived processes | No | No |
file_adapter | CLI tools, local development | Yes | Limited |
cloud_adapter | Production, multi-instance | Yes | Yes |
Memory Adapter
Stores tokens in memory. Tokens are lost when the process exits.
from routermcp import RouterMCPClient
from routermcp.oauth import memory_adapter
from routermcp.types import OAuthConfig
client = RouterMCPClient(
url='https://gateway.routermcp.com/v1/mcp/my-project',
oauth=OAuthConfig(adapter=memory_adapter()),
)Only use the memory adapter for testing or short-lived processes. Tokens will be lost on restart.
File Adapter
Stores tokens as JSON files on disk. Compatible with the RouterMCP CLI's token storage.
from routermcp import RouterMCPClient
from routermcp.oauth import file_adapter
from routermcp.types import OAuthConfig
# Default: ~/.routermcp/mcp-oauth/
client = RouterMCPClient(
url='https://gateway.routermcp.com/v1/mcp/my-project',
oauth=OAuthConfig(adapter=file_adapter()),
)
# Custom path
client = RouterMCPClient(
url='https://gateway.routermcp.com/v1/mcp/my-project',
oauth=OAuthConfig(adapter=file_adapter('/path/to/tokens')),
)File Structure
Tokens are stored as:
~/.routermcp/mcp-oauth/
{user_id}/
{server_id}.jsonThe file adapter is compatible with the RouterMCP CLI. Tokens authorized via the CLI will be available to the SDK and vice versa.
Cloud Adapter
Stores tokens in RouterMCP Cloud via API. Ideal for production deployments with multiple instances.
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',
oauth=OAuthConfig(
adapter=cloud_adapter(api_key='rmc_...'),
),
)
# Custom API endpoint
client = RouterMCPClient(
url='https://gateway.routermcp.com/v1/mcp/my-project',
oauth=OAuthConfig(
adapter=cloud_adapter(
api_key='rmc_...',
base_url='https://custom.routermcp.com',
),
),
)Multi-User OAuth
For applications serving multiple users, each user needs their own OAuth tokens:
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='default-user',
oauth=OAuthConfig(
adapter=cloud_adapter(api_key='rmc_...'),
auto_refresh=True,
),
)
await client.connect()
async def handle_user_request(user_id: str, tool_name: str, args: dict):
"""Handle a request for a specific user."""
session = client.create_session(user_id=user_id)
await session.connect()
try:
result = await session.call_tool(tool_name, args)
return result
finally:
await session.disconnect()
# Each user's OAuth tokens are stored separately
await handle_user_request('user-1', 'gmail_send', {'to': 'a@test.com', 'body': 'Hi'})
await handle_user_request('user-2', 'slack_post', {'channel': '#general', 'text': 'Hello'})OAuth Flow
When a tool requires OAuth and no valid token exists:
- The tool call returns an OAuth authorization URL
- Your application redirects the user to this URL
- User authorizes the application
- RouterMCP stores the tokens via your configured adapter
- Subsequent tool calls use the stored tokens
from routermcp.types import OAuthError
try:
result = await client.call_tool('gmail_send', {'to': 'test@example.com'})
except OAuthError as e:
if hasattr(e, 'authorization_url'):
# Redirect user to e.authorization_url
print(f'Please authorize: {e.authorization_url}')Custom Adapters
Implement the OAuthAdapter protocol for custom storage:
from typing import Protocol
from routermcp.types import OAuthTokens
class OAuthAdapter(Protocol):
async def get_tokens(self, user_id: str, server_id: str) -> OAuthTokens | None:
...
async def set_tokens(self, user_id: str, server_id: str, tokens: OAuthTokens) -> None:
...
async def delete_tokens(self, user_id: str, server_id: str) -> None:
...
class OAuthTokens(BaseModel):
access_token: str
refresh_token: str | None = None
expires_at: int | None = None
token_type: str = 'Bearer'Example with Redis:
import json
import redis.asyncio as redis
from routermcp.types import OAuthTokens
class RedisOAuthAdapter:
def __init__(self, redis_url: str):
self.client = redis.from_url(redis_url)
async def get_tokens(self, user_id: str, server_id: str) -> OAuthTokens | None:
key = f'oauth:{user_id}:{server_id}'
data = await self.client.get(key)
if data:
return OAuthTokens.model_validate_json(data)
return None
async def set_tokens(self, user_id: str, server_id: str, tokens: OAuthTokens) -> None:
key = f'oauth:{user_id}:{server_id}'
await self.client.set(key, tokens.model_dump_json())
async def delete_tokens(self, user_id: str, server_id: str) -> None:
key = f'oauth:{user_id}:{server_id}'
await self.client.delete(key)
def redis_adapter(redis_url: str) -> RedisOAuthAdapter:
return RedisOAuthAdapter(redis_url)
# Usage
from routermcp import RouterMCPClient
from routermcp.types import OAuthConfig
client = RouterMCPClient(
url='https://gateway.routermcp.com/v1/mcp/my-project',
oauth=OAuthConfig(adapter=redis_adapter('redis://localhost:6379')),
)