Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/_experimental/mcp_server/outbound_credentials/httpx_auth.py: 71%

1"""Concrete `httpx2.Auth` objects the resolver returns for the self-contained modes. 

2 

3These are the egress credential as the SDK consumes it: an `httpx2.Auth` attached to the 

4upstream `AsyncClient`. The OAuth-flow modes (`authorization_code`, `client_credentials`, 

5`token_exchange`) return SDK-provided auth objects instead and land later. 

6 

7`auth_flow` mutating the outbound request is the `httpx2.Auth` contract, not a house-style 

8violation: the request is httpx2's object, and these carry no state of their own. 

9""" 

10 

11from __future__ import annotations 

12 

13from collections.abc import Generator 

14 

15import httpx2 

16from pydantic import SecretStr 

17 

18 

19class NoOpAuth(httpx2.Auth): 

20 """Attaches nothing — the `none` mode (and the seam-level default).""" 

21 

22 def auth_flow(self, request: httpx2.Request) -> Generator[httpx2.Request, httpx2.Response, None]: 

23 yield request 

24 

25 

26class StaticHeaderAuth(httpx2.Auth): 

27 """Sets one fixed header on every request — the `api_key` family and `passthrough`. 

28 

29 The header value is a live credential (a bearer token, an API key, a forwarded user 

30 token), so it is held as a `SecretStr` and unwrapped only when written onto the request. 

31 That keeps it masked in reprs, `vars()`, tracebacks, and structured logs, matching the 

32 `SecretStr` discipline the config models use. 

33 """ 

34 

35 def __init__(self, header_value: str, header_name: str = "Authorization") -> None: 

36 self.header_name = header_name 

37 self._header_value = SecretStr(header_value) 

38 

39 def auth_flow(self, request: httpx2.Request) -> Generator[httpx2.Request, httpx2.Response, None]: 

40 request.headers[self.header_name] = self._header_value.get_secret_value() 

41 yield request