Merge fdb7da7c61 into 559247fa84

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -5,13 +5,14 @@ import secrets
 import time
 from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any, Final, Literal, Optional
+from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 import httpx
 from fastapi import APIRouter, Depends, Form, HTTPException, Request
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
 from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError
+from typing_extensions import ReadOnly
 
 from litellm._logging import verbose_logger
 from litellm.caching.in_memory_cache import InMemoryCache
@@ -81,6 +82,10 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import (
     encrypt_value_helper,
 )
 from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
+from litellm.proxy.guardrails.guardrail_hooks.agent_365.agent_365 import (
+    agent_365_authorization_servers,
+    agent_365_scopes_supported,
+)
 from litellm.types.mcp import MCPAuth, MCPCredentials
 from litellm.types.mcp_server.mcp_server_manager import MCPServer, MCPTokenEndpointAuthMethod
 
@@ -2398,7 +2403,7 @@ async def _build_oauth_protected_resource_response(
     request: Request,
     mcp_server_name: str | None,
     use_standard_pattern: bool,
-) -> dict:
+) -> Mapping[str, object]:
     """
     Build OAuth protected resource response with the appropriate URL pattern.
 
@@ -2497,6 +2502,15 @@ async def _build_oauth_protected_resource_response(
     if obo_response is not None:
         return obo_response
 
+    agent_365_issuers: Final = agent_365_authorization_servers(mcp_server, None) if mcp_server else ()
+    if mcp_server is not None and agent_365_issuers:
+        agent_365_metadata: Final[_ProtectedResourceMetadata] = {
+            "authorization_servers": agent_365_issuers,
+            "resource": resource_url,
+            "scopes_supported": agent_365_scopes_supported(mcp_server, None),
+        }
+        return agent_365_metadata
+
     if mcp_server is not None and mcp_server.advertises_gateway_authorization_server:
         return {
             "authorization_servers": [f"{request_base_url}/mcp"],
@@ -2516,6 +2530,12 @@ async def _build_oauth_protected_resource_response(
     }
 
 
+class _ProtectedResourceMetadata(TypedDict):
+    authorization_servers: ReadOnly[tuple[str, ...]]
+    resource: ReadOnly[str]
+    scopes_supported: ReadOnly[tuple[str, ...]]
+
+
 def _obo_protected_resource_response(mcp_server: MCPServer | None, resource_url: str) -> dict | None:
     """The OBO (token_exchange) PRM, or None when this server is not OBO / no issuer is configured.
 

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -242,6 +242,9 @@ _user_env_vars_cache: Final[dict[tuple[str, str], tuple[dict[str, str], float]]]
 _USER_ENV_VARS_CACHE_TTL: Final = 60  # seconds
 _USER_ENV_VARS_CACHE_MAX_SIZE: Final = 4096  # cap to prevent unbounded growth
 
+_NO_LISTED_TOOLS: Final[Mapping[str | None, Mapping[str, MCPTool]]] = MappingProxyType({})
+_LISTED_TOOLS_CALLERS_PER_SERVER: Final = 256
+
 # Auth types whose upstream OAuth endpoints (protected-resource + authorization-server metadata) the
 # gateway discovers from the upstream itself: interactive oauth2 and the two client-forwarded modes.
 # OBO/M2M endpoint discovery is decided separately via _obo_needs_endpoint_discovery. Shared by the
@@ -1947,6 +1950,9 @@ class MCPServerManager:
             "gmail_send_email": "zapier_mcp_server",
         }
         """
+        self._listed_tools_by_server_id: dict[
+            str, Mapping[str | None, Mapping[str, MCPTool]]
+        ] = {}  # mutable-ok: refreshed per tools/list
         self._upstream_initialize_instructions_by_server_id: dict[str, str] = {}
         # Per-server monotonic timestamp of last upstream prefetch attempt (success,
         # empty result, or failure). Used to throttle re-probes for servers that do
@@ -2642,7 +2648,7 @@ class MCPServerManager:
             self._assign_unique_short_prefix(new_server)
             _warn_internal_delegate_pkce_if_applicable(new_server, source="config")
             _warn_config_id_jag_server_outruns_sso(new_server)
-            self._invalidate_discovery_lists(server_id)
+            self._invalidate_server_definition_caches(server_id)
             self.config_mcp_servers[server_id] = new_server
             self._set_oauth_discovery_deferred(
                 server_id,
@@ -2844,7 +2850,7 @@ class MCPServerManager:
             global_mcp_tool_registry,
         )
 
-        self._invalidate_discovery_lists(server.server_id)
+        self._invalidate_server_definition_caches(server.server_id)
         prefix_root: Final = normalize_server_name(get_server_prefix(server))
         if server.spec_path and prefix_root:
             openapi_key_prefix: Final = prefix_root + MCP_TOOL_PREFIX_SEPARATOR
@@ -3221,7 +3227,7 @@ class MCPServerManager:
                 # env_vars_are_encrypted=False.
                 new_server: Final = await self.build_mcp_server_from_table(mcp_server, env_vars_are_encrypted=False)
                 self._assign_unique_short_prefix(new_server)
-                self._invalidate_discovery_lists(mcp_server.server_id)
+                self._invalidate_server_definition_caches(mcp_server.server_id)
                 self.registry[mcp_server.server_id] = new_server
                 await self._maybe_register_openapi_tools(new_server)
                 self.prime_oauth_metadata_discovery(new_server)
@@ -3258,7 +3264,7 @@ class MCPServerManager:
                     previous_server=self.registry[mcp_server.server_id],
                 )
                 self._assign_unique_short_prefix(new_server)
-                self._invalidate_discovery_lists(mcp_server.server_id)
+                self._invalidate_server_definition_caches(mcp_server.server_id)
                 self.registry[mcp_server.server_id] = new_server
                 await self._maybe_register_openapi_tools(new_server)
                 self.prime_oauth_metadata_discovery(new_server)
@@ -4092,6 +4098,7 @@ class MCPServerManager:
         oauth2_headers: dict[str, str] | None,
         user_api_key_auth: UserAPIKeyAuth | None,
         raw_headers: Mapping[str, str] | None = None,
+        resource_metadata_url: str | None = None,
     ) -> None:
         """Mint an exchange-backed server's upstream credential at the transport edge.
 
@@ -4129,7 +4136,9 @@ class MCPServerManager:
         if spec is None or not isinstance(spec.config, (TokenExchangeConfig, IdJagConfig)):
             return
         if subject_token is None and isinstance(spec.config, TokenExchangeConfig):
-            raise_token_exchange_challenge(resolved_server, root_path=get_request_root_path())
+            raise_token_exchange_challenge(
+                resolved_server, root_path=get_request_root_path(), resource_metadata_url=resource_metadata_url
+            )
         match await self._cred_provider.resolve_credentials(to_subject(user_api_key_auth, subject_token), spec):
             case Ok(_):
                 return
@@ -4139,6 +4148,7 @@ class MCPServerManager:
                         resolved_server,
                         root_path=get_request_root_path(),
                         claims=err.unauthorized.claims,
+                        resource_metadata_url=resource_metadata_url,
                     )
                 raise_public(err)
 
@@ -4431,29 +4441,25 @@ class MCPServerManager:
             if server.spec_path:
                 # OpenAPI tools were stored in the registry under the prefix
                 # active at registration time — fetch by that same prefix.
-                _tools: Final = global_mcp_tool_registry.list_tools(tool_prefix=get_server_prefix(server))
+                registry_prefix: Final = normalize_server_name(get_server_prefix(server)) + MCP_TOOL_PREFIX_SEPARATOR
+                _tools: Final = global_mcp_tool_registry.list_tools(tool_prefix=registry_prefix)
                 tools = global_mcp_tool_registry.convert_tools_to_mcp_sdk_tool_type(_tools)
                 # OpenAPI tools are stored in the registry with their prefix already
                 # applied (e.g. "test_petstore-getinventory").  Do NOT pass them
                 # through _create_prefixed_tools — that would add the prefix a second
                 # time producing "test_petstore-test_petstore-getinventory".
-                if not add_prefix:
-                    prefix: Final = get_server_prefix(server)
-                    sep: Final = MCP_TOOL_PREFIX_SEPARATOR
-                    tools = [
-                        (
-                            t.model_copy(update={"name": t.name[len(prefix) + len(sep) :]})
-                            if t.name.startswith(f"{prefix}{sep}")
-                            else t
-                        )
-                        for t in tools
-                    ]
-                return tools
+                unprefixed_tools: Final = [  # mutable-ok: returned through the list[MCPTool] listing contract
+                    t.model_copy(update=MappingProxyType({"name": t.name[len(registry_prefix) :]})) for t in tools
+                ]
+                self._record_listed_tools(server, unprefixed_tools, user_api_key_auth)
+                return tools if add_prefix else unprefixed_tools
             else:
                 tools = await self._fetch_tools_with_timeout(client, server.name)
                 self._remember_upstream_initialize_instructions(server, client)
 
-            prefixed_or_original_tools: Final = self._create_prefixed_tools(tools, server, add_prefix=add_prefix)
+            prefixed_or_original_tools: Final = self._create_prefixed_tools(
+                tools, server, add_prefix=add_prefix, user_api_key_auth=user_api_key_auth
+            )
 
             return prefixed_or_original_tools
 
@@ -4497,6 +4503,40 @@ class MCPServerManager:
         self._resource_discovery_cache.invalidate(server_id)
         self._template_discovery_cache.invalidate(server_id)
 
+    def _invalidate_server_definition_caches(self, server_id: str) -> None:
+        self._invalidate_discovery_lists(server_id)
+        self._listed_tools_by_server_id.pop(server_id, None)
+
+    def _discovers_per_caller(self, server: MCPServer) -> bool:
+        return (
+            server.requires_per_user_auth
+            or self._references_per_user_env_var(server)
+            or server.delegate_auth_to_upstream
+            or server.auth_type in (MCPAuth.oauth2_token_exchange, MCPAuth.oauth2_id_jag)
+        )
+
+    def _listed_tools_identity(self, server: MCPServer, user_api_key_auth: UserAPIKeyAuth | None) -> str | None:
+        if server.spec_path or user_api_key_auth is None or not self._discovers_per_caller(server):
+            return None
+        material: Final = json.dumps((user_api_key_auth.user_id, user_api_key_auth.api_key), separators=(",", ":"))
+        return hashlib.sha256(material.encode()).hexdigest()
+
+    def _record_listed_tools(
+        self, server: MCPServer, tools: Sequence[MCPTool], user_api_key_auth: UserAPIKeyAuth | None
+    ) -> None:
+        identity: Final = self._listed_tools_identity(server, user_api_key_auth)
+        listing: Final = MappingProxyType({tool.name: tool for tool in tools})
+        existing: Final = self._listed_tools_by_server_id.get(server.server_id, _NO_LISTED_TOOLS)
+        shared: Final = existing.get(None)
+        callers: Final = tuple((key, value) for key, value in existing.items() if key not in (None, identity))
+        evicted: Final = 0 if identity is None else max(len(callers) + 1 - _LISTED_TOOLS_CALLERS_PER_SERVER, 0)
+        entries: Final = (
+            *(() if shared is None else ((None, shared),)),
+            *callers[evicted:],
+            (identity, listing),
+        )
+        self._listed_tools_by_server_id[server.server_id] = MappingProxyType(dict(entries))
+
     def _discovery_key(
         self,
         server: MCPServer,
@@ -4507,12 +4547,7 @@ class MCPServerManager:
         subject_token: str | None,
         credential_fingerprint: str | None = None,
     ) -> _DiscoveryKey:
-        per_user: Final = (
-            server.requires_per_user_auth
-            or self._references_per_user_env_var(server)
-            or server.delegate_auth_to_upstream
-            or server.auth_type in (MCPAuth.oauth2_token_exchange, MCPAuth.oauth2_id_jag)
-        )
+        per_user: Final = self._discovers_per_caller(server)
         if not (per_user or mcp_auth_header or extra_headers or stdio_env or subject_token):
             return server.server_id, None
         identity: Final = (
@@ -5325,7 +5360,13 @@ class MCPServerManager:
             "attempts; the 3-character prefix space is too crowded."
         )
 
-    def _create_prefixed_tools(self, tools: list[MCPTool], server: MCPServer, add_prefix: bool = True) -> list[MCPTool]:
+    def _create_prefixed_tools(
+        self,
+        tools: list[MCPTool],
+        server: MCPServer,
+        add_prefix: bool = True,
+        user_api_key_auth: UserAPIKeyAuth | None = None,
+    ) -> list[MCPTool]:
         """
         Create prefixed tools and update tool mapping.
 
@@ -5357,9 +5398,19 @@ class MCPServerManager:
             for spelling in iter_known_tool_name_spellings(original_name, server):
                 self.tool_name_to_mcp_server_name_mapping[spelling] = prefix
 
+        self._record_listed_tools(server, tools, user_api_key_auth)
         verbose_logger.info("Successfully fetched %s tools from server %s", len(prefixed_tools), server.name)
         return prefixed_tools
 
+    def get_listed_tool(
+        self, server: MCPServer, name: str, user_api_key_auth: UserAPIKeyAuth | None = None
+    ) -> MCPTool | None:
+        identity: Final = self._listed_tools_identity(server, user_api_key_auth)
+        listed: Final = self._listed_tools_by_server_id.get(server.server_id, _NO_LISTED_TOOLS).get(identity)
+        if not listed:
+            return None
+        return listed.get(name) or listed.get(strip_known_server_prefix(name, server))
+
     def _create_prefixed_prompts(
         self, prompts: Sequence[Prompt], server: MCPServer, add_prefix: bool = True
     ) -> list[Prompt]:
@@ -5598,6 +5649,7 @@ class MCPServerManager:
         server: MCPServer,
         raw_headers: dict[str, str] | None = None,
         litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
+        tool: MCPTool | None = None,
     ) -> dict[str, Any]:
         """
         Run pre-call checks and guardrail hooks for an MCP tool call.
@@ -5611,6 +5663,9 @@ class MCPServerManager:
         ``pre_mcp_call`` evaluation (or a block) on the spend-log row the Guardrails
         Monitor counts. It stays optional so callers that do no logging are unchanged.
 
+        ``tool`` is the upstream tool definition when one was listed, so guardrails
+        can see its description and input schema, not just the name and arguments.
+
         Returns a dict that may contain:
         - "arguments": hook-modified tool arguments (only if changed)
         - "extra_headers": headers injected by pre_mcp_call guardrail hooks
@@ -5664,6 +5719,8 @@ class MCPServerManager:
             "user_api_key_hash": (getattr(user_api_key_auth, "api_key_hash", None) if user_api_key_auth else None),
             "incoming_bearer_token": incoming_bearer_token,
             "headers": logging_safe_mcp_headers(raw_headers),
+            "tool_description": tool.description if tool is not None else None,
+            "tool_input_schema": tool.inputSchema if tool is not None else None,
         }
 
         # Create MCP request object for processing
@@ -5718,6 +5775,7 @@ class MCPServerManager:
         proxy_logging_obj: ProxyLogging,
         start_time: datetime.datetime,
         litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
+        tool: MCPTool | None = None,
     ):
         """Create and return a during hook task for MCP tool calls.
 
@@ -5732,6 +5790,8 @@ class MCPServerManager:
             tool_name=name,
             arguments=arguments,
             server_name=server_name_from_prefix,
+            tool_description=tool.description if tool is not None else None,
+            tool_input_schema=tool.inputSchema if tool is not None else None,
             start_time=start_time.timestamp() if start_time else None,
             hidden_params=HiddenParams(),
         )
@@ -6328,6 +6388,7 @@ class MCPServerManager:
             server=mcp_server,
             raw_headers=raw_headers,
             litellm_logging_obj=litellm_logging_obj,
+            tool=self.get_listed_tool(mcp_server, name, user_api_key_auth),
         )
         if "arguments" in hook_result:
             arguments = hook_result["arguments"]
@@ -6343,6 +6404,7 @@ class MCPServerManager:
                 proxy_logging_obj=proxy_logging_obj,
                 start_time=start_time,
                 litellm_logging_obj=litellm_logging_obj,
+                tool=self.get_listed_tool(mcp_server, name, user_api_key_auth),
             )
             tasks.append(during_hook_task)
 
@@ -6610,7 +6672,7 @@ class MCPServerManager:
 
         for server_id in previous_registry.keys() | registered_registry.keys():
             if previous_registry.get(server_id) != registered_registry.get(server_id):
-                self._invalidate_discovery_lists(server_id)
+                self._invalidate_server_definition_caches(server_id)
         self.registry = registered_registry
         # A discovery task may have published into ``previous_registry`` while
         # this replacement was being staged. Reconcile every published entry

litellm/proxy/_experimental/mcp_server/oauth_utils.py

@@ -199,7 +199,7 @@ def get_route_relative_request_path(scope: Scope) -> str:
     :func:`litellm.proxy.auth.auth_utils.get_request_route`, which the rest of the MCP auth path
     already routes through, so ``/litellmfoo`` is not truncated under ``root_path=/litellm``."""
     raw_path = str(scope.get("_original_path") or scope.get("path", "") or "")
-    root_path = str(scope.get("app_root_path") or scope.get("root_path") or "").rstrip("/")
+    root_path = str(scope.get("app_root_path", scope.get("root_path")) or "").rstrip("/")
     if root_path and (raw_path == root_path or raw_path.startswith(f"{root_path}/")):
         return raw_path[len(root_path) :]
     return raw_path

litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py

@@ -358,6 +358,7 @@ def raise_token_exchange_challenge(
     *,
     root_path: str,
     claims: str | None = None,
+    resource_metadata_url: str | None = None,
 ) -> NoReturn:
     """Raise the RFC 9728 / RFC 6750 challenge an OBO (``token_exchange``) server returns when the
     caller's subject token is missing or the IdP rejected it.
@@ -375,8 +376,13 @@ def raise_token_exchange_challenge(
     ``error="invalid_token"`` and is byte-identical to the static one. Both the error value (one of
     two literals) and the base64 claims draw from a fixed alphabet, so nothing from the IdP body
     reaches the header unescaped.
+
+    ``resource_metadata_url`` overrides the derived path with the absolute metadata URL matching the
+    route spelling the request arrived on: RFC 9728 §3.3 clients reject a ``resource`` that differs
+    from the URL they connected to, and a ``/{server}/mcp`` connect must not be sent to the
+    ``/mcp/{server}`` document.
     """
-    resource_metadata: Final = oauth_protected_resource_path(root_path, server)
+    resource_metadata: Final = resource_metadata_url or oauth_protected_resource_path(root_path, server)
     encoded_claims: Final = base64.b64encode(claims.encode()).decode() if claims else None
     error: Final = "insufficient_claims" if encoded_claims else "invalid_token"
     error_description: Final = (

litellm/proxy/_experimental/mcp_server/server.py

@@ -57,6 +57,7 @@ from litellm.proxy._experimental.mcp_server.mcp_debug import (
 from litellm.proxy._experimental.mcp_server.oauth_utils import (
     _redact_mcp_resource_url,
     get_byok_www_authenticate,
+    get_passthrough_resource_metadata_url,
     get_passthrough_www_authenticate,
     get_route_relative_request_path,
     well_known_root_suffix,
@@ -80,12 +81,17 @@ from litellm.proxy._types import (
     UserAPIKeyAuth,
 )
 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
+from litellm.proxy.guardrails.guardrail_hooks.agent_365.agent_365 import (
+    agent_365_authorization_servers,
+    agent_365_subject_token_present,
+)
 from litellm.proxy.litellm_pre_call_utils import (
     LiteLLMProxyRequestSetup,
     get_chain_id_from_headers,
 )
 from litellm.types.mcp import MCPAuth, MCPSpecVersion
 from litellm.types.mcp_server.mcp_server_manager import MCPInfo, MCPServer
+from litellm.types.mcp_server.tool_registry import MCPTool as RegisteredTool
 from litellm.types.utils import CallTypes, StandardLoggingMCPToolCall
 from litellm.utils import Rules, client, function_setup
 
@@ -2777,6 +2783,9 @@ if MCP_AVAILABLE:
 
         return managed_resource_templates
 
+    def _registered_tool_metadata(name: str, registered: RegisteredTool) -> MCPTool:
+        return MCPTool(name=name, description=registered.description, inputSchema=registered.input_schema)
+
     def _resolve_display_name_to_original(
         name: str,
         allowed_mcp_servers: list[MCPServer],
@@ -3115,6 +3124,7 @@ if MCP_AVAILABLE:
                 server=mcp_server,
                 raw_headers=raw_headers,
                 litellm_logging_obj=litellm_logging_obj,
+                tool=_registered_tool_metadata(original_tool_name, local_tool),
             )
             # `pre_call_tool_check` may return guardrail-modified
             # arguments; honor them on the local path too.
@@ -3180,7 +3190,8 @@ if MCP_AVAILABLE:
             # not in the registry either, `_handle_local_mcp_tool` below reports
             # 404 and nothing runs, so demanding a server here would turn every
             # unknown tool name into a misleading 503.
-            if global_mcp_tool_registry.get_tool(original_tool_name) is not None:
+            registered_local_tool: Final = global_mcp_tool_registry.get_tool(original_tool_name)
+            if registered_local_tool is not None:
                 # `mcp_server` is None here because the tool name is not in the
                 # tool -> server mapping, but the name still carries a prefix
                 # that the server-level check above compared against the
@@ -3221,6 +3232,7 @@ if MCP_AVAILABLE:
                     server=prefix_server,
                     raw_headers=raw_headers,
                     litellm_logging_obj=litellm_logging_obj,
+                    tool=_registered_tool_metadata(original_tool_name, registered_local_tool),
                 )
                 if "arguments" in hook_result:
                     arguments = hook_result["arguments"]  # pyright: ignore[reportAny]  # hook returns untyped args
@@ -4129,8 +4141,19 @@ if MCP_AVAILABLE:
             # (transport level, where WWW-Authenticate survives) with the RFC 9728 resource_metadata
             # so the client discovers the IdP, SSOs, and retries with a subject token, which LiteLLM
             # then exchanges. A tool-call-time 401 would be wrapped into a JSON-RPC error and the
-            # header lost, so the discovery flow needs this pre-emptive challenge.
-            if server and server.auth_type == MCPAuth.oauth2_token_exchange and not oauth2_headers:
+            # header lost, so the discovery flow needs this pre-emptive challenge. Servers gated by an
+            # Agent 365 guardrail (OBO to the evaluate API) get the same challenge, also when the only
+            # bearer is the LiteLLM key itself, which admits the caller but is not an exchangeable subject.
+            # Only on the server's own route: the per-server metadata ``resource`` must equal the URL the
+            # client connected to (RFC 9728 3.3), which aggregate ``/mcp`` and multi-server connects never do.
+            if server and (
+                (server.auth_type == MCPAuth.oauth2_token_exchange and not oauth2_headers)
+                or (
+                    tuple(_get_mcp_servers_in_path(get_route_relative_request_path(scope)) or ()) == (server_name,)
+                    and not agent_365_subject_token_present(oauth2_headers)
+                    and agent_365_authorization_servers(server, user_api_key_auth)
+                )
+            ):
                 from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import (  # noqa: PLC0415  # lazy: adapter pulls MCP subgraph
                     raise_token_exchange_challenge,
                 )
@@ -4138,7 +4161,11 @@ if MCP_AVAILABLE:
                     get_request_root_path,
                 )
 
-                raise_token_exchange_challenge(server, root_path=get_request_root_path())
+                raise_token_exchange_challenge(
+                    server,
+                    root_path=get_request_root_path(),
+                    resource_metadata_url=get_passthrough_resource_metadata_url(scope=scope, server_name=server_name),
+                )
 
             # Exchange-backed modes (token_exchange's OBO mint, id_jag's stored-assertion mint): run
             # the exchange here at the transport edge, so a rejected subject raises the RFC 9728
@@ -4163,6 +4190,7 @@ if MCP_AVAILABLE:
                     oauth2_headers=oauth2_headers,
                     user_api_key_auth=user_api_key_auth,
                     raw_headers=raw_headers,
+                    resource_metadata_url=get_passthrough_resource_metadata_url(scope=scope, server_name=server_name),
                 )
 
             # Pass-through OAuth: when the admin has opted a server into

litellm/proxy/_lazy_openapi_snapshot.json

@@ -9932,7 +9932,7 @@
             },
             "unreachable_fallback": {
               "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'. 'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed.",
+              "description": "Behavior when a guardrail endpoint is unreachable due to network errors. 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.",
               "enum": [
                 "fail_closed",
                 "fail_open"
@@ -10894,6 +10894,18 @@
               "description": "Custom advisory message template used when on_flagged='inject_system_message'. Must contain a {reason} placeholder. Defaults to a generic advisory message if unset.",
               "title": "Advisory System Message"
             },
+            "agent_id": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Agent identity reported to Agent 365 with every tool evaluation. When unset, the caller's key alias is used.",
+              "title": "Agent Id"
+            },
             "akto_account_id": {
               "anyOf": [
                 {
@@ -11396,6 +11408,30 @@
               "title": "Chunk Budget Chars",
               "type": "integer"
             },
+            "client_id": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Client id of the gateway's Entra app registration (a confidential client). Falls back to the AGENT365_CLIENT_ID environment variable.",
+              "title": "Client Id"
+            },
+            "client_secret": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "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.",
+              "title": "Client Secret"
+            },
             "confidence_threshold": {
               "default": 0.5,
               "default_value": 0.5,
@@ -12436,6 +12472,18 @@
               "description": "The message the bot speaks aloud when a /v1/realtime guardrail fires. Falls back to violation_message_template if not set.",
               "title": "Realtime Violation Message"
             },
+            "resource_app_id": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Application id of the Agent 365 resource the OBO token is minted for. Defaults to the production resource ea9ffc3e-8a23-4a7d-836d-234d7c7565c1; the Test and PreProd environments use a different id. Falls back to the AGENT365_RESOURCE_APP_ID environment variable.",
+              "title": "Resource App Id"
+            },
             "rules": {
               "anyOf": [
                 {
@@ -12673,6 +12721,18 @@
               "description": "The ID of your Model Armor template",
               "title": "Template Id"
             },
+            "tenant_id": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Entra tenant id used for the On-Behalf-Of token exchange. Falls back to the AGENT365_TENANT_ID environment variable.",
+              "title": "Tenant Id"
+            },
             "timeout": {
               "anyOf": [
                 {

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

@@ -0,0 +1,63 @@
+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 is required. Set client_secret, api_key, or AGENT365_CLIENT_SECRET"
+        )
+
+    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,695 @@
+"""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 pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError
+from typing_extensions import ReadOnly, TypedDict
+
+import litellm
+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.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+from litellm.proxy.litellm_pre_call_utils import add_guardrails_from_auth_metadata
+from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.mcp import MCPAuth
+from litellm.types.mcp_server.mcp_server_manager import MCPServer
+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"
+ENTRA_ISSUER_TEMPLATE: Final = "https://login.microsoftonline.com/{tenant_id}/v2.0"
+EVALUATE_PATH: Final = "/agents/tool-evaluation/evaluate"
+MCP_SESSION_ID_HEADER: Final = "mcp-session-id"
+DEFENDER_STATUS_EVALUATED: Final = "Evaluated"
+GATEWAY_SCOPE_TEMPLATE: Final = "api://{client_id}/access_as_user"
+_GATEWAY_OWNED_TOKEN_ERRORS: Final = frozenset(
+    {"invalid_client", "unauthorized_client", "invalid_scope", "invalid_resource"}
+)
+_MCP_CALL_TYPES: Final[tuple[str, ...]] = ("mcp_call", "call_mcp_tool")
+_TOOL_INPUT_SCHEMA_ADAPTER: Final = TypeAdapter(dict[str, object])
+_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
+
+
+def _parse_tool_input_schema(raw: object) -> dict[str, object] | None:
+    try:
+        return _TOOL_INPUT_SCHEMA_ADAPTER.validate_python(raw)
+    except ValidationError:
+        return None
+
+
+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(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    name: str
+    description: str | None = None
+    input_schema: dict[str, object] | None = Field(default=None, serialization_alias="inputSchema")
+
+
+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 = entra_assertion(data.get("incoming_bearer_token"))
+        if assertion is None:
+            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:
+            if exc.error_code in _GATEWAY_OWNED_TOKEN_ERRORS:
+                return self._handle_unavailable(
+                    data=data,
+                    tool_name=tool_name,
+                    reason=(
+                        f"Entra rejected the gateway's own Agent 365 credentials ({exc.error_code}); "
+                        "check the guardrail's client_id, client_secret and resource_app_id"
+                    ),
+                )
+            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
+        allowed: Final = verdict.get("allowed")
+        if not isinstance(allowed, bool):
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason="the Agent 365 endpoint returned a verdict without a boolean 'allowed' field",
+            )
+        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
+        defender_status: Final = defender.get("status")
+        if allowed and defender_status != DEFENDER_STATUS_EVALUATED:
+            return self._handle_unavailable(
+                data=data,
+                tool_name=tool_name,
+                reason=f"Microsoft Defender did not evaluate the call (defender.status={defender_status or 'missing'})",
+                defender_status=defender_status,
+                correlation_id=correlation_id,
+                latency_ms=latency_ms,
+            )
+        self._record_verdict(
+            data=data,
+            verdict="Allow" if allowed else "Block",
+            guardrail_status="success" if allowed else "guardrail_intervened",
+            defender_status=defender_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 user_api_key_dict.key_alias
+        description: Final = data.get("mcp_tool_description")
+        tool_reference: Final = _ToolReference(
+            name=tool_name,
+            description=description if isinstance(description, str) and description else None,
+            input_schema=_parse_tool_input_schema(data.get("mcp_tool_input_schema")),
+        )
+        payload: Final[dict[str, object]] = {  # mutable-ok: JSON body with optional fields added below
+            "tool": tool_reference.model_dump(by_alias=True, exclude_none=True),
+            "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:
+        raw_logging_obj: Final = data.get("litellm_logging_obj")
+        logging_obj: Final = raw_logging_obj if isinstance(raw_logging_obj, LiteLLMLoggingObj) else None
+        call_id: Final = data.get("litellm_call_id") or (logging_obj.litellm_call_id if logging_obj else None)
+        if isinstance(call_id, str) and call_id:
+            return call_id
+        if logging_obj is not None:
+            tool_call_metadata: Final = logging_obj.model_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
+        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
+        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,
+        defender_status: str | None = None,
+        correlation_id: str | None = None,
+        latency_ms: float | None = None,
+    ) -> 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=defender_status,
+                correlation_id=correlation_id,
+                latency_ms=latency_ms,
+                reason=reason,
+            )
+            return data
+        self._record_verdict(
+            data=data,
+            verdict="Unavailable",
+            guardrail_status="guardrail_failed_to_respond",
+            defender_status=defender_status,
+            correlation_id=correlation_id,
+            latency_ms=latency_ms,
+            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,
+        )
+
+
+def _applies_to_caller(guardrail: Agent365Guardrail, user_api_key_auth: "UserAPIKeyAuth") -> bool:
+    probe: Final[dict[str, object]] = {"metadata": {}}  # mutable-ok: filled in place by the key resolver
+    add_guardrails_from_auth_metadata(
+        user_api_key_dict=user_api_key_auth, data=probe, metadata_variable_name="metadata"
+    )
+    return guardrail.should_run_guardrail(data=probe, event_type=GuardrailEventHooks.pre_mcp_call)
+
+
+def _applicable_guardrails(
+    server: MCPServer, user_api_key_auth: "UserAPIKeyAuth | None"
+) -> tuple[Agent365Guardrail, ...]:
+    """Agent 365 guardrails that gate ``server`` for this caller: every registered one for the anonymous
+    discovery fetch, otherwise those the caller's key, team, or policies select. Empty when the gateway
+    does not own sign-in for the server."""
+    if server.auth_type == MCPAuth.oauth2 or not server.advertises_gateway_authorization_server:
+        return ()
+    registered: Final = tuple(
+        callback
+        for callback in litellm.logging_callback_manager.get_custom_loggers_for_type(Agent365Guardrail)
+        if isinstance(callback, Agent365Guardrail)
+    )
+    if user_api_key_auth is None:
+        return registered
+    return tuple(g for g in registered if _applies_to_caller(g, user_api_key_auth))
+
+
+def entra_assertion(value: object) -> str | None:
+    """``value`` when it is a compact JWS, the only bearer shape the OBO exchange accepts as its assertion.
+    A LiteLLM virtual key, session bearer, or opaque upstream token in ``Authorization`` yields ``None``."""
+    return value if isinstance(value, str) and value.count(".") == 2 else None
+
+
+def agent_365_subject_token_present(oauth2_headers: Mapping[str, str] | None) -> bool:
+    """Whether the request's ``Authorization`` carries an Entra assertion the guardrail can exchange."""
+    authorization: Final = oauth2_headers.get("Authorization", "") if oauth2_headers else ""
+    if not authorization.lower().startswith("bearer "):
+        return False
+    return entra_assertion(authorization[len("bearer ") :].strip()) is not None
+
+
+def agent_365_authorization_servers(server: MCPServer, user_api_key_auth: "UserAPIKeyAuth | None") -> tuple[str, ...]:
+    """Entra issuers an MCP client signs in with before calling ``server`` through an Agent 365 guardrail."""
+    return tuple(
+        dict.fromkeys(
+            ENTRA_ISSUER_TEMPLATE.format(tenant_id=g.tenant_id)
+            for g in _applicable_guardrails(server, user_api_key_auth)
+        )
+    )
+
+
+def agent_365_scopes_supported(server: MCPServer, user_api_key_auth: "UserAPIKeyAuth | None") -> tuple[str, ...]:
+    """Scopes the client requests from Entra for ``server``: the admin's ``scopes`` when set, otherwise the
+    ``access_as_user`` scope of each gating guardrail's gateway app registration (``api://<client_id>``)."""
+    if server.scopes:
+        return tuple(server.scopes)
+    return tuple(
+        dict.fromkeys(
+            GATEWAY_SCOPE_TEMPLATE.format(client_id=g.client_id)
+            for g in _applicable_guardrails(server, user_api_key_auth)
+        )
+    )

litellm/proxy/utils.py

@@ -1231,6 +1231,8 @@ class ProxyLogging:
             "user_api_key_request_route": kwargs.get("user_api_key_request_route"),
             "mcp_tool_name": request_obj.tool_name,  # Keep original for reference
             "mcp_arguments": request_obj.arguments,  # Keep original for reference
+            "mcp_tool_description": request_obj.tool_description,
+            "mcp_tool_input_schema": request_obj.tool_input_schema,
             # Surface the per-MCP-server rate-limit identity so the
             # ParallelRequestLimiterV3 hook can apply mcp_rpm_limit on the
             # synthetic call_mcp_tool payload (otherwise a key with
@@ -1452,6 +1454,8 @@ class ProxyLogging:
             tool_name=kwargs.get("name", ""),
             arguments=kwargs.get("arguments", {}),
             server_name=kwargs.get("server_name"),
+            tool_description=kwargs.get("tool_description"),
+            tool_input_schema=kwargs.get("tool_input_schema"),
             user_api_key_auth=user_api_key_auth_dict,
             hidden_params=HiddenParams(),
         )

litellm/types/guardrails.py

@@ -8,6 +8,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,
 )
@@ -137,6 +140,7 @@ class SupportedGuardrailIntegrations(Enum):
     COMPRESR = "compresr"
     STRAIKER = "straiker"
     ALICE = "alice"
+    AGENT_365 = "agent_365"
     CONDUCT = "conduct"
 
 
@@ -945,7 +949,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."
         ),
     )
@@ -1083,6 +1087,7 @@ class LitellmParams(  # pyright: ignore[reportIncompatibleVariableOverride]  # o
     QostodianNexusConfigModel,
     VigilGuardGuardrailConfigModel,
     SingulrGuardrailConfigModel,
+    Agent365GuardrailConfigModel,
 ):
     guardrail: str = Field(description="The type of guardrail integration to use")
     mode: str | list[str] | Mode = Field(

litellm/types/mcp.py

@@ -391,6 +391,8 @@ class MCPPreCallRequestObject(BaseModel):
     tool_name: str
     arguments: dict[str, Any]
     server_name: str | None = None
+    tool_description: str | None = None
+    tool_input_schema: dict[str, object] | None = None
     user_api_key_auth: dict[str, Any] | None = None
     hidden_params: HiddenParams = HiddenParams()
 
@@ -414,6 +416,8 @@ class MCPDuringCallRequestObject(BaseModel):
     tool_name: str
     arguments: dict[str, Any]
     server_name: str | None = None
+    tool_description: str | None = None
+    tool_input_schema: dict[str, object] | None = None
     start_time: float | None = None
     hidden_params: HiddenParams = HiddenParams()
 

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/_experimental/mcp_server/outbound_credentials/test_adapter.py

@@ -584,6 +584,22 @@ def test_raise_token_exchange_challenge_is_rfc9728_invalid_token():
     assert "error_description=" in www
 
 
+def test_raise_token_exchange_challenge_explicit_resource_metadata_url_wins():
+    from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import (
+        raise_token_exchange_challenge,
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        raise_token_exchange_challenge(
+            _server(alias="obo-srv"),
+            root_path="/",
+            resource_metadata_url="https://gw.example.com/.well-known/oauth-protected-resource/obo-srv/mcp",
+        )
+    www = exc_info.value.headers["WWW-Authenticate"]
+    assert 'resource_metadata="https://gw.example.com/.well-known/oauth-protected-resource/obo-srv/mcp"' in www
+    assert "/mcp/obo-srv" not in www
+
+
 def test_raise_token_exchange_challenge_includes_server_root_path(monkeypatch):
     from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import (
         raise_token_exchange_challenge,

tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py

@@ -10,6 +10,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
+from fastapi.encoders import jsonable_encoder
 
 from litellm.types.mcp import MCPAuth
 
@@ -7124,6 +7125,81 @@ async def test_build_oauth_protected_resource_response_obo_end_to_end():
         global_mcp_server_manager.registry.clear()
 
 
+@pytest.fixture
+def agent_365_guardrail():
+    import litellm
+    from litellm.proxy.guardrails.guardrail_hooks.agent_365 import Agent365Guardrail
+
+    guardrail = Agent365Guardrail(
+        guardrail_name="agent-365-guard",
+        tenant_id="tenant-abc",
+        client_id="client-xyz",
+        client_secret="secret-123",
+        async_handler=AsyncMock(),
+        event_hook="pre_mcp_call",
+        default_on=True,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(guardrail)
+    try:
+        yield guardrail
+    finally:
+        litellm.logging_callback_manager.remove_callback_from_list_by_object(
+            litellm.callbacks, guardrail, require_self=False
+        )
+
+
+async def _agent_365_gated_prm(scopes):
+    from fastapi import Request
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+        _build_oauth_protected_resource_response,
+    )
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import global_mcp_server_manager
+    from litellm.proxy._types import MCPTransport
+    from litellm.types.mcp import MCPAuth
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+    global_mcp_server_manager.registry.clear()
+    global_mcp_server_manager.registry["tools"] = MCPServer(
+        server_id="tools",
+        name="tools",
+        server_name="tools",
+        alias="tools",
+        transport=MCPTransport.http,
+        auth_type=MCPAuth.none,
+        scopes=scopes,
+    )
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://litellm.example.com/"
+    mock_request.headers = {}
+    try:
+        return await _build_oauth_protected_resource_response(
+            request=mock_request, mcp_server_name="tools", use_standard_pattern=True
+        )
+    finally:
+        global_mcp_server_manager.registry.clear()
+
+
+@pytest.mark.asyncio
+async def test_agent_365_gated_server_prm_names_the_entra_tenant(agent_365_guardrail):
+    response = await _agent_365_gated_prm(scopes=["api://gateway-app/access_as_user"])
+    assert jsonable_encoder(response) == {
+        "authorization_servers": ["https://login.microsoftonline.com/tenant-abc/v2.0"],
+        "resource": "https://litellm.example.com/mcp/tools",
+        "scopes_supported": ["api://gateway-app/access_as_user"],
+    }
+
+
+@pytest.mark.asyncio
+async def test_agent_365_prm_defaults_scopeless_server_to_the_gateway_app_scope(agent_365_guardrail):
+    response = await _agent_365_gated_prm(scopes=None)
+    assert jsonable_encoder(response) == {
+        "authorization_servers": ["https://login.microsoftonline.com/tenant-abc/v2.0"],
+        "resource": "https://litellm.example.com/mcp/tools",
+        "scopes_supported": ["api://client-xyz/access_as_user"],
+    }
+
+
 def _token_request(headers):
     """A real Starlette request with case-insensitive headers (matches production)."""
     from starlette.requests import Request

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_guardrail_usage_monitor.py

@@ -48,6 +48,7 @@ def _bare_manager() -> MOD.MCPServerManager:
     reaches the guardrail hooks; they have their own coverage elsewhere.
     """
     mgr = MOD.MCPServerManager.__new__(MOD.MCPServerManager)
+    mgr._listed_tools_by_server_id = {}
     mgr.check_allowed_or_banned_tools = lambda name, server: True
     mgr.validate_allowed_params = lambda tool_name, arguments, server: None
 

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py

@@ -7109,6 +7109,8 @@ async def test_execute_mcp_tool_sets_model_in_model_call_details():
 
     fake_tool = MagicMock()
     fake_tool.name = "list_pets"
+    fake_tool.description = "test tool"
+    fake_tool.input_schema = {"type": "object"}
 
     start_time = datetime.now(timezone.utc)
     litellm_logging_obj, _ = function_setup(
@@ -7159,6 +7161,99 @@ async def test_execute_mcp_tool_sets_model_in_model_call_details():
     assert litellm_logging_obj.model == "MCP: list_pets"
 
 
+@pytest.mark.asyncio
+async def test_execute_mcp_tool_hands_openapi_registered_tool_metadata_to_pre_call_hooks():
+    """OpenAPI-generated tools dispatch through the local registry, so the pre-call hooks must get the
+    registered description and input schema on that path too, even when no tools/list ran first."""
+    from litellm.proxy._experimental.mcp_server import server as mcp_module
+
+    petstore = MCPServer(
+        server_id="petstore-id",
+        name="petstore",
+        server_name="petstore",
+        transport=MCPTransport.http,
+        url=None,
+        spec_path="https://example.com/petstore.yaml",
+    )
+    schema = {"type": "object", "properties": {"limit": {"type": "integer"}}}
+    mcp_module.global_mcp_tool_registry.register_tool(
+        name="petstore-list_pets", description="List the pets", input_schema=schema, handler=lambda limit: "ok"
+    )
+    manager = mcp_module.global_mcp_server_manager
+    manager._listed_tools_by_server_id.pop(petstore.server_id, None)
+    pre_call_tool_check = AsyncMock(return_value={})
+
+    try:
+        with (
+            patch.object(manager, "_get_mcp_server_from_tool_name", return_value=petstore),
+            patch.object(manager, "pre_call_tool_check", new=pre_call_tool_check),
+        ):
+            await mcp_module.execute_mcp_tool(
+                name="petstore-list_pets",
+                arguments={"limit": 10},
+                allowed_mcp_servers=[petstore],
+                start_time=datetime.now(),
+                user_api_key_auth=UserAPIKeyAuth(api_key="sk-user", user_id="alice"),
+            )
+    finally:
+        mcp_module.global_mcp_tool_registry.unregister_tools_with_prefix("petstore-")
+
+    handed_tool = pre_call_tool_check.call_args.kwargs["tool"]
+    assert (handed_tool.name, handed_tool.description, handed_tool.inputSchema) == (
+        "list_pets",
+        "List the pets",
+        schema,
+    )
+
+
+@pytest.mark.asyncio
+async def test_execute_mcp_tool_hands_hooks_the_metadata_of_the_operation_it_runs_when_names_collide():
+    """An OpenAPI operation whose name starts with its own server prefix must not be reported to the
+    pre-call hooks with the metadata of the shorter operation, since that is not the one that runs."""
+    from litellm.proxy._experimental.mcp_server import server as mcp_module
+
+    petstore = MCPServer(
+        server_id="petstore-id",
+        name="petstore",
+        server_name="petstore",
+        transport=MCPTransport.http,
+        url=None,
+        spec_path="https://example.com/petstore.yaml",
+    )
+    registry = mcp_module.global_mcp_tool_registry
+    registry.register_tool(name="petstore-get_pet", description="short", input_schema={}, handler=lambda: "short")
+    registry.register_tool(
+        name="petstore-petstore-get_pet",
+        description="long",
+        input_schema={"type": "object", "properties": {"petId": {"type": "integer"}}},
+        handler=lambda: "long",
+    )
+    manager = mcp_module.global_mcp_server_manager
+    pre_call_tool_check = AsyncMock(return_value={})
+
+    try:
+        with (
+            patch.object(manager, "_get_mcp_server_from_tool_name", return_value=petstore),
+            patch.object(manager, "pre_call_tool_check", new=pre_call_tool_check),
+        ):
+            result = await mcp_module.execute_mcp_tool(
+                name="petstore-petstore-get_pet",
+                arguments={},
+                allowed_mcp_servers=[petstore],
+                start_time=datetime.now(),
+                user_api_key_auth=UserAPIKeyAuth(api_key="sk-user", user_id="alice"),
+            )
+    finally:
+        registry.unregister_tools_with_prefix("petstore-")
+
+    handed_tool = pre_call_tool_check.call_args.kwargs["tool"]
+    assert (handed_tool.description, handed_tool.inputSchema) == (
+        "long",
+        {"type": "object", "properties": {"petId": {"type": "integer"}}},
+    )
+    assert result.content[0].text == "long"
+
+
 @pytest.mark.asyncio
 async def test_execute_mcp_tool_rest_unresolved_prefixed_name_routes_to_requested_server():
     """A prefixed REST name that resolves to no tool must still dispatch to the server_id.
@@ -8826,6 +8921,174 @@ class TestSingleServerPreflightReachesIdJag:
         preflight.assert_not_awaited()
 
 
+class TestAgent365ChallengeAtConnect:
+    """A missing Entra bearer on an Agent 365 gated server is challenged at connect (RFC 9728), where the
+    WWW-Authenticate header survives, instead of only inside the tools/call JSON-RPC error."""
+
+    GATEWAY_SCOPE = "api://gateway-app/access_as_user"
+
+    def _server(self, scopes: list[str] | None) -> MCPServer:
+        return MCPServer(
+            server_id="id-tools",
+            name="tools",
+            alias="tools",
+            server_name="tools",
+            url="https://tools.test/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.none,
+            scopes=scopes,
+            mcp_info={"server_name": "tools"},
+        )
+
+    @pytest.fixture
+    def agent_365_guardrail(self):
+        import litellm
+        from litellm.proxy.guardrails.guardrail_hooks.agent_365 import Agent365Guardrail
+
+        guardrail = Agent365Guardrail(
+            guardrail_name="agent-365-guard",
+            tenant_id="tenant-abc",
+            client_id="client-xyz",
+            client_secret="secret-123",
+            async_handler=AsyncMock(),
+            event_hook="pre_mcp_call",
+            default_on=True,
+        )
+        litellm.logging_callback_manager.add_litellm_callback(guardrail)
+        try:
+            yield guardrail
+        finally:
+            litellm.logging_callback_manager.remove_callback_from_list_by_object(
+                litellm.callbacks, guardrail, require_self=False
+            )
+
+    async def _connect(
+        self,
+        server: MCPServer,
+        oauth2_headers: dict[str, str] | None,
+        path: str = "/mcp/tools",
+        mount_scope: dict[str, str] | None = None,
+    ) -> HTTPException | None:
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        with (
+            patch.object(  # test-quality-ok: route wiring must use the manager's configured server
+                server_module.global_mcp_server_manager, "get_mcp_server_by_name", return_value=server
+            ),
+            patch.object(  # test-quality-ok: allowed-set resolution needs the DB; the test controls its answer
+                server_module, "_get_allowed_mcp_servers", AsyncMock(return_value=[])
+            ),
+        ):
+            try:
+                await server_module._raise_preemptive_401_for_unauthenticated_servers(
+                    scope={
+                        "type": "http",
+                        "method": "POST",
+                        "path": path,
+                        "scheme": "https",
+                        "server": ("gw.example.com", 443),
+                        "headers": [],
+                        **(mount_scope or {}),
+                    },
+                    mcp_servers=["tools"],
+                    oauth2_headers=oauth2_headers,
+                    mcp_server_auth_headers=None,
+                    user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-virtual-key", user_id="u-1"),
+                    client_ip=None,
+                )
+            except HTTPException as challenge:
+                return challenge
+            return None
+
+    @pytest.mark.asyncio
+    async def test_no_bearer_gets_the_discovery_challenge(self, agent_365_guardrail):
+        challenge = await self._connect(self._server([self.GATEWAY_SCOPE]), None)
+
+        assert challenge is not None and challenge.status_code == 401
+        www_authenticate = (challenge.headers or {}).get("WWW-Authenticate", "")
+        assert 'error="invalid_token"' in www_authenticate
+        assert (
+            'resource_metadata="https://gw.example.com/.well-known/oauth-protected-resource/mcp/tools"'
+            in www_authenticate
+        )
+
+    @pytest.mark.asyncio
+    async def test_legacy_route_challenge_points_at_its_own_metadata(self, agent_365_guardrail):
+        """RFC 9728 3.3: the metadata's ``resource`` must equal the URL the client connected to, so a
+        ``/{server}/mcp`` connect is sent to the ``/{server}/mcp`` document, not the ``/mcp/{server}`` one."""
+        challenge = await self._connect(self._server([self.GATEWAY_SCOPE]), None, path="/tools/mcp")
+
+        assert challenge is not None and challenge.status_code == 401
+        www_authenticate = (challenge.headers or {}).get("WWW-Authenticate", "")
+        assert (
+            'resource_metadata="https://gw.example.com/.well-known/oauth-protected-resource/tools/mcp"'
+            in www_authenticate
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "server_root, mount_scope",
+        [
+            ("", {"root_path": "/mcp", "app_root_path": ""}),
+            ("/litellm", {"root_path": "/litellm/mcp", "app_root_path": "/litellm"}),
+        ],
+    )
+    async def test_mounted_standard_route_is_challenged(self, agent_365_guardrail, server_root, mount_scope):
+        """``/mcp/{server}`` is served by the ``/mcp`` Mount, which moves the mount prefix into
+        ``root_path`` and leaves the app root (empty or SERVER_ROOT_PATH) in ``app_root_path``."""
+        with patch.dict(os.environ, {"SERVER_ROOT_PATH": server_root}):
+            challenge = await self._connect(
+                self._server([self.GATEWAY_SCOPE]), None, path=f"{server_root}/mcp/tools", mount_scope=mount_scope
+            )
+
+        assert challenge is not None and challenge.status_code == 401
+        www_authenticate = (challenge.headers or {}).get("WWW-Authenticate", "")
+        assert (
+            f'resource_metadata="https://gw.example.com{server_root}'
+            f'/.well-known/oauth-protected-resource{server_root}/mcp/tools"' in www_authenticate
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("path", ["/mcp", "/mcp/tools,other"])
+    async def test_aggregate_route_is_not_challenged_at_connect(self, agent_365_guardrail, path):
+        """The per-server metadata's ``resource`` can never equal the aggregate ``/mcp`` URL the client
+        connected to (RFC 9728 3.3), and one guarded server must not 401 a multi-server connect, so the
+        Agent 365 challenge is left to tools/call there."""
+        assert await self._connect(self._server([self.GATEWAY_SCOPE]), None, path=path) is None
+
+    @pytest.mark.asyncio
+    async def test_entra_assertion_present_connects(self, agent_365_guardrail):
+        bearer = {"Authorization": "Bearer eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ1LTEifQ.c2ln"}
+        assert await self._connect(self._server([self.GATEWAY_SCOPE]), bearer) is None
+
+    @pytest.mark.asyncio
+    async def test_litellm_key_in_authorization_is_still_challenged(self, agent_365_guardrail):
+        """A LiteLLM virtual key admits the caller but is no Entra assertion, so the tools/call would fail
+        401 inside JSON-RPC with the WWW-Authenticate header lost. The connect must challenge instead."""
+        challenge = await self._connect(
+            self._server([self.GATEWAY_SCOPE]), {"Authorization": "Bearer sk-litellm-virtual-key"}
+        )
+
+        assert challenge is not None and challenge.status_code == 401
+        www_authenticate = (challenge.headers or {}).get("WWW-Authenticate", "")
+        assert 'error="invalid_token"' in www_authenticate
+        assert (
+            'resource_metadata="https://gw.example.com/.well-known/oauth-protected-resource/mcp/tools"'
+            in www_authenticate
+        )
+
+    @pytest.mark.asyncio
+    async def test_scopeless_server_is_still_challenged(self, agent_365_guardrail):
+        challenge = await self._connect(self._server(None), None)
+
+        assert challenge is not None and challenge.status_code == 401
+        assert 'error="invalid_token"' in (challenge.headers or {}).get("WWW-Authenticate", "")
+
+    @pytest.mark.asyncio
+    async def test_no_registered_guardrail_means_no_challenge(self):
+        assert await self._connect(self._server([self.GATEWAY_SCOPE]), None) is None
+
+
 def _make_obo_server(alias: str) -> MCPServer:
     return MCPServer(
         server_id=f"id-{alias}",
@@ -8897,7 +9160,11 @@ class TestOboPreflightScopedToAllowedServers:
         _, preflight = await self._run(requested, allowed=[requested], user_api_key_auth=key)
 
         preflight.assert_awaited_once_with(
-            server=requested, oauth2_headers=self.SUBJECT_HEADERS, user_api_key_auth=key, raw_headers=None
+            server=requested,
+            oauth2_headers=self.SUBJECT_HEADERS,
+            user_api_key_auth=key,
+            raw_headers=None,
+            resource_metadata_url="/.well-known/oauth-protected-resource/mcp/obo_tools",
         )
 
 

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -6505,6 +6505,276 @@ class TestMCPServerManager:
         # Verify the MCP client call was awaited exactly once
         assert mock_client.call_tool.await_count == 1
 
+    @staticmethod
+    def _manager_ready_for_call_tool(listed_tools: list[MCPTool]) -> tuple[MCPServerManager, MagicMock]:
+        from mcp.types import CallToolResult
+
+        manager = MCPServerManager()
+        server = MCPServer(
+            server_id="test-server",
+            name="test-server",
+            transport=MCPTransport.http,
+            url="http://test-server.com",
+        )
+        manager.registry = {"test-server": server}
+        manager.tool_name_to_mcp_server_name_mapping["test_tool"] = "test-server"
+        manager.tool_name_to_mcp_server_name_mapping["test-server-test_tool"] = "test-server"
+        manager._create_prefixed_tools(listed_tools, server)
+
+        mock_client = AsyncMock()
+        mock_client.call_tool.return_value = MagicMock(spec=CallToolResult, content=[], isError=False)
+        manager._create_mcp_client = AsyncMock(return_value=mock_client)
+
+        proxy_logging_obj = MagicMock()
+        proxy_logging_obj._create_mcp_request_object_from_kwargs = MagicMock(return_value={})
+        proxy_logging_obj._convert_mcp_to_llm_format = MagicMock(return_value={})
+        proxy_logging_obj.pre_call_hook = AsyncMock(return_value={})
+        proxy_logging_obj.during_call_hook = AsyncMock(return_value=None)
+        return manager, proxy_logging_obj
+
+    @staticmethod
+    def _unrestricted_auth() -> MagicMock:
+        user_api_key_auth = MagicMock()
+        user_api_key_auth.object_permission = None
+        user_api_key_auth.object_permission_id = None
+        return user_api_key_auth
+
+    @pytest.mark.asyncio
+    async def test_call_tool_hands_listed_tool_description_and_schema_to_pre_call_hooks(self):
+        schema = {"type": "object", "properties": {"param": {"type": "string"}}, "required": ["param"]}
+        listed = [MCPTool(name="test_tool", description="Runs the test tool", inputSchema=schema)]
+        manager, proxy_logging_obj = self._manager_ready_for_call_tool(listed)
+
+        await manager.call_tool(
+            server_name="test-server",
+            name="test_tool",
+            arguments={"param": "value"},
+            user_api_key_auth=self._unrestricted_auth(),
+            proxy_logging_obj=proxy_logging_obj,
+        )
+
+        hook_kwargs = proxy_logging_obj._create_mcp_request_object_from_kwargs.call_args.args[0]
+        assert (hook_kwargs["tool_description"], hook_kwargs["tool_input_schema"]) == ("Runs the test tool", schema)
+
+    @pytest.mark.asyncio
+    async def test_call_tool_hands_listed_tool_metadata_to_during_call_hooks_through_real_conversion(self):
+        schema = {"type": "object", "properties": {"param": {"type": "string"}}}
+        listed = [MCPTool(name="test_tool", description="Runs the test tool", inputSchema=schema)]
+        manager, _ = self._manager_ready_for_call_tool(listed)
+        proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.pre_call_hook = AsyncMock(return_value={})
+        proxy_logging_obj.during_call_hook = AsyncMock(return_value=None)
+
+        await manager.call_tool(
+            server_name="test-server",
+            name="test_tool",
+            arguments={"param": "value"},
+            user_api_key_auth=UserAPIKeyAuth(api_key="sk-test"),
+            proxy_logging_obj=proxy_logging_obj,
+        )
+
+        during_data = proxy_logging_obj.during_call_hook.call_args.kwargs["data"]
+        assert (during_data["mcp_tool_description"], during_data["mcp_tool_input_schema"]) == (
+            "Runs the test tool",
+            schema,
+        )
+
+    @pytest.mark.asyncio
+    async def test_call_tool_passes_no_tool_metadata_when_tool_was_never_listed(self):
+        manager, proxy_logging_obj = self._manager_ready_for_call_tool(
+            [MCPTool(name="other_tool", description="Unrelated", inputSchema={"type": "object"})]
+        )
+
+        await manager.call_tool(
+            server_name="test-server",
+            name="test_tool",
+            arguments={"param": "value"},
+            user_api_key_auth=self._unrestricted_auth(),
+            proxy_logging_obj=proxy_logging_obj,
+        )
+
+        hook_kwargs = proxy_logging_obj._create_mcp_request_object_from_kwargs.call_args.args[0]
+        assert (hook_kwargs["tool_description"], hook_kwargs["tool_input_schema"]) == (None, None)
+
+    def test_get_listed_tool_resolves_prefixed_name_and_latest_listing(self):
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv", name="srv", transport=MCPTransport.http, url="http://srv")
+        manager._create_prefixed_tools([MCPTool(name="echo", description="v1", inputSchema={})], server)
+        manager._create_prefixed_tools([MCPTool(name="echo", description="v2", inputSchema={})], server)
+
+        by_prefixed_name = manager.get_listed_tool(server, "srv-echo")
+        assert by_prefixed_name is not None and by_prefixed_name.description == "v2"
+        assert manager.get_listed_tool(server, "missing") is None
+
+    def test_server_definition_change_drops_listed_tools(self):
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv", name="srv", transport=MCPTransport.http, url="http://srv")
+        other = MCPServer(server_id="other", name="other", transport=MCPTransport.http, url="http://other")
+        manager._create_prefixed_tools([MCPTool(name="echo", description="old", inputSchema={})], server)
+        manager._create_prefixed_tools([MCPTool(name="ping", description="kept", inputSchema={})], other)
+
+        manager._invalidate_server_definition_caches(server.server_id)
+
+        assert manager.get_listed_tool(server, "echo") is None
+        kept = manager.get_listed_tool(other, "ping")
+        assert kept is not None and kept.description == "kept"
+
+    @pytest.mark.asyncio
+    async def test_user_oauth_refresh_keeps_listed_tools(self):
+        """Tool definitions are server-wide, so one user's re-auth must not blank the metadata other
+        callers' tool calls hand to pre-call guardrails."""
+        manager = MCPServerManager()
+        server = MCPServer(server_id="srv", name="srv", transport=MCPTransport.http, url="http://srv")
+        manager._create_prefixed_tools([MCPTool(name="echo", description="shared", inputSchema={})], server)
+
+        await manager.invalidate_user_oauth_token_cache("alice", server.server_id)
+
+        listed = manager.get_listed_tool(server, "echo")
+        assert listed is not None and listed.description == "shared"
+
+    def test_per_caller_server_keeps_listed_tools_per_identity(self):
+        manager = MCPServerManager()
+        server = MCPServer(
+            server_id="srv",
+            name="srv",
+            transport=MCPTransport.http,
+            url="http://srv",
+            auth_type=MCPAuth.oauth2_token_exchange,
+        )
+        alice = UserAPIKeyAuth(user_id="alice", api_key="hashed-alice")
+        bob = UserAPIKeyAuth(user_id="bob", api_key="hashed-bob")
+        alice_schema = {"type": "object", "properties": {"path": {"type": "string"}}}
+        bob_schema = {"type": "object", "properties": {"path": {"type": "string"}, "site": {"type": "string"}}}
+        manager._create_prefixed_tools(
+            [MCPTool(name="read", description="alice view", inputSchema=alice_schema)], server, user_api_key_auth=alice
+        )
+        manager._create_prefixed_tools(
+            [MCPTool(name="read", description="bob view", inputSchema=bob_schema)], server, user_api_key_auth=bob
+        )
+
+        alice_tool = manager.get_listed_tool(server, "srv-read", alice)
+        bob_tool = manager.get_listed_tool(server, "srv-read", bob)
+        assert alice_tool is not None and (alice_tool.description, alice_tool.inputSchema) == ("alice view", alice_schema)
+        assert bob_tool is not None and (bob_tool.description, bob_tool.inputSchema) == ("bob view", bob_schema)
+        assert manager.get_listed_tool(server, "srv-read", UserAPIKeyAuth(user_id="carol", api_key="k")) is None
+
+        shared = MCPServer(server_id="shared", name="shared", transport=MCPTransport.http, url="http://shared")
+        manager._create_prefixed_tools(
+            [MCPTool(name="echo", description="everyone", inputSchema={})], shared, user_api_key_auth=alice
+        )
+        for_bob = manager.get_listed_tool(shared, "echo", bob)
+        assert for_bob is not None and for_bob.description == "everyone"
+
+    def test_per_caller_listed_tools_evict_oldest_caller_and_keep_shared(self):
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import _LISTED_TOOLS_CALLERS_PER_SERVER
+
+        manager = MCPServerManager()
+        server = MCPServer(
+            server_id="srv",
+            name="srv",
+            transport=MCPTransport.http,
+            url="http://srv",
+            auth_type=MCPAuth.oauth2_token_exchange,
+        )
+        manager._create_prefixed_tools([MCPTool(name="read", description="shared", inputSchema={})], server)
+        callers = [UserAPIKeyAuth(user_id=f"u{i}", api_key=f"k{i}") for i in range(_LISTED_TOOLS_CALLERS_PER_SERVER + 1)]
+        for caller in callers:
+            manager._create_prefixed_tools(
+                [MCPTool(name="read", description=caller.user_id, inputSchema={})], server, user_api_key_auth=caller
+            )
+        manager._create_prefixed_tools(
+            [MCPTool(name="read", description="u1 again", inputSchema={})], server, user_api_key_auth=callers[1]
+        )
+
+        assert manager.get_listed_tool(server, "srv-read", callers[0]) is None
+        second = manager.get_listed_tool(server, "srv-read", callers[1])
+        assert second is not None and second.description == "u1 again"
+        newest = manager.get_listed_tool(server, "srv-read", callers[-1])
+        assert newest is not None and newest.description == callers[-1].user_id
+        assert len(manager._listed_tools_by_server_id[server.server_id]) == _LISTED_TOOLS_CALLERS_PER_SERVER + 1
+        shared = manager.get_listed_tool(server, "srv-read")
+        assert shared is not None and shared.description == "shared"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("add_prefix", [True, False])
+    async def test_openapi_listing_records_listed_tools(self, add_prefix):
+        from litellm.proxy._experimental.mcp_server.tool_registry import global_mcp_tool_registry
+
+        server = MCPServer(
+            server_id="petstore-id",
+            name="petstore",
+            alias="petstore",
+            transport=MCPTransport.http,
+            url=None,
+            spec_path="/spec.yaml",
+        )
+        manager = MCPServerManager()
+        manager._create_mcp_client = AsyncMock(return_value=AsyncMock())
+
+        async def _handler(**kwargs):
+            return None
+
+        global_mcp_tool_registry.unregister_tools_with_prefix("petstore-")
+        global_mcp_tool_registry.register_tool(
+            name="petstore-list_pets",
+            description="List pets",
+            input_schema={"type": "object", "properties": {"limit": {"type": "integer"}}},
+            handler=_handler,
+        )
+        try:
+            listed = await manager._get_tools_from_server(server=server, add_prefix=add_prefix)
+        finally:
+            global_mcp_tool_registry.unregister_tools_with_prefix("petstore-")
+
+        assert [t.name for t in listed] == ["petstore-list_pets" if add_prefix else "list_pets"]
+        for name in ("list_pets", "petstore-list_pets"):
+            tool = manager.get_listed_tool(server, name)
+            assert tool is not None and tool.description == "List pets"
+            assert tool.inputSchema["properties"] == {"limit": {"type": "integer"}}
+
+    @pytest.mark.asyncio
+    async def test_openapi_listing_ignores_overlapping_server_prefix(self):
+        from litellm.proxy._experimental.mcp_server.tool_registry import global_mcp_tool_registry
+
+        server = MCPServer(
+            server_id="pet-id",
+            name="pet",
+            alias="pet",
+            transport=MCPTransport.http,
+            url=None,
+            spec_path="/spec.yaml",
+        )
+        manager = MCPServerManager()
+        manager._create_mcp_client = AsyncMock(return_value=AsyncMock())
+
+        async def _handler(**kwargs):
+            return None
+
+        for prefix in ("pet-", "petstore-"):
+            global_mcp_tool_registry.unregister_tools_with_prefix(prefix)
+        global_mcp_tool_registry.register_tool(
+            name="pet-petstore-list",
+            description="Local pet tool",
+            input_schema={"type": "object", "properties": {"limit": {"type": "integer"}}},
+            handler=_handler,
+        )
+        global_mcp_tool_registry.register_tool(
+            name="petstore-list",
+            description="Foreign petstore tool",
+            input_schema={"type": "object", "properties": {"status": {"type": "string"}}},
+            handler=_handler,
+        )
+        try:
+            listed = await manager._get_tools_from_server(server=server, add_prefix=True)
+        finally:
+            for prefix in ("pet-", "petstore-"):
+                global_mcp_tool_registry.unregister_tools_with_prefix(prefix)
+
+        assert [t.name for t in listed] == ["pet-petstore-list"]
+        tool = manager.get_listed_tool(server, "petstore-list")
+        assert tool is not None and tool.description == "Local pet tool"
+        assert tool.inputSchema["properties"] == {"limit": {"type": "integer"}}
+
     @pytest.mark.asyncio
     async def test_get_allowed_mcp_servers_with_user_api_key_auth(self):
         """

tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_tool_auth.py

@@ -43,6 +43,8 @@ async def test_openapi_local_tool_runs_pre_call_tool_check():
 
     fake_tool = MagicMock()
     fake_tool.name = "list_pets"
+    fake_tool.description = "test tool"
+    fake_tool.input_schema = {"type": "object"}
 
     pre_call = AsyncMock(return_value={})
     handle_local = AsyncMock(return_value=[])
@@ -124,6 +126,8 @@ async def test_openapi_local_tool_blocked_when_pre_call_check_raises():
 
     fake_tool = MagicMock()
     fake_tool.name = "delete_pet"
+    fake_tool.description = "test tool"
+    fake_tool.input_schema = {"type": "object"}
 
     pre_call = AsyncMock(
         side_effect=HTTPException(status_code=403, detail="not allowed")
@@ -186,6 +190,8 @@ async def test_openapi_local_tool_denied_when_server_not_resolvable():
 
     fake_tool = MagicMock()
     fake_tool.name = "list_pets"
+    fake_tool.description = "test tool"
+    fake_tool.input_schema = {"type": "object"}
 
     pre_call = AsyncMock(return_value={})
     handle_local = AsyncMock(return_value=[])
@@ -270,6 +276,8 @@ async def test_openapi_local_tool_injects_resolved_oauth_token():
 
     fake_tool = MagicMock()
     fake_tool.name = "get_values"
+    fake_tool.description = "test tool"
+    fake_tool.input_schema = {"type": "object"}
     captured: dict = {}
 
     async def handle_local(_name, _arguments):
@@ -616,6 +624,8 @@ async def test_per_server_auth_header_reaches_both_openapi_dispatch_arms(dispatc
         if dispatch_arm == "local_registry":
             fake_tool = MagicMock()
             fake_tool.name = "list_reports"
+            fake_tool.description = "test tool"
+            fake_tool.input_schema = {"type": "object"}
             with (
                 patch.object(manager, "_get_mcp_server_from_tool_name", return_value=server),
                 patch.object(mcp_module.global_mcp_tool_registry, "get_tool", return_value=fake_tool),
@@ -687,6 +697,8 @@ async def test_local_dispatch_reports_the_outcome_instead_of_success(failure: st
 
     fake_tool = MagicMock()
     fake_tool.name = "list_reports"
+    fake_tool.description = "test tool"
+    fake_tool.input_schema = {"type": "object"}
     fake_tool.handler = raising_handler
     server = MCPServer(
         server_id="srv-openapi",

tests/test_litellm/proxy/utils/proxy_logging/test_mcp_bridging.py

@@ -381,6 +381,26 @@ def test_create_mcp_request_object_from_kwargs_full(proxy_logging, make_user_api
     assert snapshot == {"tool_name": "calc", "arguments": {"x": 1}, "server_name": "math", "auth_user_id": "u-1"}
 
 
+def test_mcp_tool_metadata_flows_from_kwargs_to_synthetic_data(proxy_logging):
+    schema = {"type": "object", "properties": {"x": {"type": "integer"}}}
+    obj = proxy_logging._create_mcp_request_object_from_kwargs(
+        kwargs={
+            "name": "calc",
+            "arguments": {"x": 1},
+            "tool_description": "Adds numbers",
+            "tool_input_schema": schema,
+        }
+    )
+    out = proxy_logging._convert_mcp_to_llm_format(request_obj=obj, kwargs={})
+    assert (out["mcp_tool_description"], out["mcp_tool_input_schema"]) == ("Adds numbers", schema)
+
+
+def test_mcp_tool_metadata_absent_when_tool_was_never_listed(proxy_logging):
+    obj = proxy_logging._create_mcp_request_object_from_kwargs(kwargs={"name": "calc", "arguments": {}})
+    out = proxy_logging._convert_mcp_to_llm_format(request_obj=obj, kwargs={})
+    assert (out["mcp_tool_description"], out["mcp_tool_input_schema"]) == (None, None)
+
+
 def test_create_mcp_request_object_from_kwargs_empty(proxy_logging):
     obj = proxy_logging._create_mcp_request_object_from_kwargs(kwargs={})
     snapshot = {

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

@@ -318,6 +318,13 @@ 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,
+  },
   conduct: {
     provider: "Conduct",
     guardrailNameSuggestion: "Conduct Guard",

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

@@ -28,6 +28,7 @@ const EXPECTED_PARTNER_LOGO_FILES: Record<string, string> = {
   repelloai: "repelloai.png",
   straiker: "straiker.svg",
   alice: "alice.svg",
+  agent_365: "microsoft_azure.svg",
   conduct: "conduct.png",
 };
 

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

@@ -474,6 +474,16 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
     tags: ["Content Moderation", "Prompt Injection", "PII", "Policy"],
     providerKey: "Alice",
   },
+  {
+    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",
+  },
   {
     id: "conduct",
     name: "Conduct Guard",

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

@@ -210,6 +210,7 @@ export const guardrailLogoMap = {
   "RepelloAI Argus": repelloAiLogo.src,
   Straiker: straikerLogo.src,
   Alice: aliceLogo.src,
+  "Microsoft Agent 365": microsoftAzureLogo.src,
   "Conduct Guard": conductLogo.src,
 } satisfies Record<string, string>;
 

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -23898,7 +23898,7 @@ export interface components {
             timeout?: number | null;
             /**
              * Unreachable Fallback
-             * @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'. 'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed.
+             * @description Behavior when a guardrail endpoint is unreachable due to network errors. 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.
              * @default fail_closed
              * @enum {string}
              */
@@ -30547,6 +30547,11 @@ export interface components {
              * @description Custom advisory message template used when on_flagged='inject_system_message'. Must contain a {reason} placeholder. Defaults to a generic advisory message if unset.
              */
             advisory_system_message?: string | null;
+            /**
+             * Agent Id
+             * @description Agent identity reported to Agent 365 with every tool evaluation. When unset, the caller's key alias is used.
+             */
+            agent_id?: string | null;
             /**
              * Akto Account Id
              * @description Akto account ID for multi-tenant deployments. Env: AKTO_ACCOUNT_ID. Default: '1000000'.
@@ -30748,6 +30753,16 @@ export interface components {
              * @default 25000
              */
             chunk_budget_chars: number;
+            /**
+             * Client Id
+             * @description Client id of the gateway's Entra app registration (a confidential client). Falls back to the AGENT365_CLIENT_ID environment variable.
+             */
+            client_id?: string | null;
+            /**
+             * Client Secret
+             * @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.
+             */
+            client_secret?: string | null;
             /**
              * Confidence Threshold
              * @description Only block or mask when detection confidence >= this value; below threshold, allow or log_only.
@@ -31181,6 +31196,11 @@ export interface components {
              * @description The message the bot speaks aloud when a /v1/realtime guardrail fires. Falls back to violation_message_template if not set.
              */
             realtime_violation_message?: string | null;
+            /**
+             * Resource App Id
+             * @description Application id of the Agent 365 resource the OBO token is minted for. Defaults to the production resource ea9ffc3e-8a23-4a7d-836d-234d7c7565c1; the Test and PreProd environments use a different id. Falls back to the AGENT365_RESOURCE_APP_ID environment variable.
+             */
+            resource_app_id?: string | null;
             /**
              * Rules
              * @description Ordered allow/deny rules. Patterns use regex for tool names/types and optional regex constraints on tool arguments.
@@ -31282,6 +31302,11 @@ export interface components {
              * @description The ID of your Model Armor template
              */
             template_id?: string | null;
+            /**
+             * Tenant Id
+             * @description Entra tenant id used for the On-Behalf-Of token exchange. Falls back to the AGENT365_TENANT_ID environment variable.
+             */
+            tenant_id?: string | null;
             /**
              * Timeout
              * @description Per-request timeout for the guardrail provider API call (seconds). Accepts int, float, or numeric string; coerced to float on load. Each guardrail handler chooses its own default when unset.