Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/agent_365/agent_365.py: 23%
276 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""Microsoft Agent 365 governance guardrail for MCP tool calls.
3Before the gateway executes an MCP tool, the pending call is sent to the
4Agent 365 tool-evaluation endpoint, where Microsoft Defender scores it and
5Agent 365 records it for observability. The returned allow/block verdict is
6enforced here. Authentication is the Entra On-Behalf-Of flow: the caller's
7incoming bearer token (audienced to this gateway's app registration) is
8exchanged for a delegated Agent 365 token, so Defender evaluates and audits
9as the signed-in user.
10"""
12import hashlib
13import threading
14import time
15import uuid
16from collections import OrderedDict
17from collections.abc import Mapping
18from typing import TYPE_CHECKING, ClassVar, Final, Literal, NoReturn
20import httpx
21from fastapi import HTTPException
22from pydantic import TypeAdapter, ValidationError
23from typing_extensions import ReadOnly, TypedDict
25from litellm._logging import verbose_proxy_logger
26from litellm.exceptions import Timeout as LitellmTimeout
27from litellm.integrations.custom_guardrail import (
28 CustomGuardrail,
29 log_guardrail_information,
30)
31from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
32from litellm.llms.custom_httpx.http_handler import (
33 AsyncHTTPHandler,
34 get_async_httpx_client,
35 httpxSpecialProvider,
36)
37from litellm.types.guardrails import GuardrailEventHooks
38from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import (
39 AGENT_365_PROD_API_BASE,
40 AGENT_365_PROD_RESOURCE_APP_ID,
41 AGENT_365_SCOPE_NAME,
42 Agent365GuardrailConfigModel,
43)
45if TYPE_CHECKING: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true
46 from litellm.caching.caching import DualCache
47 from litellm.proxy._types import UserAPIKeyAuth
48 from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
49 from litellm.types.utils import GuardrailStatus
51TOKEN_ENDPOINT_TEMPLATE: Final = "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
52EVALUATE_PATH: Final = "/agents/tool-evaluation/evaluate"
53MCP_SESSION_ID_HEADER: Final = "mcp-session-id"
54DEFENDER_STATUS_EVALUATED: Final = "Evaluated"
55_GATEWAY_OWNED_TOKEN_ERRORS: Final = frozenset(
56 {"invalid_client", "unauthorized_client", "invalid_scope", "invalid_resource"}
57)
58# Entra reports a malformed or unverifiable assertion as ``invalid_client`` too; only its AADSTS50027xx
59# (InvalidJwtToken) sub-codes tell that apart from a bad gateway secret.
60_INVALID_ASSERTION_AADSTS_PREFIX: Final = "50027"
61_AADSTS_CODES_ADAPTER: Final = TypeAdapter(tuple[int, ...])
62_MCP_CALL_TYPES: Final[tuple[str, ...]] = ("mcp_call", "call_mcp_tool")
63_OBO_CACHE_MAX_ENTRIES: Final = 1000
64_DEFAULT_TOKEN_TTL_SECONDS: Final = 3599.0
65_TOKEN_EXPIRY_SLACK_SECONDS: Final = 60.0
68def _parse_expires_in(raw: object) -> float:
69 if not isinstance(raw, (int, float, str)):
70 return _DEFAULT_TOKEN_TTL_SECONDS
71 try:
72 return float(raw)
73 except ValueError:
74 return _DEFAULT_TOKEN_TTL_SECONDS
77def _parse_aadsts_codes(raw: object) -> tuple[int, ...]:
78 try:
79 return _AADSTS_CODES_ADAPTER.validate_python(raw)
80 except ValidationError:
81 return ()
84def entra_assertion(value: object) -> str | None:
85 """``value`` when it is a compact JWS, the only bearer shape the OBO exchange accepts as its assertion.
86 A LiteLLM virtual key, session bearer, or opaque upstream token in ``Authorization`` yields ``None``."""
87 return value if isinstance(value, str) and value.count(".") == 2 else None
90class _DefenderResult(TypedDict, total=False):
91 status: ReadOnly[str]
92 verdict: ReadOnly[str | None]
93 message: ReadOnly[str | None]
96class _EvaluateResponse(TypedDict, total=False):
97 allowed: ReadOnly[bool]
98 defender: ReadOnly[_DefenderResult]
99 correlationId: ReadOnly[str]
102class _UnavailableDetail(TypedDict):
103 error: ReadOnly[str]
104 message: ReadOnly[str]
105 tool: ReadOnly[str]
108class _BlockedDetail(TypedDict):
109 error: ReadOnly[str]
110 message: ReadOnly[str]
111 tool: ReadOnly[str]
112 correlation_id: ReadOnly[str | None]
115class Agent365TokenExchangeError(Exception):
116 def __init__(self, status_code: int, error_code: str, description: str, aadsts_codes: tuple[int, ...] = ()) -> None:
117 super().__init__(f"{error_code}: {description}")
118 self.status_code = status_code
119 self.error_code = error_code
120 self.description = description
121 self.aadsts_codes = aadsts_codes
123 @property
124 def gateway_owned(self) -> bool:
125 """Whether the gateway's own client credentials, scope or resource were refused, as opposed to the
126 caller's assertion. The caller cannot fix a gateway-owned rejection by signing in again."""
127 if self.error_code not in _GATEWAY_OWNED_TOKEN_ERRORS:
128 return False
129 return not any(str(code).startswith(_INVALID_ASSERTION_AADSTS_PREFIX) for code in self.aadsts_codes)
132class Agent365MalformedResponseError(Exception):
133 pass
136class Agent365ThrottledError(Exception):
137 def __init__(self, status_code: int) -> None:
138 super().__init__(f"HTTP {status_code}")
139 self.status_code = status_code
142class Agent365Guardrail(CustomGuardrail):
143 """Pre-MCP-call guardrail enforcing Microsoft Agent 365 tool-evaluation verdicts.
145 Block-only: it never rewrites the call, so it runs in the post-sequential phase and judges the
146 arguments the sequential guardrails hand upstream, whatever order the guardrails list uses."""
148 records_own_guardrail_information: ClassVar[bool] = True
150 def __init__(
151 self,
152 guardrail_name: str,
153 tenant_id: str,
154 client_id: str,
155 client_secret: str,
156 api_base: str = AGENT_365_PROD_API_BASE,
157 resource_app_id: str = AGENT_365_PROD_RESOURCE_APP_ID,
158 agent_id: str | None = None,
159 request_timeout: float = 10.0,
160 unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
161 async_handler: AsyncHTTPHandler | None = None,
162 **kwargs, # noqa: ANN003 # kwargs-ok: forwarded verbatim to CustomGuardrail (event_hook, default_on)
163 ) -> None:
164 super().__init__(
165 guardrail_name=guardrail_name,
166 supported_event_hooks=self.get_supported_event_hooks(),
167 run_in_parallel=True,
168 **kwargs,
169 )
170 self.guardrail_provider = "agent_365"
171 self.tenant_id = tenant_id
172 self.client_id = client_id
173 self.client_secret = client_secret
174 self.api_base = api_base.rstrip("/")
175 self.resource_app_id = resource_app_id
176 self.agent_id = agent_id
177 self.request_timeout = request_timeout
178 self.unreachable_fallback: Literal["fail_closed", "fail_open"] = (
179 "fail_open" if unreachable_fallback == "fail_open" else "fail_closed"
180 )
181 self.async_handler = async_handler or get_async_httpx_client(
182 llm_provider=httpxSpecialProvider.GuardrailCallback
183 )
184 self._obo_token_cache: OrderedDict[str, tuple[str, float]] = OrderedDict() # mutable-ok: lock-guarded LRU
185 self._obo_cache_lock = threading.Lock()
186 verbose_proxy_logger.info("Initialized Microsoft Agent 365 guardrail: %s", guardrail_name)
188 @staticmethod
189 def get_config_model() -> "type[GuardrailConfigModel] | None":
190 return Agent365GuardrailConfigModel
192 @classmethod
193 def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]: # mutable-ok: CustomGuardrail contract
194 return [GuardrailEventHooks.pre_mcp_call] # mutable-ok: CustomGuardrail contract expects a list
196 @log_guardrail_information
197 async def async_pre_call_hook(
198 self,
199 user_api_key_dict: "UserAPIKeyAuth",
200 cache: "DualCache",
201 data: dict, # mutable-ok: hook contract; guardrail logging appends into the request metadata in place
202 call_type: str,
203 ) -> Exception | str | dict | None: # mutable-ok: CustomGuardrail.async_pre_call_hook contract
204 if call_type not in _MCP_CALL_TYPES:
205 return data
206 if "mcp_tool_name" not in data:
207 return data
208 if self.should_run_guardrail(data=data, event_type=GuardrailEventHooks.pre_mcp_call) is not True:
209 return data
211 tool_name: Final = str(data.get("mcp_tool_name") or "")
212 assertion: Final = entra_assertion(data.get("incoming_bearer_token"))
213 if assertion is None:
214 self._handle_caller_fault(
215 data=data,
216 tool_name=tool_name,
217 status_code=401,
218 reason=(
219 "the caller did not present an Entra bearer token; the Agent 365 guardrail "
220 "authorizes tool calls On-Behalf-Of the signed-in user"
221 ),
222 )
224 try:
225 obo_token: Final = await self._get_obo_token(assertion)
226 except Agent365TokenExchangeError as exc:
227 if exc.gateway_owned:
228 return self._handle_unavailable(
229 data=data,
230 tool_name=tool_name,
231 reason=(
232 f"Entra rejected the gateway's own Agent 365 credentials ({exc.error_code}); "
233 "check the guardrail's client_id, client_secret and resource_app_id"
234 ),
235 )
236 self._handle_caller_fault(
237 data=data,
238 tool_name=tool_name,
239 status_code=401,
240 reason=f"the Entra On-Behalf-Of token exchange was rejected ({exc.error_code})",
241 )
242 except Agent365ThrottledError as exc:
243 self._handle_throttled(
244 data=data,
245 tool_name=tool_name,
246 reason=f"the Entra token endpoint returned HTTP {exc.status_code}",
247 latency_ms=None,
248 )
249 except (httpx.HTTPError, LitellmTimeout, TimeoutError) as exc:
250 return self._handle_unavailable(
251 data=data,
252 tool_name=tool_name,
253 reason=f"the Entra token endpoint could not be reached ({type(exc).__name__})",
254 )
255 except Agent365MalformedResponseError as exc:
256 return self._handle_unavailable(
257 data=data,
258 tool_name=tool_name,
259 reason=str(exc),
260 )
262 start: Final = time.perf_counter()
263 try:
264 response: Final = await self._post_allowing_error_status(
265 url=f"{self.api_base}{EVALUATE_PATH}",
266 json=self._build_evaluate_payload(data=data, user_api_key_dict=user_api_key_dict),
267 headers={"Authorization": f"Bearer {obo_token}"}, # mutable-ok: httpx header dict
268 )
269 except (httpx.HTTPError, LitellmTimeout, TimeoutError) as exc:
270 return self._handle_unavailable(
271 data=data,
272 tool_name=tool_name,
273 reason=f"the Agent 365 endpoint could not be reached ({type(exc).__name__})",
274 )
275 latency_ms: Final = (time.perf_counter() - start) * 1000.0
276 fallback: Final = self._handle_evaluate_error(
277 data=data, tool_name=tool_name, assertion=assertion, response=response, latency_ms=latency_ms
278 )
279 if fallback is not None:
280 return fallback
281 return self._enforce_verdict(data=data, tool_name=tool_name, response=response, latency_ms=latency_ms)
283 def _handle_evaluate_error(
284 self,
285 data: dict, # mutable-ok: guardrail logging appends into the request metadata in place
286 tool_name: str,
287 assertion: str,
288 response: httpx.Response,
289 latency_ms: float,
290 ) -> dict | None: # mutable-ok: returns the request data dict per hook contract on fail_open
291 if response.status_code in (408, 429):
292 self._handle_throttled(
293 data=data,
294 tool_name=tool_name,
295 reason=f"the Agent 365 endpoint returned HTTP {response.status_code}",
296 latency_ms=latency_ms,
297 )
298 if 400 <= response.status_code < 500:
299 if response.status_code == 401:
300 self._evict_obo_token(assertion)
301 self._record_verdict(
302 data=data,
303 verdict="Rejected",
304 guardrail_status="guardrail_intervened",
305 defender_status=None,
306 correlation_id=None,
307 latency_ms=latency_ms,
308 reason=f"HTTP {response.status_code}: {response.text[:512]}",
309 )
310 rejected_detail: Final[_UnavailableDetail] = {
311 "error": "Agent 365 rejected the tool evaluation request",
312 "message": response.text[:512]
313 if response.status_code == 400
314 else f"the Agent 365 evaluation request failed with HTTP {response.status_code}",
315 "tool": tool_name,
316 }
317 raise HTTPException(status_code=400, detail=rejected_detail)
318 if response.status_code != 200:
319 return self._handle_unavailable(
320 data=data,
321 tool_name=tool_name,
322 reason=f"the Agent 365 endpoint returned HTTP {response.status_code}",
323 )
324 return None
326 def _enforce_verdict(
327 self,
328 data: dict, # mutable-ok: guardrail logging appends into the request metadata in place
329 tool_name: str,
330 response: httpx.Response,
331 latency_ms: float,
332 ) -> dict: # mutable-ok: returns the request data dict per hook contract
333 try:
334 parsed_verdict: Final = response.json()
335 except ValueError:
336 return self._handle_unavailable(
337 data=data,
338 tool_name=tool_name,
339 reason="the Agent 365 endpoint returned a non-JSON body",
340 )
341 if not isinstance(parsed_verdict, dict):
342 return self._handle_unavailable(
343 data=data,
344 tool_name=tool_name,
345 reason="the Agent 365 endpoint returned a non-object JSON body",
346 )
347 verdict: Final[_EvaluateResponse] = parsed_verdict
348 allowed: Final = verdict.get("allowed")
349 if not isinstance(allowed, bool):
350 return self._handle_unavailable(
351 data=data,
352 tool_name=tool_name,
353 reason="the Agent 365 endpoint returned a verdict without a boolean 'allowed' field",
354 )
355 raw_defender: Final = verdict.get("defender")
356 defender: Final = raw_defender if isinstance(raw_defender, dict) else _DefenderResult()
357 raw_correlation_id: Final = verdict.get("correlationId")
358 correlation_id: Final = raw_correlation_id if isinstance(raw_correlation_id, str) else None
359 defender_status: Final = defender.get("status")
360 if allowed and defender_status != DEFENDER_STATUS_EVALUATED:
361 return self._handle_unavailable(
362 data=data,
363 tool_name=tool_name,
364 reason=f"Microsoft Defender did not evaluate the call (defender.status={defender_status or 'missing'})",
365 defender_status=defender_status,
366 correlation_id=correlation_id,
367 latency_ms=latency_ms,
368 )
369 self._record_verdict(
370 data=data,
371 verdict="Allow" if allowed else "Block",
372 guardrail_status="success" if allowed else "guardrail_intervened",
373 defender_status=defender_status,
374 correlation_id=correlation_id,
375 latency_ms=latency_ms,
376 )
377 if not allowed:
378 blocked_detail: Final[_BlockedDetail] = {
379 "error": "Blocked by Microsoft Defender",
380 "message": (
381 defender.get("message")
382 or f"Invocation of '{tool_name}' is blocked by Microsoft Threat Detection policies "
383 "configured by your administrator."
384 ),
385 "tool": tool_name,
386 "correlation_id": correlation_id,
387 }
388 raise HTTPException(status_code=400, detail=blocked_detail)
389 return data
391 def _build_evaluate_payload(
392 self,
393 data: Mapping[str, object],
394 user_api_key_dict: "UserAPIKeyAuth",
395 ) -> dict[str, object]: # mutable-ok: JSON body for AsyncHTTPHandler.post, which requires dict
396 tool_name: Final = str(data.get("mcp_tool_name") or "")
397 arguments: Final = data.get("mcp_arguments")
398 server_name: Final = str(data.get("mcp_server_name") or "litellm")
399 agent_id: Final = self.agent_id or user_api_key_dict.key_alias
400 payload: Final[dict[str, object]] = { # mutable-ok: JSON body with optional fields added below
401 "tool": {"name": tool_name},
402 "serverName": server_name,
403 "conversationId": self._resolve_conversation_id(data),
404 }
405 if isinstance(arguments, dict):
406 payload["arguments"] = arguments
407 if agent_id:
408 payload["agentId"] = str(agent_id)
409 return payload
411 @staticmethod
412 def _resolve_conversation_id(data: Mapping[str, object]) -> str:
413 """The MCP session groups every tool call of one client conversation, so it is the conversation id
414 when the transport carries one; stateless calls fall back to the per-call id."""
415 raw_logging_obj: Final = data.get("litellm_logging_obj")
416 logging_obj: Final = raw_logging_obj if isinstance(raw_logging_obj, LiteLLMLoggingObj) else None
417 if logging_obj is not None:
418 tool_call_metadata: Final = logging_obj.model_call_details.get("mcp_tool_call_metadata")
419 session_from_logging: Final = (
420 tool_call_metadata.get("mcp_session_id") if isinstance(tool_call_metadata, Mapping) else None
421 )
422 if isinstance(session_from_logging, str) and session_from_logging:
423 return session_from_logging
424 metadata: Final = next(
425 (m for m in (data.get("metadata"), data.get("litellm_metadata")) if isinstance(m, Mapping)),
426 None,
427 )
428 headers: Final = metadata.get("headers") if isinstance(metadata, Mapping) else None
429 if isinstance(headers, Mapping):
430 session_id: Final = next(
431 (value for name, value in headers.items() if str(name).lower() == MCP_SESSION_ID_HEADER),
432 None,
433 )
434 if isinstance(session_id, str) and session_id:
435 return session_id
436 call_id: Final = data.get("litellm_call_id") or (logging_obj.litellm_call_id if logging_obj else None)
437 if isinstance(call_id, str) and call_id:
438 return call_id
439 return str(uuid.uuid4())
441 async def _get_obo_token(self, assertion: str) -> str:
442 cache_key: Final = hashlib.sha256(assertion.encode("utf-8")).hexdigest()
443 now: Final = time.time()
444 with self._obo_cache_lock:
445 cached: Final = self._obo_token_cache.get(cache_key)
446 if cached and cached[1] > now + _TOKEN_EXPIRY_SLACK_SECONDS:
447 self._obo_token_cache.move_to_end(cache_key)
448 return cached[0]
450 response: Final = await self._post_allowing_error_status(
451 url=TOKEN_ENDPOINT_TEMPLATE.format(tenant_id=self.tenant_id),
452 data={ # mutable-ok: OAuth form body; AsyncHTTPHandler.post requires dict
453 "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
454 "client_id": self.client_id,
455 "client_secret": self.client_secret,
456 "assertion": assertion,
457 "scope": f"{self.resource_app_id}/{AGENT_365_SCOPE_NAME}",
458 "requested_token_use": "on_behalf_of",
459 },
460 headers={"Content-Type": "application/x-www-form-urlencoded"}, # mutable-ok: httpx header dict
461 )
462 if response.status_code in (408, 429):
463 raise Agent365ThrottledError(status_code=response.status_code)
464 if response.status_code >= 500:
465 raise httpx.HTTPStatusError(
466 f"Entra token endpoint returned {response.status_code}",
467 request=response.request,
468 response=response,
469 )
470 try:
471 parsed_body: Final = response.json()
472 except ValueError as exc:
473 raise Agent365MalformedResponseError("the Entra token endpoint returned a non-JSON body") from exc
474 if not isinstance(parsed_body, dict):
475 raise Agent365MalformedResponseError("the Entra token endpoint returned a non-object JSON body")
476 body: Final = parsed_body
477 if response.status_code >= 400:
478 raise Agent365TokenExchangeError(
479 status_code=response.status_code,
480 error_code=str(body.get("error", "invalid_grant")),
481 description=str(body.get("error_description", ""))[:512],
482 aadsts_codes=_parse_aadsts_codes(body.get("error_codes")),
483 )
484 if "access_token" not in body:
485 raise Agent365MalformedResponseError("the Entra token endpoint returned no access_token")
486 raw_access_token: Final = body.get("access_token")
487 if not isinstance(raw_access_token, str) or not raw_access_token:
488 raise Agent365MalformedResponseError("the Entra token endpoint returned a non-string access_token")
489 access_token: Final = raw_access_token
490 expires_at: Final = time.time() + _parse_expires_in(body.get("expires_in", 3599))
491 with self._obo_cache_lock:
492 self._obo_token_cache[cache_key] = (access_token, expires_at)
493 self._obo_token_cache.move_to_end(cache_key)
494 while len(self._obo_token_cache) > _OBO_CACHE_MAX_ENTRIES:
495 self._obo_token_cache.popitem(last=False)
496 return access_token
498 async def _post_allowing_error_status(
499 self,
500 url: str,
501 headers: dict[str, str], # mutable-ok: AsyncHTTPHandler.post requires dict
502 data: dict[str, str] | None = None, # mutable-ok: AsyncHTTPHandler.post requires dict
503 json: dict[str, object] | None = None, # mutable-ok: AsyncHTTPHandler.post requires dict
504 ) -> httpx.Response:
505 try:
506 return await self.async_handler.post(
507 url=url,
508 data=data,
509 json=json,
510 headers=headers,
511 timeout=self.request_timeout,
512 )
513 except httpx.HTTPStatusError as exc:
514 return exc.response
516 def _handle_caller_fault(
517 self,
518 data: dict, # mutable-ok: guardrail logging appends into the request metadata in place
519 tool_name: str,
520 status_code: int,
521 reason: str,
522 ) -> NoReturn:
523 self._record_verdict(
524 data=data,
525 verdict="Rejected",
526 guardrail_status="guardrail_intervened",
527 defender_status=None,
528 correlation_id=None,
529 latency_ms=None,
530 reason=reason,
531 )
532 caller_fault_detail: Final[_UnavailableDetail] = {
533 "error": "Agent 365 guardrail rejected the tool call",
534 "message": f"Tool call '{tool_name}' was blocked because {reason}.",
535 "tool": tool_name,
536 }
537 raise HTTPException(status_code=status_code, detail=caller_fault_detail)
539 def _handle_throttled(
540 self,
541 data: dict, # mutable-ok: guardrail logging appends into the request metadata in place
542 tool_name: str,
543 reason: str,
544 latency_ms: float | None,
545 ) -> NoReturn:
546 self._record_verdict(
547 data=data,
548 verdict="Throttled",
549 guardrail_status="guardrail_failed_to_respond",
550 defender_status=None,
551 correlation_id=None,
552 latency_ms=latency_ms,
553 reason=reason,
554 )
555 throttled_detail: Final[_UnavailableDetail] = {
556 "error": "Agent 365 guardrail could not authorize the tool call",
557 "message": f"Tool call '{tool_name}' was blocked because {reason}; "
558 "throttled evaluations block regardless of unreachable_fallback.",
559 "tool": tool_name,
560 }
561 raise HTTPException(status_code=503, detail=throttled_detail)
563 def _evict_obo_token(self, assertion: str) -> None:
564 cache_key: Final = hashlib.sha256(assertion.encode("utf-8")).hexdigest()
565 with self._obo_cache_lock:
566 self._obo_token_cache.pop(cache_key, None)
568 def _handle_unavailable(
569 self,
570 data: dict, # mutable-ok: guardrail logging appends into the request metadata in place
571 tool_name: str,
572 reason: str,
573 defender_status: str | None = None,
574 correlation_id: str | None = None,
575 latency_ms: float | None = None,
576 ) -> dict: # mutable-ok: returns the request data dict per hook contract
577 if self.unreachable_fallback == "fail_open":
578 verbose_proxy_logger.warning(
579 "Agent 365 guardrail (%s): %s; unreachable_fallback='fail_open', allowing tool call '%s' unscanned",
580 self.guardrail_name,
581 reason,
582 tool_name,
583 )
584 self._record_verdict(
585 data=data,
586 verdict="Unscanned",
587 guardrail_status="guardrail_failed_to_respond",
588 defender_status=defender_status,
589 correlation_id=correlation_id,
590 latency_ms=latency_ms,
591 reason=reason,
592 )
593 return data
594 self._record_verdict(
595 data=data,
596 verdict="Unavailable",
597 guardrail_status="guardrail_failed_to_respond",
598 defender_status=defender_status,
599 correlation_id=correlation_id,
600 latency_ms=latency_ms,
601 reason=reason,
602 )
603 unavailable_detail: Final[_UnavailableDetail] = {
604 "error": "Agent 365 guardrail could not authorize the tool call",
605 "message": f"Tool call '{tool_name}' was blocked because {reason} and unreachable_fallback is "
606 "'fail_closed'.",
607 "tool": tool_name,
608 }
609 raise HTTPException(status_code=503, detail=unavailable_detail)
611 def _record_verdict(
612 self,
613 data: dict[str, object], # mutable-ok: standard guardrail logging appends into the request metadata in place
614 verdict: str,
615 guardrail_status: "GuardrailStatus",
616 defender_status: str | None,
617 correlation_id: str | None,
618 latency_ms: float | None,
619 reason: str | None = None,
620 ) -> None:
621 payload: Final[dict[str, object]] = {"verdict": verdict} # mutable-ok: optional fields added below
622 if defender_status:
623 payload["defender_status"] = defender_status
624 if correlation_id:
625 payload["correlation_id"] = correlation_id
626 if latency_ms is not None:
627 payload["latency_ms"] = round(latency_ms, 1)
628 if reason:
629 payload["reason"] = reason
630 self.add_standard_logging_guardrail_information_to_request_data(
631 guardrail_json_response=payload,
632 request_data=data,
633 guardrail_status=guardrail_status,
634 duration=(latency_ms / 1000.0) if latency_ms is not None else None,
635 guardrail_provider=self.guardrail_provider,
636 event_type=GuardrailEventHooks.pre_mcp_call,
637 )