	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Multi-Tenant Server with Context Isolation

MCP Multi-Tenant Server with Context Isolation

Author: Venkata Sudhakar

When a single MCP server handles requests from multiple tenants - different companies, teams, or users - you must ensure that one tenant cannot access another tenant's data. Without context isolation, a bug in the routing logic could expose tenant A's records to tenant B. A multi-tenant MCP server enforces data boundaries by scoping every tool call to a verified tenant context before touching any data store.

In this tutorial, you will build a multi-tenant MCP server where each tool call carries a tenant_id argument. The server validates the tenant_id against an allowlist and then scopes all data access to that tenant. If the tenant_id is missing or invalid, the server returns an access denied error before executing any logic.

The pattern below uses a simple tenant registry dictionary to validate incoming tenant IDs and retrieve per-tenant configuration such as the database prefix. In production, replace this registry with a lookup against Firestore or Secret Manager to keep tenant configuration outside the server code.

# multitenant_server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("multitenant-mcp-server")

TENANT_REGISTRY = {
    "tenant_acme": {"db_prefix": "acme", "plan": "enterprise"},
    "tenant_globex": {"db_prefix": "globex", "plan": "standard"},
}

def resolve_tenant(tenant_id):
    if tenant_id not in TENANT_REGISTRY:
        return None
    return TENANT_REGISTRY[tenant_id]

def scoped_query(db_prefix, resource_type, resource_id):
    return f"{db_prefix}/{resource_type}/{resource_id}"

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_customer",
            description="Get a customer record scoped to the calling tenant.",
            inputSchema={
                "type": "object",
                "properties": {
                    "tenant_id": {"type": "string"},
                    "customer_id": {"type": "string"}
                },
                "required": ["tenant_id", "customer_id"]
            }
        ),
        Tool(
            name="list_customers",
            description="List all customers for the calling tenant.",
            inputSchema={
                "type": "object",
                "properties": {
                    "tenant_id": {"type": "string"}
                },
                "required": ["tenant_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    tenant_id = arguments.get("tenant_id")
    tenant = resolve_tenant(tenant_id)
    if not tenant:
        return [TextContent(type="text", text="Access denied: invalid tenant_id")]
    prefix = tenant["db_prefix"]
    if name == "get_customer":
        path = scoped_query(prefix, "customers", arguments["customer_id"])
        result = {"path": path, "name": "Alice Smith", "plan": tenant["plan"]}
        return [TextContent(type="text", text=json.dumps(result))]
    elif name == "list_customers":
        customers = [
            {"id": "C001", "path": scoped_query(prefix, "customers", "C001")},
            {"id": "C002", "path": scoped_query(prefix, "customers", "C002")}
        ]
        return [TextContent(type="text", text=json.dumps(customers))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

The test below sends requests from two valid tenants and one invalid tenant. Each valid tenant sees only their scoped data paths. The invalid tenant receives an access denied error without any data leaking.

This pattern ensures that even if an ADK agent sends the wrong tenant_id by mistake, the server catches it before touching any data. For stronger isolation in production, validate tenant tokens using signed JWTs or service account credentials rather than a plain string match, and log every access denial to Cloud Logging for audit purposes.

Send your comments, suggestions or queries regarding this site to [email protected].