KiQdZddlZddlmZmZddlmZmZddlm Z ddl Z ddl Z ddl m Z mZddlmZmZmZmZddlmZmZGd d eZd ed eegeeffd ZGdde ZGddeZGdde ZGddeZy)a OAuth client credential extensions for MCP. Provides OAuth providers for machine-to-machine authentication flows: - ClientCredentialsOAuthProvider: For client_credentials with client_id + client_secret - PrivateKeyJWTOAuthProvider: For client_credentials with private_key_jwt authentication (typically using a pre-built JWT from workload identity federation) - RFC7523OAuthClientProvider: For jwt-bearer grant (RFC 7523 Section 2.1) N) AwaitableCallable)AnyLiteral)uuid4) BaseModelField)OAuthClientProviderOAuthFlowErrorOAuthTokenError TokenStorage)OAuthClientInformationFullOAuthClientMetadataceZdZdZ ddedededededd edzd dffd Zdd Zd e jfd Z d e jfdZ xZ S)ClientCredentialsOAuthProvideraOAuth provider for client_credentials grant with client_id + client_secret. This provider sets client_info directly, bypassing dynamic client registration. Use this when you already have client credentials (client_id and client_secret). Example: ```python provider = ClientCredentialsOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", client_secret="my-client-secret", ) ``` N server_urlstorage client_id client_secrettoken_endpoint_auth_method)client_secret_basicclient_secret_postscopesreturnc|tddg||}t| |||dddtd||dg|||_y)aInitialize client_credentials OAuth provider. Args: server_url: The MCP server URL. storage: Token storage implementation. client_id: The OAuth client ID. client_secret: The OAuth client secret. token_endpoint_auth_method: Authentication method for token endpoint. Either "client_secret_basic" (default) or "client_secret_post". scopes: Optional space-separated list of scopes to request. Nclient_credentials redirect_uris grant_typesrscoper@)rrrrrr )rsuper__init__r_fixed_client_info) selfrrrrrrclient_metadata __class__s y/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/mcp/client/auth/extensions/client_credentials.pyr#z'ClientCredentialsOAuthProvider.__init__)sY*.-.'A   _gtT5Q"<'-.'A # cK|jjjd{|j_|j|j_d|_y76wz6Load stored tokens and set pre-configured client_info.NTcontextr get_tokenscurrent_tokensr$ client_info _initializedr%s r( _initializez*ClientCredentialsOAuthProvider._initializeOH,0LL,@,@,K,K,M&M ##'#:#:   'N(A#A!7A#c>K|jd{S7w)z)Perform client_credentials authorization.N"_exchange_token_client_credentialsr2s r(_perform_authorizationz5ClientCredentialsOAuthProvider._perform_authorizationU<<>>>> cKddi}ddi}|jj||\}}|jj|jjr|jj |d<|jj j r#|jj j |d<|j}tjd|||Sw) z:Build token exchange request for client_credentials grant. grant_typer Content-Type!application/x-www-form-urlencodedresourcer POSTdataheaders) r-prepare_token_authshould_include_resource_paramprotocol_versionget_resource_urlr&r _get_token_endpointhttpxRequestr% token_datarD token_urls r(r8zAClientCredentialsOAuthProvider._exchange_token_client_credentialsYs .& $23V"W#ll==j'R G << 5 5dll6S6S T%)\\%B%B%DJz " << ' ' - -"&,,">">"D"DJw ,,. }}VYZQQsC C")rNrN)__name__ __module__ __qualname____doc__strr rr#r3rJrKr9r8 __classcell__r's@r(rrs,\q!$ $ $  $  $ %,,W$X $ d $  $ L! ?emm?R%--Rr)rtokenrc,dtdtffd }|S)aCreate an assertion provider that returns a static JWT token. Use this when you have a pre-built JWT (e.g., from workload identity federation) that doesn't need the audience parameter. Example: ```python provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=static_assertion_provider(my_prebuilt_jwt), ) ``` Args: token: The pre-built JWT assertion string. Returns: An async callback suitable for use as an assertion_provider. audiencercKSwN)rYrWs r(providerz+static_assertion_provider..providers  srT)rWr]s` r(static_assertion_providerr_ns. Or)ceZdZUdZedZeed<edZeed<edZ eed<ed d Z eed <ed d Z e ed<edd Z eeefdzed<deegeeffdZy)SignedJWTParametersaParameters for creating SDK-signed JWT assertions. Use `create_assertion_provider()` to create an assertion provider callback for use with `PrivateKeyJWTOAuthProvider`. Example: ```python jwt_params = SignedJWTParameters( issuer="my-client-id", subject="my-client-id", signing_key=private_key_pem, ) provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=jwt_params.create_assertion_provider(), ) ``` z0Issuer for JWT assertions (typically client_id).) descriptionissuerz.providersdiik"C{{||T22257| &F%% d445::fd&6&6$BXBXY YsB,B/r^)r%r]s` r(create_assertion_providerz-SignedJWTParameters.create_assertion_providers  ZS ZS Zr))rPrQrRrSr rcrT__annotations__rdrerjrmrzrndictrrrrr\r)r(raras*$VWFCW%cdGSd)TUKU"7@ghsh!#;bccc/4TOc/dtCH~,d8SE9S>4I+Jr)rac eZdZdZ ddedededeegeefdedzddf fd Zdd Z de jfd Z d e eefddfd Zde jfdZxZS)PrivateKeyJWTOAuthProvideraqOAuth provider for client_credentials grant with private_key_jwt authentication. Uses RFC 7523 Section 2.2 for client authentication via JWT assertion. The JWT assertion's audience MUST be the authorization server's issuer identifier (per RFC 7523bis security updates). The `assertion_provider` callback receives this audience value and must return a JWT with that audience. **Option 1: Pre-built JWT via Workload Identity Federation** In production scenarios, the JWT assertion is typically obtained from a workload identity provider (e.g., GCP, AWS IAM, Azure AD): ```python async def get_workload_identity_token(audience: str) -> str: # Fetch JWT from your identity provider # The JWT's audience must match the provided audience parameter return await fetch_token_from_identity_provider(audience=audience) provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=get_workload_identity_token, ) ``` **Option 2: Static pre-built JWT** If you have a static JWT that doesn't need the audience parameter: ```python provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=static_assertion_provider(my_prebuilt_jwt), ) ``` **Option 3: SDK-signed JWT (for testing/simple setups)** For testing or simple deployments, use `SignedJWTParameters.create_assertion_provider()`: ```python jwt_params = SignedJWTParameters( issuer="my-client-id", subject="my-client-id", signing_key=private_key_pem, ) provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=jwt_params.create_assertion_provider(), ) ``` Nrrrassertion_providerrrctddgd|}t| |||ddd||_t d|dgd||_y)aInitialize private_key_jwt OAuth provider. Args: server_url: The MCP server URL. storage: Token storage implementation. client_id: The OAuth client ID. assertion_provider: Async callback that takes the audience (authorization server's issuer identifier) and returns a JWT assertion. Use `SignedJWTParameters.create_assertion_provider()` for SDK-signed JWTs, `static_assertion_provider()` for pre-built JWTs, or provide your own callback for workload identity federation. scopes: Optional space-separated list of scopes to request. Nrprivate_key_jwtrr!)rrrrr )rr"r#_assertion_providerrr$)r%rrrrrr&r's r(r#z#PrivateKeyJWTOAuthProvider.__init__s^,.-.'8   _gtT5Q#5 "<-.'8 # r)cK|jjjd{|j_|j|j_d|_y76wr+r,r2s r(r3z&PrivateKeyJWTOAuthProvider._initialize%r4r5c>K|jd{S7w)z>Perform client_credentials authorization with private_key_jwt.Nr7r2s r(r9z1PrivateKeyJWTOAuthProvider._perform_authorization+r:r;rMcK|jjs tdt|jjj}|j |d{}||d<d|d<y7w)IAdd JWT assertion for client authentication to token endpoint parameters./Missing OAuth metadata for private_key_jwt flowNclient_assertion6urn:ietf:params:oauth:client-assertion-type:jwt-bearerclient_assertion_type)r-oauth_metadatar rTrcr)r%rMrY assertions r(_add_client_authentication_jwtz9PrivateKeyJWTOAuthProvider._add_client_authentication_jwt/sk||** !RS St||2299:228<< *3 %&.f *+ =sAA3!A1"A3cKddi}ddi}|j|d{|jj|jjr|jj |d<|jj j r#|jj j |d<|j}tjd ||| S7w) zOBuild token exchange request for client_credentials grant with private_key_jwt.r=rr>r?rMNr@r rArB) rr-rFrGrHr&r rIrJrKrLs r(r8z=PrivateKeyJWTOAuthProvider._exchange_token_client_credentials=s .& $23V"W11Z1HHH << 5 5dll6S6S T%)\\%B%B%DJz " << ' ' - -"&,,">">"D"DJw ,,. }}VYZQQ IsCCB=Cr[rO)rPrQrRrSrTr rrr#r3rJrKr9rrrr8rUrVs@r(rrs9B" % % %  % %cUIcN%:; % d %  % N! ?emm? g$sCx. gUY gR%--Rr)rcNeZdZUdZeddZedzed<eddZedzed<eddZ edzed <edd Z edzed <edd Z e ee fdzed <eddZedzed<eddZedzed<eddZeed<ddedzdefdZy) JWTParameterszJWT parameters.NzeJWT assertion for JWT authentication. Will be used instead of generating a new assertion if provided.rhrzIssuer for JWT assertions.rcz&Subject identifier for JWT assertions.rdzAudience for JWT assertions.rYz%Additional claims for JWT assertions.rrfrgjwt_signing_algorithmzPrivate key for JWT signing.jwt_signing_keyrkrljwt_lifetime_secondswith_audience_fallbackrc `|j|j}|S|js td|js td|js td|j r |j n|}|s tdt tj}|j|j|||jz|ttd}|j|jxsitj||j|jxsd}|S)Nz(Missing signing key for JWT bearer grantz#Missing issuer for JWT bearer grantz$Missing subject for JWT bearer grantz%Missing audience for JWT bearer grantrqrfrx)rrr rcrdrYrzr{rrTrr|rr}r~r)r%rrrYrrs r( to_assertionzJWTParameters.to_assertioncs >> %I:7''$%OPP;;$%JKK<<$%KLL(, t}};QH$%LMMdiik"C{{||T66657| &F MM$+++ , $$44?I r)r[)rPrQrRrSr rrTrrcrdrYrrrrrrrzrr\r)r(rrRs!JIsTz t9UVFC$JV:bcGS4Zc ;YZHcDjZ$)$Dk$lFDcNT !l(-gKr(s3:s"'B`"aOS4Za %c?f g#g 3:  r)rc2eZdZdZ ddedededeegedfdzdegee eedzffdzde d e dzd dffd Z dd d edede eefdzd ej ffdZd ej ffd Zde eeffdZd ej fdZxZS)RFC7523OAuthClientProvideraOAuth client provider for RFC 7523 jwt-bearer grant. .. deprecated:: Use :class:`ClientCredentialsOAuthProvider` for client_credentials with client_id + client_secret, or :class:`PrivateKeyJWTOAuthProvider` for client_credentials with private_key_jwt authentication instead. This provider supports the jwt-bearer authorization grant (RFC 7523 Section 2.1) where the JWT itself is the authorization grant. Nrr&rredirect_handlercallback_handlertimeoutjwt_parametersrctddl}|jdtdt |||||||||_y)NrzsRFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider or PrivateKeyJWTOAuthProvider instead.) stacklevel)warningswarnDeprecationWarningr"r#r) r%rr&rrrrrrr's r(r#z#RFC7523OAuthClientProvider.__init__sH   5   _g?OQacjk,r)r auth_code code_verifierrMcK|xsi}|jjjdk(r|j|t||||d{S7w)z9Build token exchange request for authorization_code flow.rrN)r-r&rrr""_exchange_token_authorization_code)r%rrrMr's r(rz=RFC7523OAuthClientProvider._exchange_token_authorization_codes[ %2 << ' ' B BFW W  / /: / FW? =eo?ppppsAAAAcKd|jjjvr|jd{}|St|d{S77w)zPerform the authorization flow.+urn:ietf:params:oauth:grant-type:jwt-bearerN)r-r&r_exchange_token_jwt_bearerr"r9)r% token_requestr's r(r9z1RFC7523OAuthClientProvider._perform_authorizationsS 8DLL!>!@ :r)cK|jjs td|js td|jjs t dt |jjj}|jj|}d|d}|jj|jjr|jj|d<|jjjr#|jjj|d<|j}tj d ||d d i Sw) z2Build token exchange request for JWT bearer grant.zMissing client infozMissing JWT parameterszMissing OAuth metadatarr)r=rr@r rAr>r?rB)r-r0r rrr rTrcrrFrGrHr&r rIrJrK)r%rcrrMrNs r(rz5RFC7523OAuthClientProvider._exchange_token_jwt_bearers"||'' !67 7"" !9: :||**!":; ;T\\00778''44F4S H" << 5 5dll6S6S T%)\\%B%B%DJz " << ' ' - -"&,,">">"D"DJw ,,. }} IJIl8m  sEE )NNr!N)rPrQrRrSrTrr rrtuplefloatrr#rrrJrKrr9rrrUrVs@r(rrs  EISW/3---- - #C5)D/#9:TA - #2ysC$J1G'H#HIDP --&,- -,Z^qq-0qAEc3hRVAVq q:emm:ADcNA& %-- r)r)rSr{collections.abcrrtypingrruuidrrJr}pydanticrr mcp.client.authr r r r mcp.shared.authrrrrTr_rarrrr\r)r(rs / %^^KSR%8SRlSXseYs^6K-L:4)4nMR!4MR`1I1ha !4a r)