Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/mcp_jwt_signer/__init__.py: 25%

26 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 12:01 +0000

1"""MCP JWT Signer guardrail — built-in LiteLLM guardrail for zero trust MCP auth.""" 

2 

3from typing import TYPE_CHECKING, Final 

4 

5from litellm.types.guardrails import SupportedGuardrailIntegrations 

6 

7from .mcp_jwt_signer import MCPJWTSigner, get_mcp_jwt_signer 

8 

9if TYPE_CHECKING: 9 ↛ 10line 9 didn't jump to line 10 because the condition on line 9 was never true

10 from litellm.types.guardrails import Guardrail, LitellmParams 

11 

12 

13def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail") -> MCPJWTSigner: 

14 import litellm 

15 

16 guardrail_name: Final = guardrail.get("guardrail_name") 

17 if not guardrail_name: 

18 raise ValueError("MCPJWTSigner guardrail requires a guardrail_name") 

19 

20 mode: Final = litellm_params.mode 

21 if mode != "pre_mcp_call": 

22 raise ValueError( 

23 f"MCPJWTSigner guardrail '{guardrail_name}' has mode='{mode}' but must use " 

24 "mode='pre_mcp_call'. JWT injection only fires for MCP tool calls." 

25 ) 

26 

27 optional_params: Final = getattr(litellm_params, "optional_params", None) 

28 

29 def _get(key): 

30 if optional_params is not None: 

31 v: Final = getattr(optional_params, key, None) 

32 if v is not None: 

33 return v 

34 return getattr(litellm_params, key, None) 

35 

36 signer: Final = MCPJWTSigner( 

37 guardrail_name=guardrail_name, 

38 event_hook=litellm_params.mode, 

39 default_on=litellm_params.default_on, 

40 # Core signing 

41 issuer=_get("issuer"), 

42 audience=_get("audience"), 

43 ttl_seconds=_get("ttl_seconds"), 

44 # FR-5: verify + re-sign 

45 access_token_discovery_uri=_get("access_token_discovery_uri"), 

46 token_introspection_endpoint=_get("token_introspection_endpoint"), 

47 verify_issuer=_get("verify_issuer"), 

48 verify_audience=_get("verify_audience"), 

49 # FR-12: end-user identity mapping 

50 end_user_claim_sources=_get("end_user_claim_sources"), 

51 # FR-13: claim operations 

52 add_claims=_get("add_claims"), 

53 set_claims=_get("set_claims"), 

54 remove_claims=_get("remove_claims"), 

55 # FR-14: two-token model 

56 channel_token_audience=_get("channel_token_audience"), 

57 channel_token_ttl=_get("channel_token_ttl"), 

58 # FR-15: incoming claim validation 

59 required_claims=_get("required_claims"), 

60 optional_claims=_get("optional_claims"), 

61 # FR-9: debug headers 

62 debug_headers=_get("debug_headers") or False, 

63 # FR-10: configurable scopes 

64 allowed_scopes=_get("allowed_scopes"), 

65 ) 

66 litellm.logging_callback_manager.add_litellm_callback(signer) 

67 return signer 

68 

69 

70guardrail_initializer_registry: Final = { 

71 SupportedGuardrailIntegrations.MCP_JWT_SIGNER.value: initialize_guardrail, 

72} 

73 

74guardrail_class_registry: Final = { 

75 SupportedGuardrailIntegrations.MCP_JWT_SIGNER.value: MCPJWTSigner, 

76} 

77 

78__all__ = [ 

79 "MCPJWTSigner", 

80 "get_mcp_jwt_signer", 

81 "initialize_guardrail", 

82]