Identity assertion

साधारण OAuth provider (OAuth clients) MCP server से एक सवाल पूछकर शुरू करता है: आप किस authorization server पर भरोसा करते हैं? जवाब जिधर इशारा करे, वह उधर चला जाता है, और फिर या तो कोई इंसान sign in करता है या उसकी जगह कोई pre-shared secret काम आता है।

Enterprise नहीं चाहता कि इनमें से कोई भी बात हर server के हिसाब से अलग तय हो। वह पहले से एक identity provider चलाता है (Okta, Microsoft Entra ID, या आपका अपना); user आज सुबह ही उसमें sign in कर चुका है; और यही वह एक जगह है जहाँ security team तय करना चाहती है कि कौन कहाँ तक पहुँच सकता है। SEP-990, यानी Enterprise-Managed Authorization extension, यह फ़ैसला वहीं ले जाता है। IdP एक short-lived JWT sign करता है, Identity Assertion JWT Authorization Grant, यानी ID-JAG: यह बयान कि यह user, इस client के ज़रिए, इस MCP server तक पहुँच सकता है। Client इसे देकर बदले में साधारण access token ले लेता है। न browser, न consent screen, न dynamic registration।

यह page उसी लेन-देन के दोनों सिरों के बारे में है। MCP server खुद कभी नहीं बदलता: वह अब भी Authorization वाला resource server ही है, जो भी token सामने आए उसे जाँचता है।

दो token requests

यहाँ दो अलग-अलग authorities काम कर रही हैं, और उन्हें अलग-अलग नाम से पहचान लेना ही इस page को समझने का ज़्यादातर हिस्सा है। Enterprise IdP आपके organization का identity provider है: उसे पता है कि employee कौन है, policy वहीं रहती है, और ID-JAG वही जारी करता है। SDK उससे कभी बात नहीं करता। MCP authorization server वही पक्ष है जो Authorization में था: MCP server के metadata में नामित issuer, वह चीज़ जो वे tokens बनाती है जिन्हें वह MCP server स्वीकार करता है। साधारण OAuth flow में ये दोनों भूमिकाएँ आमतौर पर एक ही system निभाता है। यहाँ ये दो हैं, और पूरा grant बस इतना है कि दूसरा पहले पर भरोसा करने को राज़ी हो।

Client इनमें से हर एक को एक token request भेजता है।

1. Enterprise IdP को। Client user के sign-in (उनका OpenID Connect ID token) के बदले ID-JAG लेता है। यह RFC 8693 token exchange है, यह पूरी तरह आपके IdP का API है, और SDK इसे नहीं करता। यह आप करते हैं, एक async callback के अंदर। Policy का फ़ैसला भी यहीं होता है: जो IdP मना कर दे वह ID-JAG जारी ही नहीं करता, और पेश करने को कुछ बचता ही नहीं।
2. MCP authorization server को। Client ID-JAG को RFC 7523 jwt-bearer grant के तहत पेश करता है (grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer, ID-JAG assertion के रूप में) और access token पाता है। यही वह request है जो SDK भेजता है, और इसे स्वीकार करना ही वह एक चीज़ है जो यह page authorization server में जोड़ता है।

नीचे सब कुछ दूसरी request के बारे में है: उसे भेजने वाला client और उसका जवाब देने वाला authorization server।

Client

IdentityAssertionOAuthProvider mcp.client.auth.extensions.identity_assertion में रहता है। OAuth clients के हर provider की तरह यह भी httpx2.Auth है: एक बनाएँ, उसे auth= पर रखें, और httpx2.AsyncClient transport को सौंप दें।

import time
import uuid

import httpx2
import jwt

from mcp import Client
from mcp.client.auth.extensions.identity_assertion import IdentityAssertionOAuthProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken

IDP_SIGNING_KEY = "the-enterprise-idp-signing-key-for-this-demo"


class InMemoryTokenStorage:
    def __init__(self) -> None:
        self.tokens: OAuthToken | None = None
        self.client_info: OAuthClientInformationFull | None = None

    async def get_tokens(self) -> OAuthToken | None:
        return self.tokens

    async def set_tokens(self, tokens: OAuthToken) -> None:
        self.tokens = tokens

    async def get_client_info(self) -> OAuthClientInformationFull | None:
        return self.client_info

    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
        self.client_info = client_info


def idp_issue_id_jag(subject: str, audience: str, resource: str) -> str:
    now = int(time.time())
    claims = {
        "iss": "https://idp.example.com",
        "sub": subject,
        "aud": audience,
        "client_id": "finance-agent",
        "resource": resource,
        "scope": "notes:read",
        "jti": str(uuid.uuid4()),
        "iat": now,
        "exp": now + 300,
    }
    return jwt.encode(claims, IDP_SIGNING_KEY, algorithm="HS256", headers={"typ": "oauth-id-jag+jwt"})


async def fetch_id_jag(audience: str, resource: str) -> str:
    return idp_issue_id_jag("alice@example.com", audience, resource)


oauth = IdentityAssertionOAuthProvider(
    server_url="http://localhost:8001/mcp",
    storage=InMemoryTokenStorage(),
    client_id="finance-agent",
    client_secret="finance-agent-secret",
    issuer="https://auth.example.com/",
    assertion_provider=fetch_id_jag,
    scope="notes:read",
)


async def main() -> None:
    async with httpx2.AsyncClient(auth=oauth, follow_redirects=True) as http_client:
        transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

इसे नीचे से पढ़ें।

• main() वही standard OAuth-client वाला main() है (OAuth clients), पंक्ति-दर-पंक्ति बिना बदलाव। यही बात है: एक बार provider बन जाए तो आगे किसी को पता नहीं चलता कि token किस grant से आया।
• Provider वह लेता है जो बाकी providers discover नहीं कर सकते: client_id और client_secret जो किसी ने authorization server के साथ पहले से register कर रखे हैं, उस authorization server का issuer, और assertion_provider, एक async callback जो माँगने पर ताज़ा ID-JAG लौटाता है।
• storage वही TokenStorage protocol है। सिर्फ़ दो token methods ही कभी call होते हैं; यहाँ dynamic registration नहीं है, इसलिए याद रखने को कोई client_info नहीं है।

Assertion provider

fetch_id_jag(audience, resource) ही वह इकलौता code है जो आप लिखते हैं। यह हर token exchange पर एक बार await होता है, construction के समय कभी नहीं, और सिर्फ़ तब जब authorization server का metadata fetch और validate हो चुका हो, इसलिए गलत configure किया गया issuer कभी assertion leak नहीं करवाता। इसके दो arguments उन claims में से दो हैं जिनके साथ ID-JAG बनना ज़रूरी है: audience authorization server का issuer है (ID-JAG का aud) और resource MCP server का canonical identifier है (ID-JAG का resource)। तीसरा वह है जो आपके पास पहले से है: ID-JAG के client_id claim में वही client_id होना चाहिए जो आपने provider को दिया, वरना authorization server exchange से मना कर देता है।

उसके ऊपर वाला idp_issue_id_jag आपका code नहीं है। वह identity provider की जगह खड़ा है, और assertion को उसी process में sign करता है ताकि file पूरी रहे और आप ID-JAG में जाने वाला हर claim पढ़ सकें। असली fetch_id_jag इसकी जगह पिछले section की पहली token request भेजता है: आपके IdP के सामने RFC 8693 token exchange, जिसे Identity Assertion JWT Authorization Grant draft परिभाषित करता है और जिसे SEP-990 profile करता है। Sign in किए हुए user का ID token subject_token के रूप में जाता है, requested_token_type ID-JAG का अपना URN है (urn:ietf:params:oauth:token-type:id-jag), audience और resource जस के तस आगे जाते हैं, और response में ID-JAG आता है। अपने IdP की documentation में इन्हीं नामों के साथ यही exchange ढूँढें।

Issuer configuration है

उलटफेर यहाँ है। OAuthClientProvider resource server से पूछता है कि कौन सा authorization server इस्तेमाल करे, और जवाब जिधर इशारा करे उधर चला जाता है। यह provider ऐसा करने से मना करता है: issuer ज़रूरी है, RFC 8414 metadata उसी issuer के अपने well-known path से fetch होता है, token endpoint उसी issuer के origin पर होना चाहिए, और resource server से कभी कुछ नहीं पूछा जाता।

Extension इसकी माँग नहीं करता; यह जान-बूझकर चुना गया ज़्यादा सख़्त रास्ता है। इस client के पास चुराने लायक दो चीज़ें हैं, एक pre-registered secret और एक audience-bound assertion, और जो client किसी compromised MCP server को खुद को हमलावर के authorization server की तरफ़ मोड़ने दे, वह दोनों उसी को post कर देगा। Construction के समय issuer को pin कर देने से वह बातचीत ही ख़त्म हो जाती है।

Confidential client

client_secret ज़रूरी है; इसके बिना constructor ValueError raise करता है। SEP-990 के नीचे वाला IETF profile इस grant को confidential clients के लिए आरक्षित रखता है, SEP-990 client से authenticate करने की माँग करता है, और यह SDK shared secret पर ज़ोर देकर दोनों लागू करता है। token_endpoint_auth_method तय करता है कि यह कहाँ से होकर जाए: client_secret_post (default, form body में) या client_secret_basic (HTTP Basic header)। Profile private_key_jwt की भी इजाज़त देता है; यह provider उसे support नहीं करता।

Provider आपके लिए क्या करता है

पहली request बिना authentication के जाती है, और server का 401 flow शुरू करता है।

1. Discovery। यह configure किए गए issuer के RFC 8414 well-known path से authorization server metadata fetch करता है, जाँचता है कि document का issuer मेल खाता है, और जाँचता है कि token endpoint issuer के origin पर है।
2. Assertion। यह आपके assertion_provider को await करता है।
3. Exchange। यह token endpoint पर jwt-bearer grant POST करता है, OAuthToken store करता है, और आपकी मूल request Authorization: Bearer ... के साथ दोबारा भेजता है।

जिस 403 का WWW-Authenticate insufficient_scope बताता है, वह चरण 2 और 3 को आपके scope और challenge किए गए scope के union के साथ दोबारा चलाता है। (scope हमेशा सिर्फ़ एक माँग है; इस page का authorization server वही देता है जो ID-JAG कहता है, उससे ज़्यादा कुछ नहीं।) इसमें कहीं कोई refresh token नहीं है: access token expire होने पर अगला 401 ताज़ा ID-JAG बनवाता है और फिर exchange करता है, और यही वह lever है जो IdP के हाथ में है। नाकामियाँ OAuth clients के बाकी हिस्से वाले वही दो exceptions हैं: discovery और validation के लिए OAuthFlowError, और जब token endpoint मना करे तब उसका subclass OAuthTokenError।

Authorization server

ज़्यादातर बार आप यहीं रुक जाते हैं। MCP authorization server किसी और का product है, ID-JAGs स्वीकार करना उसकी configuration में चालू करने की चीज़ है, और SEP-990 का SDK वाला आधा हिस्सा ऊपर का client है।

SDK खुद authorization server भी बन सकता है: create_auth_routes authorization server के routes एक list के रूप में लौटाता है जिसे कोई भी Starlette app mount कर सकता है; repository में examples/servers/simple-auth/ इसी तरह एक चलाता है। SEP-990 उस surface में एक flag और एक method जोड़ता है:

import secrets
import time

import jwt
from pydantic import AnyHttpUrl
from starlette.applications import Starlette

from mcp.server.auth.provider import (
    AccessToken,
    AuthorizationCode,
    AuthorizationParams,
    AuthorizeError,
    IdentityAssertionParams,
    OAuthAuthorizationServerProvider,
    RefreshToken,
    TokenError,
)
from mcp.server.auth.routes import create_auth_routes
from mcp.shared.auth import JWT_BEARER_GRANT_TYPE, OAuthClientInformationFull, OAuthToken

ISSUER = "https://auth.example.com/"
MCP_SERVER = "http://localhost:8001/mcp"
IDP_ISSUER = "https://idp.example.com"
IDP_SIGNING_KEY = "the-enterprise-idp-signing-key-for-this-demo"

REGISTERED_CLIENTS = {
    "finance-agent": OAuthClientInformationFull(
        client_id="finance-agent",
        client_secret="finance-agent-secret",
        redirect_uris=None,
        grant_types=[JWT_BEARER_GRANT_TYPE],
        token_endpoint_auth_method="client_secret_post",
    )
}


class EnterpriseAuthorizationServer(OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]):
    def __init__(self) -> None:
        self.access_tokens: dict[str, AccessToken] = {}
        self.seen_jtis: set[str] = set()

    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
        return REGISTERED_CLIENTS.get(client_id)

    async def load_access_token(self, token: str) -> AccessToken | None:
        return self.access_tokens.get(token)

    async def exchange_identity_assertion(
        self, client: OAuthClientInformationFull, params: IdentityAssertionParams
    ) -> OAuthToken:
        try:
            header = jwt.get_unverified_header(params.assertion)
            claims = jwt.decode(
                params.assertion,
                IDP_SIGNING_KEY,
                algorithms=["HS256"],
                issuer=IDP_ISSUER,
                audience=ISSUER,
                options={"require": ["iss", "sub", "aud", "exp", "iat", "jti", "client_id", "resource", "scope"]},
            )
        except jwt.InvalidTokenError as error:
            raise TokenError("invalid_grant", "the assertion did not verify") from error
        if header.get("typ") != "oauth-id-jag+jwt":
            raise TokenError("invalid_grant", "the assertion is not an ID-JAG")
        if claims["client_id"] != client.client_id:
            raise TokenError("invalid_grant", "the assertion was issued to a different client")
        if claims["resource"] != MCP_SERVER:
            raise TokenError("invalid_target", "the assertion is for a resource this server does not serve")
        if claims["jti"] in self.seen_jtis:
            raise TokenError("invalid_grant", "the assertion has already been used")
        self.seen_jtis.add(claims["jti"])
        scopes = claims["scope"].split()
        access_token = f"mcp_{secrets.token_hex(16)}"
        self.access_tokens[access_token] = AccessToken(
            token=access_token,
            client_id=claims["client_id"],
            scopes=scopes,
            expires_at=int(time.time()) + 300,
            resource=claims["resource"],
            subject=claims["sub"],
        )
        return OAuthToken(access_token=access_token, token_type="Bearer", expires_in=300, scope=" ".join(scopes))

    async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
        raise AuthorizeError("unauthorized_client", "this authorization server only accepts ID-JAGs")

    async def load_authorization_code(self, client: OAuthClientInformationFull, authorization_code: str) -> None:
        return None

    async def exchange_authorization_code(
        self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
    ) -> OAuthToken:
        raise TokenError("invalid_grant", "this authorization server only accepts ID-JAGs")

    async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_token: str) -> None:
        return None

    async def exchange_refresh_token(
        self, client: OAuthClientInformationFull, refresh_token: RefreshToken, scopes: list[str]
    ) -> OAuthToken:
        raise TokenError("invalid_grant", "this authorization server only accepts ID-JAGs")


provider = EnterpriseAuthorizationServer()
auth_app = Starlette(
    routes=create_auth_routes(provider, issuer_url=AnyHttpUrl(ISSUER), identity_assertion_enabled=True)
)

• identity_assertion_enabled=True सब कुछ gate करता है। बंद होने पर, जो default है, /token इस grant का जवाब unsupported_grant_type से देता है चाहे आपने hook implement किया हो, और metadata में इसका ज़िक्र नहीं होता। चालू होने पर metadata में jwt-bearer grant type जुड़ जाता है और authorization_grant_profiles_supported में urn:ietf:params:oauth:grant-profile:id-jag सूचीबद्ध हो जाता है; यही वह field है जिससे extension support का ऐलान करता है। (इस SDK का client इसे कभी नहीं पढ़ता: वह एक issuer के लिए provision किया गया है और सीधे माँग लेता है।)
• exchange_identity_assertion ही hook है। इसके चलने से पहले SDK client को authenticate कर चुका होता है, public clients को मना कर चुका होता है, और उन clients को मना कर चुका होता है जिनके registration में यह grant सूचीबद्ध नहीं है। आपको IdentityAssertionParams मिलता है (कच्चा assertion, माँगे गए scopes और resource) और आप सादा OAuthToken लौटाते हैं।
• Dynamic client registration इस grant को बिना शर्त मना करता है, इसलिए यहाँ get_client हाथ से provision किया गया client देता है। ID-JAG client खुद को register करके अस्तित्व में नहीं ला सकता।
• आधी class इनकारों से भरी है। OAuthAuthorizationServerProvider पूरा authorization server है, इसलिए वह authorization-code flow भी माँगता है; जो server users को sign in भी कराता है वह उन्हें सच में implement करता है, और इस वाले में ठीक एक ही दरवाज़ा है।

ख़राब assertion को TokenError("invalid_grant", ...) से reject करें। इस flow का दूसरा error code invalid_target है: जो ID-JAG किसी ऐसे resource का नाम ले जिसे आप serve नहीं करते, उसे इसी से मना किया जाता है, और यही इस server को किसी और के resource के लिए tokens बनाने से रोकता है। और दिए गए scopes ID-JAG के scope claim से आते हैं (जिस assertion में यह न हो उसे भी मना किया जाता है); आपका server शायद इसकी जगह user के groups map करे।

और ध्यान दें कि लौटाए गए OAuthToken में क्या नहीं है: refresh token। IdP यह तय करके कि अगला ID-JAG जारी करना है या नहीं, तय करता है कि इस user की पहुँच कब तक बनी रहे। यहाँ बना refresh token वह फ़ैसला चुपचाप वापस सौंप देता।

grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer
assertion=eyJhbGciOiJIUzI1NiIsInR5cCI6Im9hdXRoLWlkLWphZytqd3QifQ...
client_id=finance-agent
resource=http://localhost:8001/mcp
scope=notes:read
client_secret=finance-agent-secret

HTTP/1.1 200 OK
{"access_token": "mcp_...", "token_type": "Bearer", "expires_in": 300, "scope": "notes:read"}

इसे आज़माएँ

SDK repository में examples/stories/identity_assertion/ यही page असल में चलता हुआ है: वही exchange_identity_assertion validator, उसके tokens पर gate किया गया MCP server, एक stand-in IdP, और client, सब एक self-checking program में। uv run python -m stories.identity_assertion.client --http पूरा exchange चलाता है और assert करता है कि जिस user का नाम IdP ने लिया, tool को वही user दिखता है।

सारांश

• SEP-990 यह फ़ैसला end user के बजाय enterprise identity provider को करने देता है कि client किन MCP servers तक पहुँच सकता है। IdP उस फ़ैसले को sign करके ID-JAG में डाल देता है।
• ID-JAG हासिल करना आपके IdP के सामने RFC 8693 token exchange है, और SDK इसे नहीं करता। उसे MCP authorization server के सामने पेश करना RFC 7523 jwt-bearer grant है, और SDK उसके दोनों पक्ष करता है।
• IdentityAssertionOAuthProvider एक और httpx2.Auth है: pre-registered confidential client, pin किया गया issuer, और एक assertion_provider(audience, resource) callback। न browser, न registration, न refresh token।
• Authorization server कभी resource server से discover नहीं होता। issuer को ठीक उसी string पर configure करें जो उसका metadata document देता है; तुलना एक-एक character की होती है।
• Server की तरफ़, identity_assertion_enabled=True और exchange_identity_assertion। SDK client को authenticate करता है और grant को gate करता है; ID-JAG validate करना पूरी तरह आपका काम है, और जारी किया गया token ID-JAG के resource से बँधा होता है, request के नहीं।

इकलौता पक्ष जिसे इस page ने कभी नहीं छुआ, वह MCP server है। अभी-अभी बनाए गए token के साथ वह जो करता है, वह Authorization में पहले से कर ही रहा था।