guMCP is built on a unified backend architecture that enables every server to work seamlessly with both local (stdio) and remote (SSE) transports. This design eliminates the need for separate implementations while providing flexibility in deployment models.
The key insight: servers are transport-agnostic. The same server code runs locally in Claude Desktop via stdio or remotely in Cursor via SSE without any modifications.
Servers follow a factory pattern for creation. Example from src/servers/simple-tools-server/main.py:16-24:
def create_server(user_id=None, api_key=None): server = Server("simple-tools-server") if user_id: server.user_id = user_id # Initialize user-specific state if user_id not in user_data_stores: user_data_stores[user_id] = {} # Register tools, resources, prompts @server.list_tools() async def handle_list_tools() -> list[types.Tool]: # Tool definitions return server
The factory pattern enables user-specific server instances, which is critical for multi-tenant remote deployments where each user needs isolated state.
The remote server (src/servers/remote.py:113-179) creates user-specific connections with session management:
async def handle_sse(request): session_key_encoded = request.path_params["session_key"] session_key = f"{server_name}:{session_key_encoded}" # Parse user_id and api_key from session if ":" in session_key_encoded: user_id, api_key = session_key_encoded.split(":") # Create SSE transport sse_transport = SseServerTransport( f"/{server_name}/{session_key_encoded}/messages/" ) # Reuse or create server instance for this user if session_key not in user_server_instances: server_instance = server_factory(user_id, api_key) user_server_instances[session_key] = server_instance # Run server with SSE streams async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams: await server_instance.run(streams[0], streams[1], init_options)
SSE connections maintain server instances per user session, allowing state persistence across reconnections. Instances are cleaned up when connections close.
The generic CredentialsT type allows each implementation to work with service-specific credential formats (Google’s Credentials, Slack’s AccessToken, etc.).
The LocalAuthClient is designed for single-user local development. For production deployments, implement a custom BaseAuthClient that integrates with your authentication infrastructure.
Servers that require authentication use the factory-created auth client:
from src.auth.factory import create_auth_clientdef create_server(user_id=None, api_key=None): server = Server("authenticated-server") # Create auth client (automatically selects implementation) auth_client = create_auth_client(api_key=api_key) @server.call_tool() async def handle_call_tool(name: str, arguments: dict | None): # Get credentials for this user credentials = auth_client.get_user_credentials("service-name", user_id) if not credentials: raise ValueError(f"No credentials found for user {user_id}") # Use credentials to call service API # ... return server
For servers that maintain state, use user-specific storage:
user_data_stores = {} # Global storedef create_server(user_id=None, api_key=None): server = Server("stateful-server") if user_id and user_id not in user_data_stores: user_data_stores[user_id] = {} @server.call_tool() async def handle_call_tool(name: str, arguments: dict | None): current_user = getattr(server, "user_id", None) data_store = user_data_stores.get(current_user, {}) # Use user-specific data store # ... return server
For remote deployments, be mindful of memory usage when storing user state in-memory. Consider using a database or cache for production workloads.