Secure Multi-Tenant MCP Servers with Python SDK 2.3.0: OAuth, Scopes & Tenant Isolation (2026)

Reviewed: 3 November 2026. A remote MCP server can reject anonymous requests, validate OAuth tokens, and still expose one customer’s records to another customer. Authentication answers “who sent this request?” It does not automatically answer “which tenant’s rows may this identity access?”

This guide builds that missing boundary. You will use the official MCP Python SDK 2.3.0 as an OAuth-protected resource server, validate tokens through an authorization server’s introspection endpoint, check the token’s issuer and audience, read a trusted tenant_id claim, and enforce tenant isolation inside every database query. The example uses SQLite to keep the code small; the same authorization pattern applies to PostgreSQL and production services.

Who this is for: Python developers exposing internal APIs, documents, tickets, CRM records, or business operations through remote MCP tools—especially SaaS teams serving multiple customers from one MCP endpoint.

1. The architecture: three checks, not one

Keep these three decisions separate:

• Authentication: Is this access token valid, active, and issued by the identity provider we trust?
• Resource authorization: Was the token issued for this exact MCP resource, and does it carry the broad permission needed to enter?
• Tenant authorization: Which tenant and user does the token represent, and are all data operations constrained to that tenant?

The request path looks like this:

MCP client
   |
   | Authorization: Bearer <access_token>
   v
MCP resource server (Python SDK)
   |-- TokenVerifier: introspect token
   |-- Validate issuer, expiry, audience, scopes
   |-- Read trusted tenant_id and subject claims
   v
MCP tool handler
   |-- Check tool-specific scope
   |-- Derive tenant from authenticated token
   v
Database
   |-- WHERE tenant_id = authenticated_tenant
   v
Only that tenant's records

The client must never be allowed to choose its own tenant by passing tenant_id as a tool argument. A model can generate or alter tool arguments. Tenant identity must come from a trusted, validated identity claim and be applied again at the data-access boundary.

2. Why the 2026 MCP authorization model matters

The current MCP specification revision is 2026-07-28. It separates the authorization server (AS), which signs users in and issues tokens, from the MCP resource server (RS), which validates tokens and protects tools. The Python SDK’s current v2 API uses MCPServer; older examples using the previous FastMCP import may not match this version.

The MCP server does not need to implement a login screen or issue user tokens. It needs a trusted verifier. The SDK can publish OAuth Protected Resource Metadata under the well-known route and return a 401 challenge that tells compatible clients where to discover authorization information. That is the job of RFC 9728.

The 2026-07-28 specification also hardens authorization: it calls for issuer validation in the authorization response, binds client credentials to the issuer that minted them, and formally moves away from Dynamic Client Registration (DCR) toward Client ID Metadata Documents (CIMD). DCR remains for backward compatibility, but do not design a new authorization deployment assuming DCR is the only registration mechanism.

3. Prerequisites

• Python 3.11 or newer is recommended.
• uv or another Python environment manager.
• MCP Python SDK 2.3.0.
• An OAuth 2.1-compatible authorization server such as Keycloak, Auth0, or Microsoft Entra ID, configured to issue access tokens for your MCP resource.
• An introspection endpoint accessible to the MCP server over HTTPS in production.
• A custom token claim named tenant_id (or your own consistent tenant claim name), plus a subject claim identifying the user.

Install the dependencies:

mkdir tenant-mcp
cd tenant-mcp
uv init
uv add "mcp[cli]==2.3.0" "httpx2>=2.10.0" aiosqlite

Pinning the SDK makes this example reproducible. The official PyPI release history lists 2.3.0 as released on 2 October 2026. Check the release page before upgrading because the SDK evolves alongside the MCP protocol.

4. Configure the identity provider first

Before writing the MCP server, configure the authorization server to issue a token containing the claims your resource server will verify. At minimum, the introspection response used by this example must provide:

tenant_id is an application-specific claim, not a standard MCP claim. Configure it in your identity provider from a trusted user or organization membership record. Do not copy it from an arbitrary login form field or a client-supplied MCP argument.

Register the MCP resource as a distinct audience/resource. The resource indicator defined by RFC 8707 exists to identify where a token may be redeemed; scopes describe what access is allowed. Do not use a broad audience such as api for every service. Use the exact public MCP URL, for example https://mcp.example.com/mcp, and make the token issuer return that resource in aud.

Create separate permissions such as:

• mcp:access — may reach the MCP service.
• documents:read — may list and read documents in the caller’s tenant.
• documents:write — may create or update documents in the caller’s tenant.

Finally, create a confidential client for the MCP resource server to authenticate to the introspection endpoint. These are server-to-server credentials; they are not the user’s bearer token. Store them in environment variables or a secrets manager. In this design, mcp:access is the SDK-wide gate; documents:read and documents:write are app-level scopes checked by the tool handlers. Ensure your client obtains those app-level scopes from the identity provider even though they are not listed as globally required by AuthSettings.

5. Add server configuration

Set these environment variables. The resource URL must be the exact URL clients use, not an internal Docker hostname or an unproxied container address.

export MCP_RESOURCE_URL="https://mcp.example.com/mcp"
export OAUTH_ISSUER_URL="https://login.example.com/"
export OAUTH_INTROSPECTION_URL="https://login.example.com/oauth2/introspect"
export OAUTH_INTROSPECTION_CLIENT_ID="mcp-resource-server"
export OAUTH_INTROSPECTION_CLIENT_SECRET="replace-with-a-secret"
export DATABASE_PATH="./tenant_docs.db"

For local development, use a loopback URL and a development identity provider. Do not send bearer tokens or introspection credentials over plain HTTP across a network.

6. Implement a real token verifier

The SDK defines a small TokenVerifier protocol: one asynchronous method receives the bearer token and returns an AccessToken if it is valid, or None if it is not. This example uses RFC 7662 token introspection so the resource server can ask the authorization server whether a token is active.

Create server.py:

import logging
import os
from typing import Any

import aiosqlite
import httpx2
from pydantic import AnyHttpUrl

from mcp.server import MCPServer
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver.exceptions import ToolError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("tenant-mcp")

RESOURCE_URL = os.environ["MCP_RESOURCE_URL"]
ISSUER_URL = os.environ["OAUTH_ISSUER_URL"]
INTROSPECTION_URL = os.environ["OAUTH_INTROSPECTION_URL"]
INTROSPECTION_CLIENT_ID = os.environ["OAUTH_INTROSPECTION_CLIENT_ID"]
INTROSPECTION_CLIENT_SECRET = os.environ["OAUTH_INTROSPECTION_CLIENT_SECRET"]
DATABASE_PATH = os.getenv("DATABASE_PATH", "tenant_docs.db")


class IntrospectionVerifier(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:
        timeout = httpx2.Timeout(10.0, connect=5.0)

        try:
            async with httpx2.AsyncClient(
                timeout=timeout,
                verify=True,
            ) as client:
                response = await client.post(
                    INTROSPECTION_URL,
                    data={
                        "token": token,
                        "client_id": INTROSPECTION_CLIENT_ID,
                        "client_secret": INTROSPECTION_CLIENT_SECRET,
                    },
                    headers={"Accept": "application/json"},
                )

            if response.status_code != 200:
                logger.warning(
                    "Token introspection returned HTTP %s",
                    response.status_code,
                )
                return None

            data: dict[str, Any] = response.json()

            if data.get("active") is not True:
                return None

            # Pin the issuer. Do not accept a token from another realm/tenant.
            if data.get("iss") != ISSUER_URL:
                return None

            # RFC 8707 audience may be a string or an array.
            audience = data.get("aud")
            audiences = audience if isinstance(audience, list) else [audience]
            if RESOURCE_URL not in audiences:
                return None

            tenant_id = data.get("tenant_id")
            subject = data.get("sub")
            expires_at = data.get("exp")
            raw_scope = data.get("scope", "")

            if not isinstance(tenant_id, str) or not tenant_id.strip():
                return None
            if not isinstance(subject, str) or not subject.strip():
                return None
            if not isinstance(expires_at, (int, float)):
                return None
            if not isinstance(raw_scope, str):
                return None

            return AccessToken(
                token=token,
                client_id=str(data.get("client_id") or "unknown"),
                scopes=raw_scope.split(),
                expires_at=int(expires_at),
                resource=RESOURCE_URL,
                subject=subject,
                claims=data,
            )

        except Exception:
            # Never log the bearer token or the raw introspection response.
            logger.exception("Token introspection failed")
            return None


mcp = MCPServer(
    "Tenant Documents",
    instructions="Read and create documents inside the authenticated tenant only.",
    token_verifier=IntrospectionVerifier(),
    auth=AuthSettings(
        issuer_url=AnyHttpUrl(ISSUER_URL),
        resource_server_url=AnyHttpUrl(RESOURCE_URL),
        required_scopes=["mcp:access"],
        validate_token_resource=True,
    ),
)

There are several deliberate checks here:

1. Active status: an expired or revoked token must not pass just because its string is well-formed.
2. Issuer pinning: a token from an unrelated realm or identity provider is rejected.
3. Audience validation: the token must name this MCP resource. A token for your billing API is not automatically valid for your MCP endpoint.
4. Tenant and subject presence: a token without a trusted tenant assignment or user identity is rejected.
5. Expiry: the expiry is returned in AccessToken as well, allowing the SDK to enforce it.

The code sends client credentials as form fields because that is supported by the reference integration pattern. Some identity providers require HTTP Basic client authentication instead. Follow your provider’s introspection documentation; do not blindly copy the authentication method. Also note that this self-contained example creates an HTTP client for each verification. For a high-throughput service, reuse a pooled httpx2.AsyncClient through the server lifecycle and close it cleanly at shutdown.

7. Derive the tenant from the authenticated request

Now create one helper that every tool uses. It reads the access token from the SDK’s per-request authentication context, checks the tool-specific scope, and obtains the tenant ID and subject from the validated token claims.

def require_principal(required_scope: str) -> tuple[str, str]:
    token = get_access_token()

    if token is None:
        raise ToolError("Authentication is required.")

    if required_scope not in token.scopes:
        raise ToolError("Your account does not have permission for this action.")

    claims = token.claims or {}
    tenant_id = claims.get("tenant_id")
    subject = token.subject

    if not isinstance(tenant_id, str) or not tenant_id:
        raise ToolError("Your account has no tenant assignment.")
    if not subject:
        raise ToolError("The authenticated user identity is unavailable.")

    return tenant_id, subject

Do not accept tenant_id in the tool signature. The model should not be able to call list_documents(tenant_id="another-company"). Even if a caller guesses another tenant’s ID, the server must ignore that value and use the authenticated principal instead.

8. Put tenant isolation in every database query

For this example, create a small SQLite table. In production, the same predicate belongs in your repository layer, ORM query builder, or database row-level security policy.

async def initialize_db() -> None:
    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                tenant_id TEXT NOT NULL,
                owner_subject TEXT NOT NULL,
                title TEXT NOT NULL,
                body TEXT NOT NULL
            )
        """)
        await db.execute("""
            CREATE INDEX IF NOT EXISTS idx_documents_tenant_id_id
            ON documents (tenant_id, id)
        """)
        await db.commit()


@mcp.tool()
async def list_documents() -> list[dict[str, Any]]:
    """List recent documents belonging to the authenticated tenant."""
    tenant_id, _subject = require_principal("documents:read")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        async with db.execute(
            """
            SELECT id, title
            FROM documents
            WHERE tenant_id = ?
            ORDER BY id DESC
            LIMIT 50
            """,
            (tenant_id,),
        ) as cursor:
            rows = await cursor.fetchall()

    return [dict(row) for row in rows]


@mcp.tool()
async def get_document(document_id: int) -> dict[str, Any]:
    """Read one document, but only if it belongs to the authenticated tenant."""
    tenant_id, _subject = require_principal("documents:read")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        async with db.execute(
            """
            SELECT id, title, body
            FROM documents
            WHERE tenant_id = ? AND id = ?
            """,
            (tenant_id, document_id),
        ) as cursor:
            row = await cursor.fetchone()

    if row is None:
        # Do not reveal whether the ID exists in another tenant.
        raise ToolError("Document not found.")

    return dict(row)


@mcp.tool()
async def create_document(title: str, body: str) -> dict[str, Any]:
    """Create a document inside the authenticated tenant."""
    tenant_id, subject = require_principal("documents:write")

    if not title.strip() or len(title) > 160:
        raise ToolError("Title must contain 1 to 160 characters.")
    if not body.strip() or len(body) > 20000:
        raise ToolError("Body must contain 1 to 20000 characters.")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        cursor = await db.execute(
            """
            INSERT INTO documents (tenant_id, owner_subject, title, body)
            VALUES (?, ?, ?, ?)
            """,
            (tenant_id, subject, title.strip(), body),
        )
        await db.commit()
        document_id = cursor.lastrowid

    logger.info(
        "Document created tenant=%s subject=%s document_id=%s",
        tenant_id,
        subject,
        document_id,
    )
    return {"id": document_id, "title": title.strip()}


if __name__ == "__main__":
    import asyncio

    asyncio.run(initialize_db())
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=8000,
    )

Notice that all three database operations use the authenticated tenant_id in the SQL predicate. The create operation also assigns the tenant and owner on the server. There is no path where the model supplies an organization ID and the server trusts it.

9. Test the authorization boundary—not only the tool function

Start the server behind a TLS reverse proxy in production. For local development, run the configured service on loopback and make sure the resource URL in the identity provider exactly matches the URL used by the MCP client.

First, check the protected-resource metadata route documented by the SDK:

curl -i "https://mcp.example.com/.well-known/oauth-protected-resource/mcp"

The response should describe the resource URL, authorization server, and supported scopes. Then make an unauthenticated request to the MCP endpoint:

curl -i "$MCP_RESOURCE_URL"

It should be rejected with 401 Unauthorized and a WWW-Authenticate challenge containing a resource_metadata pointer. A compatible MCP client can use that information to discover the authorization server and start its OAuth flow.

After connecting through an OAuth-capable MCP client, test with two separate users:

1. Issue a token for user A with tenant_id=tenant-a and documents:read.
2. Issue a token for user B with tenant_id=tenant-b and documents:read.
3. Insert one test document for each tenant.
4. Call list_documents as user A. It should return only tenant A’s document.
5. Call get_document as user A using the ID belonging to tenant B. It should return “Document not found.”
6. Call create_document with a read-only token. It should fail because documents:write is missing.

Important testing caveat: the SDK documentation explicitly notes that authorization middleware is an HTTP concern. An in-memory client or a server launched over stdio does not exercise the HTTP bearer-token gate. Test the deployed Streamable HTTP path with a real OAuth-capable client; a passing unit test of the database function alone does not prove the remote endpoint is protected.

10. Introspection vs JWT verification: which should you choose?

For either JWT verification or introspection, the important output is the same: a trusted principal, a verified issuer, a resource audience, scopes, expiry, and the tenant assignment. Use a mature library or the identity provider’s supported introspection endpoint; do not write your own JWT signature verifier.

11. Production hardening checklist

• Use HTTPS: protect the MCP endpoint and the introspection endpoint. RFC 7662 requires transport security for introspection.
• Keep audiences narrow: accept only the configured MCP resource. Do not accept a generic API audience or forward the incoming token to unrelated downstream APIs.
• Use short-lived access tokens: decide how quickly a revoked user must lose access. If you cache introspection results, keep the TTL short and document the revocation delay.
• Do not log credentials: never log bearer tokens, authorization codes, client secrets, or full introspection responses.
• Use least-privilege scopes: distinguish read, write, administrative, and destructive operations. Check the required scope inside each tool.
• Enforce tenant isolation twice: filter every query by tenant and consider PostgreSQL row-level security for high-value data. With connection pools, set tenant context transaction-locally and reset it reliably.
• Keep tenant identity immutable: derive it from the validated token or a trusted membership lookup. Never accept it from the model as an authority.
• Protect the introspection client: use a dedicated confidential client, rotate its secret, restrict its permissions, and store it in a secret manager.
• Fail closed: if introspection is unavailable, the issuer is wrong, the audience is missing, or the tenant claim is absent, do not run the tool.
• Audit safely: log the tenant, subject, tool name, outcome, and correlation ID where appropriate—but not document bodies or tokens.
• Review authorization-server registration: the current MCP spec moves toward CIMD and strengthens issuer handling. Keep your client registration and issuer validation compatible with the spec revision you support.

12. Troubleshooting

Every request returns 401

• Check that the introspection endpoint is reachable from the MCP container or host.
• Verify the resource server’s client ID and secret with the identity provider’s expected authentication method.
• Confirm active is true and that iss exactly matches the configured issuer.
• Check that aud contains the exact public MCP resource URL, including the correct path.
• Make sure the token has not expired and that the introspection service is returning an integer expiry.

The metadata endpoint returns 404

Check the well-known route and your reverse-proxy path handling. The SDK documents a path that includes the MCP endpoint suffix, such as /.well-known/oauth-protected-resource/mcp. Do not assume that a proxy mounted under a prefix will preserve the path automatically.

The tool says permission is missing

Inspect the scopes returned by introspection. This example requires mcp:access at the HTTP resource-server layer and checks documents:read or documents:write inside the tool. A token with only the global scope can authenticate but still cannot call a tool whose specific scope is missing.

It works over stdio but not over HTTP

That is not proof of a broken token. The stdio and in-memory paths do not run the HTTP bearer middleware. Test the actual Streamable HTTP endpoint with an OAuth-capable client.

One tenant can see another tenant’s data

Search every query, cache key, background job, export, and object-storage path for missing tenant constraints. A correct token verifier cannot compensate for a query that selects rows globally. Also check whether any tool accepts a tenant ID from its arguments and uses it without comparing it to the authenticated tenant.

13. Limitations and trade-offs

This is a resource-server example, not a complete identity platform. It assumes an external authorization server already handles user login, consent, client registration, token issuance, and tenant membership. The custom tenant_id claim must be configured and maintained by that identity system.

The SQLite database is for demonstrating the isolation predicate. For a multi-instance production service, use a shared database and enforce tenant constraints at the repository and preferably database-policy layers. The sample creates a fresh HTTP client for each introspection call; use a pooled client and a carefully bounded cache if request volume requires it. Any cache trades lower latency for a window in which revocation may not be observed.

Finally, a token’s tenant claim is only as trustworthy as the process that issues it. If users can change their own tenant attribute, or if a token is issued before a membership change and remains valid for hours, the MCP server will faithfully enforce stale identity data. Keep token lifetimes and membership-change revocation aligned with your threat model.

14. FAQ

Does the MCP Python SDK issue OAuth tokens?

No. In this architecture the MCP server is a resource server. An external authorization server authenticates users and issues tokens; the MCP server verifies them.

Are OAuth scopes enough for multi-tenant isolation?

No. Scopes describe allowed actions, such as reading or writing documents. Tenant isolation determines which tenant’s records those actions can touch. You need both checks.

Can I pass tenant_id to a tool for convenience?

You can accept it as a non-authoritative hint only if you compare it to the authenticated tenant and reject mismatches. The safer default is not to expose it as a tool argument at all.

Should I use JWT verification or introspection?

Choose based on revocation requirements, traffic, and identity-provider support. Introspection centralizes active-state checks but adds a network dependency. JWT verification can reduce latency but requires correct signature, issuer, expiry, and audience validation.

Does this protect a local stdio MCP server?

No. This OAuth middleware protects HTTP transports. A stdio server relies on the local process boundary and its environment or credential-management design.

Can I use this with PostgreSQL?

Yes. Keep the same authenticated-principal helper and add the tenant predicate to every query. For stronger defense in depth, configure PostgreSQL row-level security and set tenant context within each transaction.

15. Further reading on GyanAangan

• Build a Local Notes MCP Server in Python: SQLite Tools + MCP Inspector — the basic server and tool-building foundation.
• MCP 2026-07-28 Migration Guide: Python SDK v2 and Stateless HTTP — protocol and SDK migration context.
• MCP Python SDK v2.3.0 Upgrade Guide — SDK release-specific changes and production gotchas.
• AI Developer Automation with MCP — patterns for connecting MCP tools to real developer workflows.

Primary sources and verification notes

• Official MCP Python SDK authorization guide — TokenVerifier, AccessToken, AuthSettings, protected-resource metadata, and request identity.
• Official MCP Python SDK repository — current SDK source and examples.
• Official Python SDK split authorization/resource-server example — reference implementation using introspection.
• MCP Python SDK 2.3.0 on PyPI — release metadata and provenance; released 2 October 2026.
• MCP specification 2026-07-28 announcement — stateless protocol, authorization hardening, issuer handling, and the move from DCR toward CIMD.
• RFC 7662 — OAuth token introspection.
• RFC 8707 — OAuth resource indicators and audience restriction.
• RFC 9728 — OAuth protected-resource metadata.
• RFC 9207 — authorization-server issuer identification.

Verification caveat: The SDK API names, package release, and protocol behavior above were checked against official MCP documentation, the official repository, PyPI, and the cited RFCs on 3 November 2026. The example is an implementation template, not a claim of a live deployment or penetration test. The identity-provider-specific introspection URL, client-authentication method, and custom tenant_id mapper must be adapted to your provider and verified in your own environment.