scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

refactor(mcp): extract the authorization_code arm into a helper

Browse source 
Mirror the api_key arm's structure: the inline AuthorizationCodeConfig body moves into
_authorization_code(subject, server), keeping resolve_credentials a flat one-line-per-arm dispatch.
The helper is annotated with the concrete StaticHeaderAuth it returns rather than the abstract
httpx.Auth (which api_key uses) because a new method carrying the unresolved httpx.Auth return
would add reportUnknownMemberType; the concrete type is both precise and budget-neutral.
This commit is contained in:
Tin Chi Lo 2026-06-25 17:41:04 -07:00
parent 9437ce11fe
commit d2f9917f02
1 changed files with 17 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py
View file

@ -83,18 +83,7 @@ class UpstreamCredentialProvider:
case TokenExchangeConfig():
return _not_implemented(AuthSpecKind.token_exchange)
case AuthorizationCodeConfig():
token = await self._authz_token(subject, server)
if token is None:
return Error(
CredError.of_unauthorized(
"Authorization required: complete the OAuth flow for this server."
)
)
return Ok(
StaticHeaderAuth(
f"Bearer {token.access_token}", header_name="Authorization"
)
)
return await self._authorization_code(subject, server)
case AwsSigV4Config():
return _not_implemented(AuthSpecKind.aws_sigv4)
assert_never(server.config)
@ -125,6 +114,22 @@ class UpstreamCredentialProvider:
)
assert_never(config.key_source)
async def _authorization_code(
self, subject: Subject, server: ServerSpec
) -> Result[StaticHeaderAuth, CredError]:
token = await self._authz_token(subject, server)
if token is None:
return Error(
CredError.of_unauthorized(
"Authorization required: complete the OAuth flow for this server."
)
)
return Ok(
StaticHeaderAuth(
f"Bearer {token.access_token}", header_name="Authorization"
)
)
async def _authz_token(
self, subject: Subject, server: ServerSpec
) -> OAuthToken | None: