Merge 28467d404a into 1df25e26cf

litellm/proxy/guardrails/guardrail_hooks/agent_365/__init__.py

@@ -0,0 +1,61 @@
+from typing import TYPE_CHECKING, Final
+
+from litellm.types.guardrails import SupportedGuardrailIntegrations
+from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import (
+    AGENT_365_PROD_API_BASE,
+    AGENT_365_PROD_RESOURCE_APP_ID,
+)
+
+from .agent_365 import Agent365Guardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+
+def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail") -> Agent365Guardrail:
+    import litellm
+    from litellm.secret_managers.main import get_secret_str
+
+    tenant_id: Final = litellm_params.tenant_id or get_secret_str("AGENT365_TENANT_ID")
+    client_id: Final = litellm_params.client_id or get_secret_str("AGENT365_CLIENT_ID")
+    client_secret: Final = (
+        litellm_params.client_secret or litellm_params.api_key or get_secret_str("AGENT365_CLIENT_SECRET")
+    )
+    api_base: Final = litellm_params.api_base or get_secret_str("AGENT365_API_BASE")
+    resource_app_id: Final = litellm_params.resource_app_id or get_secret_str("AGENT365_RESOURCE_APP_ID")
+
+    if not tenant_id:
+        raise ValueError("Microsoft Agent 365: tenant_id is required")
+    if not client_id:
+        raise ValueError("Microsoft Agent 365: client_id is required")
+    if not client_secret:
+        raise ValueError("Microsoft Agent 365: client_secret (or api_key) is required")
+
+    guardrail_name: Final = guardrail.get("guardrail_name")
+    if not guardrail_name:
+        raise ValueError("Microsoft Agent 365: guardrail_name is required")
+
+    agent_365_guardrail: Final = Agent365Guardrail(
+        guardrail_name=guardrail_name,
+        tenant_id=tenant_id,
+        client_id=client_id,
+        client_secret=client_secret,
+        api_base=api_base or AGENT_365_PROD_API_BASE,
+        resource_app_id=resource_app_id or AGENT_365_PROD_RESOURCE_APP_ID,
+        agent_id=litellm_params.agent_id,
+        request_timeout=litellm_params.timeout if litellm_params.timeout is not None else 10.0,
+        unreachable_fallback=litellm_params.unreachable_fallback,
+        event_hook=litellm_params.mode,
+        default_on=litellm_params.default_on,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(agent_365_guardrail)
+    return agent_365_guardrail
+
+
+guardrail_initializer_registry: Final = {  # mutable-ok: registry auto-discovery requires a dict instance
+    SupportedGuardrailIntegrations.AGENT_365.value: initialize_guardrail,
+}
+
+guardrail_class_registry: Final = {  # mutable-ok: registry auto-discovery requires a dict instance
+    SupportedGuardrailIntegrations.AGENT_365.value: Agent365Guardrail,
+}

litellm/proxy/guardrails/guardrail_hooks/agent_365/agent_365.py

@@ -0,0 +1,575 @@
+"""Microsoft Agent 365 governance guardrail for MCP tool calls.
+
+Before the gateway executes an MCP tool, the pending call is sent to the
+Agent 365 tool-evaluation endpoint, where Microsoft Defender scores it and
+Agent 365 records it for observability. The returned allow/block verdict is
+enforced here. Authentication is the Entra On-Behalf-Of flow: the caller's
+incoming bearer token (audienced to this gateway's app registration) is
+exchanged for a delegated Agent 365 token, so Defender evaluates and audits
+as the signed-in user.
+"""
+
+import hashlib
+import threading
+import time
+import uuid
+from collections import OrderedDict
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, ClassVar, Final, Literal, NoReturn
+
+import httpx
+from fastapi import HTTPException
+from typing_extensions import ReadOnly, TypedDict
+
+from litellm._logging import verbose_proxy_logger
+from litellm.exceptions import Timeout as LitellmTimeout
+from litellm.integrations.custom_guardrail import (
+    CustomGuardrail,
+    log_guardrail_information,
+)
+from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import (
+    AGENT_365_PROD_API_BASE,
+    AGENT_365_PROD_RESOURCE_APP_ID,
+    AGENT_365_SCOPE_NAME,
+    Agent365GuardrailConfigModel,
+)
+
+if TYPE_CHECKING:
+    from litellm.caching.caching import DualCache
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
+    from litellm.types.utils import GuardrailStatus
+
+TOKEN_ENDPOINT_TEMPLATE: Final = "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
+EVALUATE_PATH: Final = "/agents/tool-evaluation/evaluate"
+MCP_SESSION_ID_HEADER: Final = "mcp-session-id"
+_MCP_CALL_TYPES: Final[tuple[str, ...]] = ("mcp_call", "call_mcp_tool")
+_OBO_CACHE_MAX_ENTRIES: Final = 1000
+_DEFAULT_TOKEN_TTL_SECONDS: Final = 3599.0
+_TOKEN_EXPIRY_SLACK_SECONDS: Final = 60.0
+
+
+def _parse_expires_in(raw: object) -> float:
+    if not isinstance(raw, (int, float, str)):
+        return _DEFAULT_TOKEN_TTL_SECONDS
+    try:
+        return float(raw)
+    except ValueError:
+        return _DEFAULT_TOKEN_TTL_SECONDS
+
+
+class _DefenderResult(TypedDict, total=False):
+    status: ReadOnly[str]
+    verdict: ReadOnly[str | None]
+    message: ReadOnly[str | None]
+
+
+class _EvaluateResponse(TypedDict, total=False):
+    allowed: ReadOnly[bool]
+    defender: ReadOnly[_DefenderResult]
+    correlationId: ReadOnly[str]
+
+
+class _ToolReference(TypedDict):
+    name: ReadOnly[str]
+
+
+class _UnavailableDetail(TypedDict):
+    error: ReadOnly[str]
+    message: ReadOnly[str]
+    tool: ReadOnly[str]
+
+
+class _BlockedDetail(TypedDict):
+    error: ReadOnly[str]
+    message: ReadOnly[str]
+    tool: ReadOnly[str]
+    correlation_id: ReadOnly[str | None]
+
+
+class Agent365TokenExchangeError(Exception):
+    def __init__(self, status_code: int, error_code: str, description: str) -> None:
+        super().__init__(f"{error_code}: {description}")
+        self.status_code = status_code
+        self.error_code = error_code
+        self.description = description
+
+
+class Agent365MalformedResponseError(Exception):
+    pass
+
+
+class Agent365ThrottledError(Exception):
+    def __init__(self, status_code: int) -> None:
+        super().__init__(f"HTTP {status_code}")
+        self.status_code = status_code
+
+
+class Agent365Guardrail(CustomGuardrail):
+    """Pre-MCP-call guardrail enforcing Microsoft Agent 365 tool-evaluation verdicts."""
+
+    records_own_guardrail_information: ClassVar[bool] = True
+
+    def __init__(
+        self,
+        guardrail_name: str,
+        tenant_id: str,
+        client_id: str,
+        client_secret: str,
+        api_base: str = AGENT_365_PROD_API_BASE,
+        resource_app_id: str = AGENT_365_PROD_RESOURCE_APP_ID,
+        agent_id: str | None = None,
+        request_timeout: float = 10.0,
+        unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
+        async_handler: AsyncHTTPHandler | None = None,
+        **kwargs,  # noqa: ANN003  # kwargs-ok: forwarded verbatim to CustomGuardrail (event_hook, default_on)
+    ) -> None:
+        super().__init__(
+            guardrail_name=guardrail_name,
+            supported_event_hooks=self.get_supported_event_hooks(),
+            **kwargs,
+        )
+        self.guardrail_provider = "agent_365"
+        self.tenant_id = tenant_id
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.api_base = api_base.rstrip("/")
+        self.resource_app_id = resource_app_id
+        self.agent_id = agent_id
+        self.request_timeout = request_timeout
+        self.unreachable_fallback: Literal["fail_closed", "fail_open"] = (
+            "fail_open" if unreachable_fallback == "fail_open" else "fail_closed"
+        )
+        self.async_handler = async_handler or get_async_httpx_client(
+            llm_provider=httpxSpecialProvider.GuardrailCallback
+        )
+        self._obo_token_cache: OrderedDict[str, tuple[str, float]] = OrderedDict()  # mutable-ok: lock-guarded LRU
+        self._obo_cache_lock = threading.Lock()
+        verbose_proxy_logger.info("Initialized Microsoft Agent 365 guardrail: %s", guardrail_name)
+
+    @staticmethod
+    def get_config_model() -> "type[GuardrailConfigModel] | None":
+        return Agent365GuardrailConfigModel
+
+    @classmethod
+    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:  # mutable-ok: CustomGuardrail contract
+        return [GuardrailEventHooks.pre_mcp_call]  # mutable-ok: CustomGuardrail contract expects a list
+
+    @log_guardrail_information
+    async def async_pre_call_hook(
+        self,
+        user_api_key_dict: "UserAPIKeyAuth",
+        cache: "DualCache",
+        data: dict,  # mutable-ok: hook contract; guardrail logging appends into the request metadata in place
+        call_type: str,
+    ) -> Exception | str | dict | None:  # mutable-ok: CustomGuardrail.async_pre_call_hook contract
+        if call_type not in _MCP_CALL_TYPES:
+            return data
+        if "mcp_tool_name" not in data:
+            return data
+        if self.should_run_guardrail(data=data, event_type=GuardrailEventHooks.pre_mcp_call) is not True:
+            return data
+
+        tool_name: Final = str(data.get("mcp_tool_name") or "")
+        assertion: Final = data.get("incoming_bearer_token")
+        if not isinstance(assertion, str) or assertion.count(".") != 2:
+            self._handle_caller_fault(
+                data=data,
+                tool_name=tool_name,
+                status_code=401,
+                reason=(
+                    "the caller did not present an Entra bearer token; the Agent 365 guardrail "
+                    "authorizes tool calls On-Behalf-Of the signed-in user"
+                ),
+            )
+
+        try:
+            obo_token: Final = await self._get_obo_token(assertion)
+        except Agent365TokenExchangeError as exc:
+            self._handle_caller_fault(
+                data=data,
+                tool_name=tool_name,
+                status_code=401,
+                reason=f"the Entra On-Behalf-Of token exchange was rejected ({exc.error_code})",
+            )
+        except Agent365ThrottledError as exc:
+            self._handle_throttled(
+                data=data,
+                tool_name=tool_name,
+                reason=f"the Entra token endpoint returned HTTP {exc.status_code}",
+                latency_ms=None,
+            )
+        except (httpx.HTTPError, LitellmTimeout, TimeoutError) as exc:
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason=f"the Entra token endpoint could not be reached ({type(exc).__name__})",
+            )
+        except Agent365MalformedResponseError as exc:
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason=str(exc),
+            )
+
+        start: Final = time.perf_counter()
+        try:
+            response: Final = await self._post_allowing_error_status(
+                url=f"{self.api_base}{EVALUATE_PATH}",
+                json=self._build_evaluate_payload(data=data, user_api_key_dict=user_api_key_dict),
+                headers={"Authorization": f"Bearer {obo_token}"},  # mutable-ok: httpx header dict
+            )
+        except (httpx.HTTPError, LitellmTimeout, TimeoutError) as exc:
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason=f"the Agent 365 endpoint could not be reached ({type(exc).__name__})",
+            )
+        latency_ms: Final = (time.perf_counter() - start) * 1000.0
+        fallback: Final = self._handle_evaluate_error(
+            data=data, tool_name=tool_name, assertion=assertion, response=response, latency_ms=latency_ms
+        )
+        if fallback is not None:
+            return fallback
+        return self._enforce_verdict(data=data, tool_name=tool_name, response=response, latency_ms=latency_ms)
+
+    def _handle_evaluate_error(
+        self,
+        data: dict,  # mutable-ok: guardrail logging appends into the request metadata in place
+        tool_name: str,
+        assertion: str,
+        response: httpx.Response,
+        latency_ms: float,
+    ) -> dict | None:  # mutable-ok: returns the request data dict per hook contract on fail_open
+        if response.status_code in (408, 429):
+            self._handle_throttled(
+                data=data,
+                tool_name=tool_name,
+                reason=f"the Agent 365 endpoint returned HTTP {response.status_code}",
+                latency_ms=latency_ms,
+            )
+        if 400 <= response.status_code < 500:
+            if response.status_code == 401:
+                self._evict_obo_token(assertion)
+            self._record_verdict(
+                data=data,
+                verdict="Rejected",
+                guardrail_status="guardrail_intervened",
+                defender_status=None,
+                correlation_id=None,
+                latency_ms=latency_ms,
+                reason=f"HTTP {response.status_code}: {response.text[:512]}",
+            )
+            rejected_detail: Final[_UnavailableDetail] = {
+                "error": "Agent 365 rejected the tool evaluation request",
+                "message": response.text[:512]
+                if response.status_code == 400
+                else f"the Agent 365 evaluation request failed with HTTP {response.status_code}",
+                "tool": tool_name,
+            }
+            raise HTTPException(status_code=400, detail=rejected_detail)
+        if response.status_code != 200:
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason=f"the Agent 365 endpoint returned HTTP {response.status_code}",
+            )
+        return None
+
+    def _enforce_verdict(
+        self,
+        data: dict,  # mutable-ok: guardrail logging appends into the request metadata in place
+        tool_name: str,
+        response: httpx.Response,
+        latency_ms: float,
+    ) -> dict:  # mutable-ok: returns the request data dict per hook contract
+        try:
+            parsed_verdict: Final = response.json()
+        except ValueError:
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason="the Agent 365 endpoint returned a non-JSON body",
+            )
+        if not isinstance(parsed_verdict, dict):
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason="the Agent 365 endpoint returned a non-object JSON body",
+            )
+        verdict: Final[_EvaluateResponse] = parsed_verdict
+        raw_defender: Final = verdict.get("defender")
+        defender: Final = raw_defender if isinstance(raw_defender, dict) else _DefenderResult()
+        raw_correlation_id: Final = verdict.get("correlationId")
+        correlation_id: Final = raw_correlation_id if isinstance(raw_correlation_id, str) else None
+        allowed: Final = verdict.get("allowed") is True
+        self._record_verdict(
+            data=data,
+            verdict="Allow" if allowed else "Block",
+            guardrail_status="success" if allowed else "guardrail_intervened",
+            defender_status=defender.get("status"),
+            correlation_id=correlation_id,
+            latency_ms=latency_ms,
+        )
+        if not allowed:
+            blocked_detail: Final[_BlockedDetail] = {
+                "error": "Blocked by Microsoft Defender",
+                "message": (
+                    defender.get("message")
+                    or f"Invocation of '{tool_name}' is blocked by Microsoft Threat Detection policies "
+                    "configured by your administrator."
+                ),
+                "tool": tool_name,
+                "correlation_id": correlation_id,
+            }
+            raise HTTPException(status_code=400, detail=blocked_detail)
+        return data
+
+    def _build_evaluate_payload(
+        self,
+        data: Mapping[str, object],
+        user_api_key_dict: "UserAPIKeyAuth",
+    ) -> dict[str, object]:  # mutable-ok: JSON body for AsyncHTTPHandler.post, which requires dict
+        tool_name: Final = str(data.get("mcp_tool_name") or "")
+        arguments: Final = data.get("mcp_arguments")
+        server_name: Final = str(data.get("mcp_server_name") or "litellm")
+        agent_id: Final = self.agent_id or getattr(user_api_key_dict, "key_alias", None)
+        tool_reference: Final[_ToolReference] = {"name": tool_name}
+        payload: Final[dict[str, object]] = {  # mutable-ok: JSON body with optional fields added below
+            "tool": tool_reference,
+            "serverName": server_name,
+            "conversationId": self._resolve_conversation_id(data),
+        }
+        if isinstance(arguments, dict):
+            payload["arguments"] = arguments
+        if agent_id:
+            payload["agentId"] = str(agent_id)
+        return payload
+
+    @staticmethod
+    def _resolve_conversation_id(data: Mapping[str, object]) -> str:
+        metadata: Final = next(
+            (m for m in (data.get("metadata"), data.get("litellm_metadata")) if isinstance(m, Mapping)),
+            None,
+        )
+        headers: Final = metadata.get("headers") if isinstance(metadata, Mapping) else None
+        if isinstance(headers, Mapping):
+            session_id: Final = next(
+                (value for name, value in headers.items() if str(name).lower() == MCP_SESSION_ID_HEADER),
+                None,
+            )
+            if isinstance(session_id, str) and session_id:
+                return session_id
+        logging_obj: Final = data.get("litellm_logging_obj")
+        call_details: Final = getattr(logging_obj, "model_call_details", None)
+        if isinstance(call_details, Mapping):
+            tool_call_metadata: Final = call_details.get("mcp_tool_call_metadata")
+            session_from_logging: Final = (
+                tool_call_metadata.get("mcp_session_id") if isinstance(tool_call_metadata, Mapping) else None
+            )
+            if isinstance(session_from_logging, str) and session_from_logging:
+                return session_from_logging
+        call_id: Final = data.get("litellm_call_id") or getattr(logging_obj, "litellm_call_id", None)
+        if isinstance(call_id, str) and call_id:
+            return call_id
+        return str(uuid.uuid4())
+
+    async def _get_obo_token(self, assertion: str) -> str:
+        cache_key: Final = hashlib.sha256(assertion.encode("utf-8")).hexdigest()
+        now: Final = time.time()
+        with self._obo_cache_lock:
+            cached: Final = self._obo_token_cache.get(cache_key)
+            if cached and cached[1] > now + _TOKEN_EXPIRY_SLACK_SECONDS:
+                self._obo_token_cache.move_to_end(cache_key)
+                return cached[0]
+
+        response: Final = await self._post_allowing_error_status(
+            url=TOKEN_ENDPOINT_TEMPLATE.format(tenant_id=self.tenant_id),
+            data={  # mutable-ok: OAuth form body; AsyncHTTPHandler.post requires dict
+                "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+                "client_id": self.client_id,
+                "client_secret": self.client_secret,
+                "assertion": assertion,
+                "scope": f"{self.resource_app_id}/{AGENT_365_SCOPE_NAME}",
+                "requested_token_use": "on_behalf_of",
+            },
+            headers={"Content-Type": "application/x-www-form-urlencoded"},  # mutable-ok: httpx header dict
+        )
+        if response.status_code in (408, 429):
+            raise Agent365ThrottledError(status_code=response.status_code)
+        if response.status_code >= 500:
+            raise httpx.HTTPStatusError(
+                f"Entra token endpoint returned {response.status_code}",
+                request=response.request,
+                response=response,
+            )
+        try:
+            parsed_body: Final = response.json()
+        except ValueError as exc:
+            raise Agent365MalformedResponseError("the Entra token endpoint returned a non-JSON body") from exc
+        if not isinstance(parsed_body, dict):
+            raise Agent365MalformedResponseError("the Entra token endpoint returned a non-object JSON body")
+        body: Final = parsed_body
+        if response.status_code >= 400:
+            raise Agent365TokenExchangeError(
+                status_code=response.status_code,
+                error_code=str(body.get("error", "invalid_grant")),
+                description=str(body.get("error_description", ""))[:512],
+            )
+        if "access_token" not in body:
+            raise Agent365MalformedResponseError("the Entra token endpoint returned no access_token")
+        raw_access_token: Final = body.get("access_token")
+        if not isinstance(raw_access_token, str) or not raw_access_token:
+            raise Agent365MalformedResponseError("the Entra token endpoint returned a non-string access_token")
+        access_token: Final = raw_access_token
+        expires_at: Final = time.time() + _parse_expires_in(body.get("expires_in", 3599))
+        with self._obo_cache_lock:
+            self._obo_token_cache[cache_key] = (access_token, expires_at)
+            self._obo_token_cache.move_to_end(cache_key)
+            while len(self._obo_token_cache) > _OBO_CACHE_MAX_ENTRIES:
+                self._obo_token_cache.popitem(last=False)
+        return access_token
+
+    async def _post_allowing_error_status(
+        self,
+        url: str,
+        headers: dict[str, str],  # mutable-ok: AsyncHTTPHandler.post requires dict
+        data: dict[str, str] | None = None,  # mutable-ok: AsyncHTTPHandler.post requires dict
+        json: dict[str, object] | None = None,  # mutable-ok: AsyncHTTPHandler.post requires dict
+    ) -> httpx.Response:
+        try:
+            return await self.async_handler.post(
+                url=url,
+                data=data,
+                json=json,
+                headers=headers,
+                timeout=self.request_timeout,
+            )
+        except httpx.HTTPStatusError as exc:
+            return exc.response
+
+    def _handle_caller_fault(
+        self,
+        data: dict,  # mutable-ok: guardrail logging appends into the request metadata in place
+        tool_name: str,
+        status_code: int,
+        reason: str,
+    ) -> NoReturn:
+        self._record_verdict(
+            data=data,
+            verdict="Rejected",
+            guardrail_status="guardrail_intervened",
+            defender_status=None,
+            correlation_id=None,
+            latency_ms=None,
+            reason=reason,
+        )
+        caller_fault_detail: Final[_UnavailableDetail] = {
+            "error": "Agent 365 guardrail rejected the tool call",
+            "message": f"Tool call '{tool_name}' was blocked because {reason}.",
+            "tool": tool_name,
+        }
+        raise HTTPException(status_code=status_code, detail=caller_fault_detail)
+
+    def _handle_throttled(
+        self,
+        data: dict,  # mutable-ok: guardrail logging appends into the request metadata in place
+        tool_name: str,
+        reason: str,
+        latency_ms: float | None,
+    ) -> NoReturn:
+        self._record_verdict(
+            data=data,
+            verdict="Throttled",
+            guardrail_status="guardrail_failed_to_respond",
+            defender_status=None,
+            correlation_id=None,
+            latency_ms=latency_ms,
+            reason=reason,
+        )
+        throttled_detail: Final[_UnavailableDetail] = {
+            "error": "Agent 365 guardrail could not authorize the tool call",
+            "message": f"Tool call '{tool_name}' was blocked because {reason}; "
+            "throttled evaluations block regardless of unreachable_fallback.",
+            "tool": tool_name,
+        }
+        raise HTTPException(status_code=503, detail=throttled_detail)
+
+    def _evict_obo_token(self, assertion: str) -> None:
+        cache_key: Final = hashlib.sha256(assertion.encode("utf-8")).hexdigest()
+        with self._obo_cache_lock:
+            self._obo_token_cache.pop(cache_key, None)
+
+    def _handle_unavailable(
+        self,
+        data: dict,  # mutable-ok: guardrail logging appends into the request metadata in place
+        tool_name: str,
+        reason: str,
+    ) -> dict:  # mutable-ok: returns the request data dict per hook contract
+        if self.unreachable_fallback == "fail_open":
+            verbose_proxy_logger.warning(
+                "Agent 365 guardrail (%s): %s; unreachable_fallback='fail_open', allowing tool call '%s' unscanned",
+                self.guardrail_name,
+                reason,
+                tool_name,
+            )
+            self._record_verdict(
+                data=data,
+                verdict="Unscanned",
+                guardrail_status="guardrail_failed_to_respond",
+                defender_status=None,
+                correlation_id=None,
+                latency_ms=None,
+                reason=reason,
+            )
+            return data
+        self._record_verdict(
+            data=data,
+            verdict="Unavailable",
+            guardrail_status="guardrail_failed_to_respond",
+            defender_status=None,
+            correlation_id=None,
+            latency_ms=None,
+            reason=reason,
+        )
+        unavailable_detail: Final[_UnavailableDetail] = {
+            "error": "Agent 365 guardrail could not authorize the tool call",
+            "message": f"Tool call '{tool_name}' was blocked because {reason} and unreachable_fallback is "
+            "'fail_closed'.",
+            "tool": tool_name,
+        }
+        raise HTTPException(status_code=503, detail=unavailable_detail)
+
+    def _record_verdict(
+        self,
+        data: dict[str, object],  # mutable-ok: standard guardrail logging appends into the request metadata in place
+        verdict: str,
+        guardrail_status: "GuardrailStatus",
+        defender_status: str | None,
+        correlation_id: str | None,
+        latency_ms: float | None,
+        reason: str | None = None,
+    ) -> None:
+        payload: Final[dict[str, object]] = {"verdict": verdict}  # mutable-ok: optional fields added below
+        if defender_status:
+            payload["defender_status"] = defender_status
+        if correlation_id:
+            payload["correlation_id"] = correlation_id
+        if latency_ms is not None:
+            payload["latency_ms"] = round(latency_ms, 1)
+        if reason:
+            payload["reason"] = reason
+        self.add_standard_logging_guardrail_information_to_request_data(
+            guardrail_json_response=payload,
+            request_data=data,
+            guardrail_status=guardrail_status,
+            duration=(latency_ms / 1000.0) if latency_ms is not None else None,
+            guardrail_provider=self.guardrail_provider,
+            event_type=GuardrailEventHooks.pre_mcp_call,
+        )

litellm/types/guardrails.py

@@ -6,6 +6,9 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida
 from typing_extensions import Required, TypedDict
 
 from litellm.constants import BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS
+from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import (
+    Agent365GuardrailConfigModel,
+)
 from litellm.types.proxy.guardrails.guardrail_hooks.akto import (
     AktoConfigModel,
 )
@@ -134,6 +137,7 @@ class SupportedGuardrailIntegrations(Enum):
     HEADROOM = "headroom"
     COMPRESR = "compresr"
     STRAIKER = "straiker"
+    AGENT_365 = "agent_365"
 
 
 class Role(Enum):
@@ -872,7 +876,7 @@ class BaseLitellmParams(ContentFilterConfigModel):  # works for new and patch up
         default="fail_closed",
         description=(
             "Behavior when a guardrail endpoint is unreachable due to network errors. "
-            "Implemented by guardrail='generic_guardrail_api', 'akto', 'vigil_guard', 'repelloai', 'headroom', and 'compresr'. "
+            "Implemented by guardrail='generic_guardrail_api', 'agent_365', 'akto', 'vigil_guard', 'repelloai', 'headroom', and 'compresr'. "
             "'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed."
         ),
     )
@@ -999,6 +1003,7 @@ class LitellmParams(
     QostodianNexusConfigModel,
     VigilGuardGuardrailConfigModel,
     SingulrGuardrailConfigModel,
+    Agent365GuardrailConfigModel,
 ):
     guardrail: str = Field(description="The type of guardrail integration to use")
     mode: str | list[str] | Mode = Field(

litellm/types/proxy/guardrails/guardrail_hooks/agent_365.py

@@ -0,0 +1,66 @@
+from typing import Final
+
+from pydantic import Field
+
+from .base import GuardrailConfigModel
+
+AGENT_365_PROD_API_BASE: Final = "https://agent365.svc.cloud.microsoft"
+AGENT_365_PROD_RESOURCE_APP_ID: Final = "ea9ffc3e-8a23-4a7d-836d-234d7c7565c1"
+AGENT_365_SCOPE_NAME: Final = "ThreatProtection.Evaluate.All"
+
+
+class Agent365GuardrailConfigModel(GuardrailConfigModel):
+    tenant_id: str | None = Field(
+        default=None,
+        description=(
+            "Entra tenant id used for the On-Behalf-Of token exchange. "
+            "Falls back to the AGENT365_TENANT_ID environment variable."
+        ),
+    )
+
+    client_id: str | None = Field(
+        default=None,
+        description=(
+            "Client id of the gateway's Entra app registration (a confidential client). "
+            "Falls back to the AGENT365_CLIENT_ID environment variable."
+        ),
+    )
+
+    client_secret: str | None = Field(
+        default=None,
+        description=(
+            "Client secret of the gateway's Entra app registration, used to perform the "
+            "On-Behalf-Of exchange. Falls back to the AGENT365_CLIENT_SECRET environment variable."
+        ),
+    )
+
+    api_base: str | None = Field(
+        default=None,
+        description=(
+            "Base URL of the Microsoft Agent 365 tool-evaluation endpoint. "
+            f"Defaults to the production endpoint {AGENT_365_PROD_API_BASE}. "
+            "Falls back to the AGENT365_API_BASE environment variable."
+        ),
+    )
+
+    resource_app_id: str | None = Field(
+        default=None,
+        description=(
+            "Application id of the Agent 365 resource the OBO token is minted for. "
+            f"Defaults to the production resource {AGENT_365_PROD_RESOURCE_APP_ID}; "
+            "the Test and PreProd environments use a different id. "
+            "Falls back to the AGENT365_RESOURCE_APP_ID environment variable."
+        ),
+    )
+
+    agent_id: str | None = Field(
+        default=None,
+        description=(
+            "Agent identity reported to Agent 365 with every tool evaluation. "
+            "When unset, the caller's key alias is used."
+        ),
+    )
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "Microsoft Agent 365"

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_agent_365.py

@@ -0,0 +1,742 @@
+from types import SimpleNamespace
+from typing import Any, Final
+
+import httpx
+import time
+
+import pytest
+from fastapi import HTTPException
+
+from litellm.exceptions import Timeout as LitellmTimeout
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.guardrails.guardrail_hooks.agent_365 import (
+    Agent365Guardrail,
+    guardrail_class_registry,
+    guardrail_initializer_registry,
+    initialize_guardrail,
+)
+from litellm.types.guardrails import (
+    GuardrailEventHooks,
+    LitellmParams,
+    SupportedGuardrailIntegrations,
+)
+from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import (
+    AGENT_365_PROD_API_BASE,
+    AGENT_365_PROD_RESOURCE_APP_ID,
+    Agent365GuardrailConfigModel,
+)
+
+FAKE_ASSERTION: Final = "eyJhbGciOi.eyJhdWQiOi.c2lnbmF0dXJl"
+TOKEN_URL: Final = "https://login.microsoftonline.com/tenant-abc/oauth2/v2.0/token"
+EVALUATE_URL: Final = f"{AGENT_365_PROD_API_BASE}/agents/tool-evaluation/evaluate"
+
+
+def _response(status_code: int, payload: Any = None, text: str | None = None) -> httpx.Response:
+    request: Final = httpx.Request("POST", "https://example.test")
+    if payload is not None:
+        return httpx.Response(status_code=status_code, json=payload, request=request)
+    return httpx.Response(status_code=status_code, text=text or "", request=request)
+
+
+def _token_response(access_token: str = "obo-access-token", expires_in: int = 3599) -> httpx.Response:
+    return _response(200, {"access_token": access_token, "expires_in": expires_in})
+
+
+def _allow_response(correlation_id: str = "corr-1") -> httpx.Response:
+    return _response(
+        200,
+        {
+            "allowed": True,
+            "defender": {"status": "Evaluated", "verdict": "Allow", "message": None},
+            "observability": {"status": "Recorded"},
+            "correlationId": correlation_id,
+        },
+    )
+
+
+def _block_response(message: str = "Blocked by policy", correlation_id: str = "corr-2") -> httpx.Response:
+    return _response(
+        200,
+        {
+            "allowed": False,
+            "defender": {"status": "Evaluated", "verdict": "Block", "message": message},
+            "correlationId": correlation_id,
+        },
+    )
+
+
+class FakeHandler:
+    def __init__(self, items: list[Any]):
+        self._items = list(items)
+        self.calls: list[SimpleNamespace] = []
+
+    async def post(self, *, url, headers=None, data=None, json=None, timeout=None):
+        self.calls.append(SimpleNamespace(url=url, headers=headers, data=data, json=json, timeout=timeout))
+        if not self._items:
+            raise AssertionError("FakeHandler ran out of programmed responses")
+        item = self._items.pop(0)
+        if isinstance(item, BaseException):
+            raise item
+        if item.status_code >= 400:
+            raise httpx.HTTPStatusError("error status", request=item.request, response=item)
+        return item
+
+
+def _make_guardrail(
+    handler: FakeHandler,
+    *,
+    unreachable_fallback: str = "fail_closed",
+    agent_id: str | None = None,
+    api_base: str = AGENT_365_PROD_API_BASE,
+) -> Agent365Guardrail:
+    return Agent365Guardrail(
+        guardrail_name="agent-365-guard",
+        tenant_id="tenant-abc",
+        client_id="client-xyz",
+        client_secret="secret-123",
+        api_base=api_base,
+        agent_id=agent_id,
+        unreachable_fallback=unreachable_fallback,
+        async_handler=handler,
+        event_hook="pre_mcp_call",
+        default_on=True,
+    )
+
+
+def _mcp_data(**overrides: Any) -> dict:
+    data: Final[dict] = {
+        "mcp_tool_name": "send_email",
+        "mcp_arguments": {"to": "user@example.com", "body": "hello"},
+        "mcp_server_name": "outlook_mcp",
+        "incoming_bearer_token": FAKE_ASSERTION,
+        "metadata": {"headers": {"mcp-session-id": "sess-123"}},
+    }
+    data.update(overrides)
+    return data
+
+
+def _user() -> UserAPIKeyAuth:
+    return UserAPIKeyAuth(api_key="hashed-key", key_alias="my-agent-key")
+
+
+async def _run(guardrail: Agent365Guardrail, data: dict, call_type: str = "call_mcp_tool"):
+    return await guardrail.async_pre_call_hook(
+        user_api_key_dict=_user(),
+        cache=None,
+        data=data,
+        call_type=call_type,
+    )
+
+
+class TestRegistryWiring:
+    def test_enum_member_exists(self):
+        assert SupportedGuardrailIntegrations.AGENT_365.value == "agent_365"
+
+    def test_initializer_registry(self):
+        assert guardrail_initializer_registry["agent_365"] is initialize_guardrail
+
+    def test_class_registry(self):
+        assert guardrail_class_registry["agent_365"] is Agent365Guardrail
+
+    def test_config_model_wired(self):
+        assert Agent365Guardrail.get_config_model() is Agent365GuardrailConfigModel
+        assert Agent365GuardrailConfigModel.ui_friendly_name() == "Microsoft Agent 365"
+
+    def test_supported_event_hooks(self):
+        assert Agent365Guardrail.get_supported_event_hooks() == [GuardrailEventHooks.pre_mcp_call]
+
+
+class TestInitializeGuardrail:
+    def test_requires_tenant_id(self, monkeypatch):
+        monkeypatch.delenv("AGENT365_TENANT_ID", raising=False)
+        params: Final = LitellmParams(
+            guardrail="agent_365",
+            mode="pre_mcp_call",
+            client_id="client-xyz",
+            api_key="secret-123",
+        )
+        with pytest.raises(ValueError, match="tenant_id is required"):
+            initialize_guardrail(params, {"guardrail_name": "a365"})
+
+    def test_requires_client_secret(self, monkeypatch):
+        monkeypatch.delenv("AGENT365_CLIENT_SECRET", raising=False)
+        params: Final = LitellmParams(
+            guardrail="agent_365",
+            mode="pre_mcp_call",
+            tenant_id="tenant-abc",
+            client_id="client-xyz",
+        )
+        with pytest.raises(ValueError, match="client_secret"):
+            initialize_guardrail(params, {"guardrail_name": "a365"})
+
+    def test_env_var_fallbacks(self, monkeypatch):
+        monkeypatch.delenv("AGENT365_RESOURCE_APP_ID", raising=False)
+        monkeypatch.setenv("AGENT365_TENANT_ID", "env-tenant")
+        monkeypatch.setenv("AGENT365_CLIENT_ID", "env-client")
+        monkeypatch.setenv("AGENT365_CLIENT_SECRET", "env-secret")
+        monkeypatch.setenv("AGENT365_API_BASE", "https://env.example.test")
+        params: Final = LitellmParams(guardrail="agent_365", mode="pre_mcp_call")
+        guardrail: Final = initialize_guardrail(params, {"guardrail_name": "a365-env"})
+        assert guardrail.tenant_id == "env-tenant"
+        assert guardrail.client_id == "env-client"
+        assert guardrail.client_secret == "env-secret"
+        assert guardrail.api_base == "https://env.example.test"
+        assert guardrail.resource_app_id == AGENT_365_PROD_RESOURCE_APP_ID
+        assert guardrail.unreachable_fallback == "fail_closed"
+
+    def test_explicit_params_win(self, monkeypatch):
+        monkeypatch.setenv("AGENT365_TENANT_ID", "env-tenant")
+        params: Final = LitellmParams(
+            guardrail="agent_365",
+            mode="pre_mcp_call",
+            tenant_id="param-tenant",
+            client_id="client-xyz",
+            client_secret="param-secret",
+            agent_id="agent-007",
+            unreachable_fallback="fail_open",
+            timeout=5,
+        )
+        guardrail: Final = initialize_guardrail(params, {"guardrail_name": "a365-params"})
+        assert guardrail.tenant_id == "param-tenant"
+        assert guardrail.client_secret == "param-secret"
+        assert guardrail.agent_id == "agent-007"
+        assert guardrail.unreachable_fallback == "fail_open"
+        assert guardrail.request_timeout == 5.0
+
+    def test_wrong_mode_rejected(self):
+        params: Final = LitellmParams(
+            guardrail="agent_365",
+            mode="post_call",
+            tenant_id="tenant-abc",
+            client_id="client-xyz",
+            api_key="secret-123",
+        )
+        with pytest.raises(Exception, match="post_call"):
+            initialize_guardrail(params, {"guardrail_name": "a365-badmode"})
+
+
+def _guardrail_info(data: dict) -> dict:
+    entries: Final = data["metadata"]["standard_logging_guardrail_information"]
+    return entries[-1]
+
+
+class TestAllowFlow:
+    @pytest.mark.asyncio
+    async def test_allowed_call_passes_through(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "success"
+        assert info["guardrail_provider"] == "agent_365"
+        assert info["guardrail_response"]["verdict"] == "Allow"
+        assert info["guardrail_response"]["defender_status"] == "Evaluated"
+        assert info["guardrail_response"]["correlation_id"] == "corr-1"
+        assert info["guardrail_response"]["latency_ms"] >= 0
+
+    @pytest.mark.asyncio
+    async def test_obo_exchange_form(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        await _run(guardrail, _mcp_data())
+        token_call: Final = handler.calls[0]
+        assert token_call.url == TOKEN_URL
+        assert token_call.data["grant_type"] == "urn:ietf:params:oauth:grant-type:jwt-bearer"
+        assert token_call.data["requested_token_use"] == "on_behalf_of"
+        assert token_call.data["assertion"] == FAKE_ASSERTION
+        assert token_call.data["client_id"] == "client-xyz"
+        assert token_call.data["client_secret"] == "secret-123"
+        assert token_call.data["scope"] == f"{AGENT_365_PROD_RESOURCE_APP_ID}/ThreatProtection.Evaluate.All"
+
+    @pytest.mark.asyncio
+    async def test_evaluate_payload(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler, agent_id="agent-007")
+        await _run(guardrail, _mcp_data())
+        evaluate_call: Final = handler.calls[1]
+        assert evaluate_call.url == EVALUATE_URL
+        assert evaluate_call.headers["Authorization"] == "Bearer obo-access-token"
+        assert evaluate_call.json["tool"] == {"name": "send_email"}
+        assert evaluate_call.json["serverName"] == "outlook_mcp"
+        assert evaluate_call.json["arguments"] == {"to": "user@example.com", "body": "hello"}
+        assert evaluate_call.json["conversationId"] == "sess-123"
+        assert evaluate_call.json["agentId"] == "agent-007"
+
+    @pytest.mark.asyncio
+    async def test_agent_id_falls_back_to_key_alias(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        await _run(guardrail, _mcp_data())
+        assert handler.calls[1].json["agentId"] == "my-agent-key"
+
+    @pytest.mark.asyncio
+    async def test_non_mcp_call_type_skipped(self):
+        handler: Final = FakeHandler([])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data, call_type="completion")
+        assert result is data
+        assert handler.calls == []
+
+
+class TestConversationId:
+    @pytest.mark.asyncio
+    async def test_session_id_header_case_insensitive(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data(metadata={"headers": {"Mcp-Session-Id": "sess-CASED"}})
+        await _run(guardrail, data)
+        assert handler.calls[1].json["conversationId"] == "sess-CASED"
+
+    @pytest.mark.asyncio
+    async def test_falls_back_to_logging_obj_session_id(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        logging_obj: Final = SimpleNamespace(
+            model_call_details={"mcp_tool_call_metadata": {"mcp_session_id": "sess-from-logging"}},
+            litellm_call_id="call-id-1",
+        )
+        data: Final = _mcp_data(metadata={"headers": {}}, litellm_logging_obj=logging_obj)
+        await _run(guardrail, data)
+        assert handler.calls[1].json["conversationId"] == "sess-from-logging"
+
+    @pytest.mark.asyncio
+    async def test_falls_back_to_litellm_call_id(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data(metadata={"headers": {}}, litellm_call_id="call-id-2")
+        await _run(guardrail, data)
+        assert handler.calls[1].json["conversationId"] == "call-id-2"
+
+
+class TestBlockFlow:
+    @pytest.mark.asyncio
+    async def test_blocked_call_raises_400(self):
+        handler: Final = FakeHandler([_token_response(), _block_response(message="Injection detected")])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data()
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 400
+        assert exc_info.value.detail["error"] == "Blocked by Microsoft Defender"
+        assert exc_info.value.detail["message"] == "Injection detected"
+        assert exc_info.value.detail["tool"] == "send_email"
+        assert exc_info.value.detail["correlation_id"] == "corr-2"
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_intervened"
+        assert info["guardrail_response"]["verdict"] == "Block"
+
+    @pytest.mark.asyncio
+    async def test_blocked_even_with_fail_open(self):
+        handler: Final = FakeHandler([_token_response(), _block_response()])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_http_400_always_blocks_even_fail_open(self):
+        handler: Final = FakeHandler([_token_response(), _response(400, text="Bad request: serverName missing")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 400
+        assert "rejected" in exc_info.value.detail["error"]
+
+
+class TestUnreachableFallback:
+    @pytest.mark.asyncio
+    async def test_evaluate_litellm_timeout_fail_closed(self):
+        handler: Final = FakeHandler(
+            [
+                _token_response(),
+                LitellmTimeout(message="Connection timed out", model="default-model-name", llm_provider="httpx"),
+            ]
+        )
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_evaluate_timeout_fail_closed(self):
+        handler: Final = FakeHandler([_token_response(), httpx.ReadTimeout("timed out")])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+        assert "fail_closed" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_evaluate_timeout_fail_open(self):
+        handler: Final = FakeHandler([_token_response(), httpx.ReadTimeout("timed out")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_failed_to_respond"
+        assert info["guardrail_response"]["verdict"] == "Unscanned"
+
+    @pytest.mark.asyncio
+    async def test_evaluate_5xx_fail_closed(self):
+        handler: Final = FakeHandler([_token_response(), _response(502, text="bad gateway")])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+        assert "502" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_missing_bearer_token_fail_closed(self):
+        handler: Final = FakeHandler([])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data(incoming_bearer_token=None))
+        assert exc_info.value.status_code == 401
+        assert handler.calls == []
+
+    @pytest.mark.asyncio
+    async def test_non_jwt_bearer_token_fail_closed(self):
+        handler: Final = FakeHandler([])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data(incoming_bearer_token="sk-litellm-virtual-key"))
+        assert exc_info.value.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_missing_bearer_token_blocks_even_fail_open(self):
+        handler: Final = FakeHandler([])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data(incoming_bearer_token=None)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 401
+        assert handler.calls == []
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_intervened"
+        assert info["guardrail_response"]["verdict"] == "Rejected"
+
+    @pytest.mark.asyncio
+    async def test_obo_rejected_blocks_even_fail_open(self):
+        handler: Final = FakeHandler(
+            [_response(400, {"error": "invalid_grant", "error_description": "AADSTS50013: bad assertion"})]
+        )
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 401
+        assert "invalid_grant" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_evaluate_4xx_blocks_even_fail_open(self):
+        handler: Final = FakeHandler([_token_response(), _response(403, text="obo token lacks the scope")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 400
+        assert "403" in exc_info.value.detail["message"]
+        assert "lacks the scope" not in exc_info.value.detail["message"]
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_intervened"
+        assert info["guardrail_response"]["reason"] == "HTTP 403: obo token lacks the scope"
+
+    @pytest.mark.asyncio
+    async def test_obo_rejected_fail_closed(self):
+        handler: Final = FakeHandler(
+            [_response(400, {"error": "invalid_grant", "error_description": "AADSTS50013: bad assertion"})]
+        )
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 401
+        assert "invalid_grant" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_obo_endpoint_5xx_fail_open(self):
+        handler: Final = FakeHandler([_response(503, text="entra down")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_failed_to_respond"
+        assert info["guardrail_response"]["verdict"] == "Unscanned"
+
+
+class TestOboTokenCache:
+    @pytest.mark.asyncio
+    async def test_same_assertion_reuses_token(self):
+        handler: Final = FakeHandler([_token_response(), _allow_response(), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        await _run(guardrail, _mcp_data())
+        await _run(guardrail, _mcp_data())
+        token_calls: Final = [c for c in handler.calls if c.url == TOKEN_URL]
+        assert len(token_calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_different_assertions_get_distinct_tokens(self):
+        other_assertion: Final = "eyJhbGciOi.eyJvdGhlciI.b3RoZXJzaWc"
+        handler: Final = FakeHandler(
+            [
+                _token_response(access_token="token-a"),
+                _allow_response(),
+                _token_response(access_token="token-b"),
+                _allow_response(),
+            ]
+        )
+        guardrail: Final = _make_guardrail(handler)
+        await _run(guardrail, _mcp_data())
+        await _run(guardrail, _mcp_data(incoming_bearer_token=other_assertion))
+        token_calls: Final = [c for c in handler.calls if c.url == TOKEN_URL]
+        assert len(token_calls) == 2
+        assert handler.calls[3].headers["Authorization"] == "Bearer token-b"
+
+    @pytest.mark.asyncio
+    async def test_expired_token_refreshed(self):
+        handler: Final = FakeHandler(
+            [
+                _token_response(access_token="short-lived", expires_in=1),
+                _allow_response(),
+                _token_response(access_token="fresh"),
+                _allow_response(),
+            ]
+        )
+        guardrail: Final = _make_guardrail(handler)
+        await _run(guardrail, _mcp_data())
+        await _run(guardrail, _mcp_data())
+        token_calls: Final = [c for c in handler.calls if c.url == TOKEN_URL]
+        assert len(token_calls) == 2
+        assert handler.calls[3].headers["Authorization"] == "Bearer fresh"
+
+
+class TestEarlyPhasePassthrough:
+    @pytest.mark.asyncio
+    async def test_rest_body_shape_without_mcp_fields_skipped(self):
+        handler: Final = FakeHandler([])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = {
+            "server_id": "266024044f9612bf481c78f6cfef1ff0",
+            "name": "deepwiki-read_wiki_structure",
+            "arguments": {"repoName": "BerriAI/litellm"},
+            "metadata": {"headers": {"mcp-session-id": "sess-123"}},
+        }
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        assert handler.calls == []
+        assert "standard_logging_guardrail_information" not in data["metadata"]
+
+
+class TestRegistryDiscovery:
+    def test_auto_discovery_finds_agent_365(self):
+        from litellm.proxy.guardrails.guardrail_registry import (
+            get_guardrail_class_from_hooks,
+            get_guardrail_initializer_from_hooks,
+        )
+
+        assert "agent_365" in get_guardrail_initializer_from_hooks()
+        assert get_guardrail_class_from_hooks()["agent_365"] is Agent365Guardrail
+
+
+class TestMalformedResponses:
+    @pytest.mark.asyncio
+    async def test_obo_html_body_fail_open(self):
+        handler: Final = FakeHandler([_response(200, text="<html>blocked by egress proxy</html>")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        assert _guardrail_info(data)["guardrail_response"]["verdict"] == "Unscanned"
+
+    @pytest.mark.asyncio
+    async def test_obo_html_body_fail_closed(self):
+        handler: Final = FakeHandler([_response(200, text="<html>outage</html>")])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+        assert "non-JSON" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_obo_non_object_json_fail_closed(self):
+        handler: Final = FakeHandler([_response(200, ["not", "a", "dict"])])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_evaluate_html_body_fail_open(self):
+        handler: Final = FakeHandler([_token_response(), _response(200, text="<html>waf page</html>")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        assert _guardrail_info(data)["guardrail_response"]["verdict"] == "Unscanned"
+
+    @pytest.mark.asyncio
+    async def test_evaluate_html_body_fail_closed(self):
+        handler: Final = FakeHandler([_token_response(), _response(200, text="<html>waf page</html>")])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_evaluate_non_object_json_fail_closed(self):
+        handler: Final = FakeHandler([_token_response(), _response(200, "allowed")])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_bad_expires_in_still_allows(self):
+        handler: Final = FakeHandler([_response(200, {"access_token": "tok-1", "expires_in": "soon"}), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+
+    @pytest.mark.asyncio
+    async def test_obo_litellm_timeout_fail_open(self):
+        handler: Final = FakeHandler(
+            [LitellmTimeout(message="Connection timed out", model="default-model-name", llm_provider="httpx")]
+        )
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        assert _guardrail_info(data)["guardrail_status"] == "guardrail_failed_to_respond"
+
+
+class TestDeltaHardening:
+    @pytest.mark.asyncio
+    async def test_non_string_access_token_fail_closed(self):
+        handler: Final = FakeHandler([_response(200, {"access_token": None, "expires_in": 3599})])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+        assert "access_token" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_numeric_string_expires_in_honored(self):
+        handler: Final = FakeHandler([_response(200, {"access_token": "tok-9", "expires_in": "120"}), _allow_response()])
+        guardrail: Final = _make_guardrail(handler)
+        await _run(guardrail, _mcp_data())
+        entries: Final = list(guardrail._obo_token_cache.values())
+        assert len(entries) == 1
+        assert entries[0][1] - time.time() < 200
+
+    @pytest.mark.asyncio
+    async def test_evaluate_400_records_intervention(self):
+        handler: Final = FakeHandler([_token_response(), _response(400, text="bad request shape")])
+        guardrail: Final = _make_guardrail(handler)
+        data: Final = _mcp_data()
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 400
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_intervened"
+        assert info["guardrail_response"]["verdict"] == "Rejected"
+
+
+class TestVeriaHardening:
+    @pytest.mark.asyncio
+    async def test_evaluate_401_evicts_cached_obo_token(self):
+        handler: Final = FakeHandler(
+            [
+                _token_response(),
+                _response(401, text="token expired"),
+                _token_response(access_token="tok-2"),
+                _allow_response(),
+            ]
+        )
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException):
+            await _run(guardrail, _mcp_data())
+        result: Final = await _run(guardrail, _mcp_data())
+        assert result is not None
+        token_calls: Final = [c for c in handler.calls if c.url == TOKEN_URL]
+        assert len(token_calls) == 2
+
+    @pytest.mark.asyncio
+    async def test_evaluate_429_blocks_even_fail_open_as_throttled(self):
+        handler: Final = FakeHandler([_token_response(), _response(429, text="slow down")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 503
+        assert "429" in exc_info.value.detail["message"]
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_failed_to_respond"
+        assert info["guardrail_response"]["verdict"] == "Throttled"
+
+    @pytest.mark.asyncio
+    async def test_evaluate_500_is_unavailable(self):
+        handler: Final = FakeHandler([_token_response(), _response(500, text="oops")])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+        assert "500" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_token_endpoint_429_blocks_even_fail_open_as_throttled(self):
+        handler: Final = FakeHandler(
+            [_response(429, {"error": "temporarily_throttled", "error_description": "AADSTS90056"})]
+        )
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 503
+        assert "429" in exc_info.value.detail["message"]
+        info: Final = _guardrail_info(data)
+        assert info["guardrail_status"] == "guardrail_failed_to_respond"
+        assert info["guardrail_response"]["verdict"] == "Throttled"
+
+    @pytest.mark.asyncio
+    async def test_token_endpoint_408_non_json_blocks_as_throttled(self):
+        handler: Final = FakeHandler([_response(408, text="Request Timeout")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, data)
+        assert exc_info.value.status_code == 503
+        assert _guardrail_info(data)["guardrail_response"]["verdict"] == "Throttled"
+
+    @pytest.mark.asyncio
+    async def test_token_endpoint_4xx_html_stays_infra_fail_open(self):
+        handler: Final = FakeHandler([_response(403, text="<html>waf block page</html>")])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        assert _guardrail_info(data)["guardrail_response"]["verdict"] == "Unscanned"
+
+    @pytest.mark.asyncio
+    async def test_entra_200_missing_access_token_is_malformed(self):
+        handler: Final = FakeHandler([_response(200, {"token_type": "Bearer"})])
+        guardrail: Final = _make_guardrail(handler)
+        with pytest.raises(HTTPException) as exc_info:
+            await _run(guardrail, _mcp_data())
+        assert exc_info.value.status_code == 503
+        assert "access_token" in exc_info.value.detail["message"]
+
+    @pytest.mark.asyncio
+    async def test_evaluate_5xx_fail_open_allows_unscanned_once(self):
+        handler: Final = FakeHandler([_token_response(), _response(502, text='{"error": "bad gateway"}')])
+        guardrail: Final = _make_guardrail(handler, unreachable_fallback="fail_open")
+        data: Final = _mcp_data()
+        result: Final = await _run(guardrail, data)
+        assert result is data
+        records: Final = data["metadata"]["standard_logging_guardrail_information"]
+        assert len(records) == 1
+        assert records[0]["guardrail_response"]["verdict"] == "Unscanned"
+        assert records[0]["guardrail_status"] == "guardrail_failed_to_respond"

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_configs.ts

@@ -312,4 +312,11 @@ export const GUARDRAIL_PRESETS: Record<string, GuardrailPreset> = {
     mode: "pre_call",
     defaultOn: false,
   },
+  agent_365: {
+    provider: "Agent365",
+    guardrailNameSuggestion: "Microsoft Agent 365 Guardrail",
+    mode: "pre_mcp_call",
+    // MCP-only: default_on is the only activation path on the MCP hook
+    defaultOn: true,
+  },
 };

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.test.ts

@@ -27,6 +27,7 @@ const EXPECTED_PARTNER_LOGO_FILES: Record<string, string> = {
   deepkeep: "deepkeep.svg",
   repelloai: "repelloai.png",
   straiker: "straiker.svg",
+  agent_365: "microsoft_azure.svg",
 };
 
 describe("guardrail_garden_data logos", () => {

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.ts

@@ -464,6 +464,16 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
     tags: ["Agentic", "Prompt Injection", "Tool Misuse", "MCP", "Skills"],
     providerKey: "Straiker",
   },
+  {
+    id: "agent_365",
+    name: "Microsoft Agent 365",
+    description:
+      "Microsoft Agent 365 tool-call governance: Defender threat evaluation and observability for MCP tool calls, acting on behalf of the signed-in user",
+    category: "partner",
+    logo: guardrailLogoMap["Microsoft Agent 365"],
+    tags: ["Agentic", "MCP", "Tool Misuse", "Observability"],
+    providerKey: "Agent365",
+  },
 ];
 
 export const ALL_CARDS = [...LITELLM_CONTENT_FILTER_CARDS, ...PARTNER_GUARDRAIL_CARDS];

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_info_helpers.tsx

@@ -204,6 +204,7 @@ export const guardrailLogoMap = {
   "Qostodian Nexus": qohashLogo.src,
   "RepelloAI Argus": repelloAiLogo.src,
   Straiker: straikerLogo.src,
+  "Microsoft Agent 365": microsoftAzureLogo.src,
 } satisfies Record<string, string>;
 
 export const getGuardrailLogo = (displayName: string): string | undefined =>