scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-25 05:54:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 175

refactor(proxy): make MCP trust scoring any-discipline clean

Browse source 
Drop the call_tool rewrite that added trust-ranked fallback-on-failure. It rewrote a large, previously type-clean stretch of the critical tool-call path, which was both a risky drive-by refactor and the source of most Any-discipline findings. Trust filtering still runs on every list/call path, and the ranking helpers stay as tested public API.

Tighten the scoring module's typed/untyped boundaries so the gate passes without blanket suppressions: pydantic fields use Annotated plus literal defaults instead of Field values, the shared cache stores the result JSON string directly so there is no dict round-trip, parsing goes through TypeAdapter instead of isinstance narrowing, and the api_key secret is read via the typed get_secret_str. The only remaining any-ok markers sit on awaited coroutines whose Send/Recv Any is a typing artifact and on required model fields mypy exports as Any, each with a reason.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
marcusacosta 2026-06-16 22:06:15 -07:00
parent 9942d28a68
commit 41bf08cc2b
5 changed files with 224 additions and 350 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

372
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -13,19 +13,7 @@ import json
import os
import re
import time
from typing import (
Any,
Callable,
Dict,
List,
Literal,
Optional,
Sequence,
Set,
Tuple,
Union,
cast,
)
from typing import Any, Callable, Dict, List, Literal, Optional, Set, Tuple, Union, cast
from urllib.parse import urlparse
import anyio
@ -1647,12 +1635,10 @@ class MCPServerManager:
for server_id in allowed_mcp_servers
if (server := self.get_mcp_server_by_id(server_id)) is not None
)
allowed_mcp_servers = [
server.server_id
for server in await apply_trust_filter_to_allowed_mcp_servers(
allowed_server_objects
)
]
trusted_servers = await apply_trust_filter_to_allowed_mcp_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
allowed_server_objects
)
allowed_mcp_servers = [server.server_id for server in trusted_servers]
verbose_logger.debug("SERVER MANAGER LISTING TOOLS")
@ -3583,246 +3569,6 @@ class MCPServerManager:
return cast(CallToolResult, result)
@staticmethod
def _should_attempt_trust_fallback(error: BaseException) -> bool:
if isinstance(
error, (BlockedPiiEntityError, GuardrailRaisedException, ValueError)
):
return False
if isinstance(error, HTTPException):
return error.status_code not in {401, 403, 412}
return True
def _server_exposes_tool(
self,
server: MCPServer,
tool_name: str,
server_name: str,
) -> bool:
if server_name:
server_matches = False
for identifier in (server.alias, server.server_name, server.name):
if identifier and normalize_server_name(
identifier
) == normalize_server_name(server_name):
server_matches = True
break
if not server_matches:
return False
for known_prefix in iter_known_server_prefixes(server):
candidate_names = (
tool_name,
add_server_prefix_to_name(tool_name, known_prefix),
)
for candidate_name in candidate_names:
mapped_server = self.tool_name_to_mcp_server_name_mapping.get(
candidate_name
)
if mapped_server is None:
continue
normalized_mapped = normalize_server_name(mapped_server)
for identifier in (
server.alias,
server.server_name,
server.name,
known_prefix,
):
if identifier and normalized_mapped == normalize_server_name(
identifier
):
return True
return False
async def _ordered_mcp_servers_for_tool_call(
self,
server_name: str,
name: str,
allowed_mcp_servers: Sequence[MCPServer] | None,
) -> tuple[MCPServer, ...]:
primary = self._resolve_mcp_server_for_tool_call(server_name, name)
from litellm.proxy._experimental.mcp_server.mcp_trust_scoring import (
get_mcp_trust_scoring_client,
rank_mcp_servers_by_trust,
)
client = get_mcp_trust_scoring_client()
if client is None or not client.enabled or not allowed_mcp_servers:
return (primary,)
allowed_ids = {server.server_id for server in allowed_mcp_servers}
if primary.server_id not in allowed_ids:
return (primary,)
candidates = tuple(
server
for server in allowed_mcp_servers
if self._server_exposes_tool(
server=server,
tool_name=name,
server_name=server_name,
)
)
if len(candidates) <= 1:
return candidates if candidates else (primary,)
ranked = await rank_mcp_servers_by_trust(candidates)
return ranked if ranked else (primary,)
async def _call_tool_on_mcp_server(
self,
*,
mcp_server: MCPServer,
server_name: str,
name: str,
arguments: dict[str, Any],
user_api_key_auth: UserAPIKeyAuth | None,
mcp_auth_header: str | None,
mcp_server_auth_headers: dict[str, dict[str, str]] | None,
proxy_logging_obj: ProxyLogging | None,
oauth2_headers: dict[str, str] | None,
raw_headers: dict[str, str] | None,
host_progress_callback: Callable | None,
start_time: datetime.datetime,
) -> CallToolResult:
hook_result: dict[str, Any] = {}
if proxy_logging_obj:
hook_result = await self.pre_call_tool_check(
name=name,
arguments=arguments,
server_name=server_name,
user_api_key_auth=user_api_key_auth,
proxy_logging_obj=proxy_logging_obj,
server=mcp_server,
raw_headers=raw_headers,
)
if "arguments" in hook_result:
arguments = hook_result["arguments"]
tasks: list = []
if proxy_logging_obj:
during_hook_task = self._create_during_hook_task(
name=name,
arguments=arguments,
server_name_from_prefix=server_name,
user_api_key_auth=user_api_key_auth,
proxy_logging_obj=proxy_logging_obj,
start_time=start_time,
)
tasks.append(during_hook_task)
resolved_oauth2_headers = await self._resolve_oauth2_headers_for_tool_call(
mcp_server, oauth2_headers, user_api_key_auth
)
if mcp_server.spec_path:
verbose_logger.debug(
"Calling OpenAPI tool %s directly via HTTP handler", name
)
if hook_result.get("extra_headers"):
verbose_logger.warning(
"pre_mcp_call hook returned extra_headers for OpenAPI-backed "
"MCP server '%s' — header injection is not supported for "
"OpenAPI servers; headers will be ignored. Use SSE/HTTP "
"transport to enable hook header injection.",
server_name,
)
tasks.append(
asyncio.create_task(
self._call_openapi_tool_handler(mcp_server, name, arguments)
)
)
return await self._gather_openapi_tool_tasks(tasks, proxy_logging_obj)
return await self._call_regular_mcp_tool(
mcp_server=mcp_server,
original_tool_name=name,
arguments=arguments,
tasks=tasks,
mcp_auth_header=mcp_auth_header,
mcp_server_auth_headers=mcp_server_auth_headers,
oauth2_headers=resolved_oauth2_headers,
raw_headers=raw_headers,
proxy_logging_obj=proxy_logging_obj,
host_progress_callback=host_progress_callback,
hook_extra_headers=hook_result.get("extra_headers"),
user_api_key_auth=user_api_key_auth,
)
async def call_tool(
self,
server_name: str,
name: str,
arguments: dict[str, Any],
user_api_key_auth: UserAPIKeyAuth | None = None,
mcp_auth_header: str | None = None,
mcp_server_auth_headers: dict[str, dict[str, str]] | None = None,
proxy_logging_obj: ProxyLogging | None = None,
oauth2_headers: dict[str, str] | None = None,
raw_headers: dict[str, str] | None = None,
host_progress_callback: Callable | None = None,
allowed_mcp_servers: Sequence[MCPServer] | None = None,
) -> CallToolResult:
"""
Call a tool with the given name and arguments
Args:
server_name: Server name
name: Tool name
arguments: Tool arguments
user_api_key_auth: User authentication
mcp_auth_header: MCP auth header (deprecated)
mcp_server_auth_headers: Optional dict of server-specific auth headers {server_alias: auth_value}
proxy_logging_obj: Optional ProxyLogging object for hook integration
allowed_mcp_servers: Optional allowed servers for trust-ranked fallback
Returns:
CallToolResult from the MCP server
"""
start_time = datetime.datetime.now()
ordered_servers = await self._ordered_mcp_servers_for_tool_call(
server_name=server_name,
name=name,
allowed_mcp_servers=allowed_mcp_servers,
)
last_error: BaseException | None = None
for attempt_index, mcp_server in enumerate(ordered_servers):
effective_server_name = server_name or mcp_server.name
try:
return await self._call_tool_on_mcp_server(
mcp_server=mcp_server,
server_name=effective_server_name,
name=name,
arguments=arguments,
user_api_key_auth=user_api_key_auth,
mcp_auth_header=mcp_auth_header,
mcp_server_auth_headers=mcp_server_auth_headers,
proxy_logging_obj=proxy_logging_obj,
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
host_progress_callback=host_progress_callback,
start_time=start_time,
)
except Exception as exc:
if not self._should_attempt_trust_fallback(exc):
raise
if attempt_index + 1 >= len(ordered_servers):
raise
last_error = exc
verbose_logger.debug(
"MCP trust fallback after failure on server %s for tool %s: %s",
mcp_server.name,
name,
exc,
)
if last_error is not None:
raise last_error
raise ValueError(f"Tool {name} not found")
def _resolve_mcp_server_for_tool_call(
self,
server_name: str,
@ -3911,8 +3657,8 @@ class MCPServerManager:
async def _gather_openapi_tool_tasks(
self,
tasks: list[Any],
proxy_logging_obj: ProxyLogging | None,
tasks: List[Any],
proxy_logging_obj: Optional[ProxyLogging],
) -> CallToolResult:
"""Await OpenAPI tool tasks and return the tool call result."""
try:
@ -3929,6 +3675,110 @@ class MCPServerManager:
)
raise e
async def call_tool(
self,
server_name: str,
name: str,
arguments: Dict[str, Any],
user_api_key_auth: Optional[UserAPIKeyAuth] = None,
mcp_auth_header: Optional[str] = None,
mcp_server_auth_headers: Optional[Dict[str, Dict[str, str]]] = None,
proxy_logging_obj: Optional[ProxyLogging] = None,
oauth2_headers: Optional[Dict[str, str]] = None,
raw_headers: Optional[Dict[str, str]] = None,
host_progress_callback: Optional[Callable] = None,
) -> CallToolResult:
"""
Call a tool with the given name and arguments
Args:
server_name: Server name
name: Tool name
arguments: Tool arguments
user_api_key_auth: User authentication
mcp_auth_header: MCP auth header (deprecated)
mcp_server_auth_headers: Optional dict of server-specific auth headers {server_alias: auth_value}
proxy_logging_obj: Optional ProxyLogging object for hook integration
Returns:
CallToolResult from the MCP server
"""
start_time = datetime.datetime.now()
mcp_server = self._resolve_mcp_server_for_tool_call(server_name, name)
#########################################################
# Pre MCP Tool Call Hook
# Allow validation and modification of tool calls before execution
# Using standard pre_call_hook
#########################################################
hook_result: Dict[str, Any] = {}
if proxy_logging_obj:
hook_result = await self.pre_call_tool_check(
name=name,
arguments=arguments,
server_name=server_name,
user_api_key_auth=user_api_key_auth,
proxy_logging_obj=proxy_logging_obj,
server=mcp_server,
raw_headers=raw_headers,
)
if "arguments" in hook_result:
arguments = hook_result["arguments"]
# Prepare tasks for during hooks
tasks = []
if proxy_logging_obj:
during_hook_task = self._create_during_hook_task(
name=name,
arguments=arguments,
server_name_from_prefix=server_name,
user_api_key_auth=user_api_key_auth,
proxy_logging_obj=proxy_logging_obj,
start_time=start_time,
)
tasks.append(during_hook_task)
oauth2_headers = await self._resolve_oauth2_headers_for_tool_call(
mcp_server, oauth2_headers, user_api_key_auth
)
# For OpenAPI servers, call the tool handler directly instead of via MCP client
if mcp_server.spec_path:
verbose_logger.debug(
"Calling OpenAPI tool %s directly via HTTP handler", name
)
if hook_result.get("extra_headers"):
verbose_logger.warning(
"pre_mcp_call hook returned extra_headers for OpenAPI-backed "
"MCP server '%s' — header injection is not supported for "
"OpenAPI servers; headers will be ignored. Use SSE/HTTP "
"transport to enable hook header injection.",
server_name,
)
tasks.append(
asyncio.create_task(
self._call_openapi_tool_handler(mcp_server, name, arguments)
)
)
else:
return await self._call_regular_mcp_tool(
mcp_server=mcp_server,
original_tool_name=name,
arguments=arguments,
tasks=tasks,
mcp_auth_header=mcp_auth_header,
mcp_server_auth_headers=mcp_server_auth_headers,
oauth2_headers=oauth2_headers,
raw_headers=raw_headers,
proxy_logging_obj=proxy_logging_obj,
host_progress_callback=host_progress_callback,
hook_extra_headers=hook_result.get("extra_headers"),
user_api_key_auth=user_api_key_auth,
)
return await self._gather_openapi_tool_tasks(tasks, proxy_logging_obj)
#########################################################
# End of Methods that call the upstream MCP servers
#########################################################

167
litellm/proxy/_experimental/mcp_server/mcp_trust_scoring.py
View file

@ -11,10 +11,10 @@ from __future__ import annotations
import asyncio
import hashlib
from dataclasses import dataclass
from typing import TYPE_CHECKING, Callable, Literal, Protocol, Sequence, cast
from typing import TYPE_CHECKING, Annotated, Callable, Literal, Protocol, Sequence, cast
import httpx
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError
from litellm._logging import verbose_logger
from litellm.caching.in_memory_cache import InMemoryCache
@ -34,44 +34,50 @@ class MCPTrustScoringConfig(BaseModel):
model_config = ConfigDict(frozen=True)
enabled: bool = False
min_trust_score: float = Field(
default=DEFAULT_MCP_TRUST_SCORING_MIN_SCORE,
ge=0.0,
le=1.0,
description="Minimum normalized trust score (0-1) required to expose or route to a server",
)
cache_ttl_seconds: int = Field(
default=DEFAULT_MCP_TRUST_SCORING_CACHE_TTL,
ge=1,
description="Seconds to cache Dominion Observatory responses per server URL",
)
api_url: str = Field(
default=DEFAULT_MCP_TRUST_SCORING_API_URL,
description="Dominion Observatory trust endpoint base URL",
)
fail_open: bool = Field(
default=True,
description="When True, keep servers reachable if DO is down, rate-limited, or has no score yet",
)
api_key: str | None = Field(
default=None,
description="Optional API key for paid Dominion Observatory tiers",
)
min_trust_score: Annotated[
float,
Field(
ge=0.0,
le=1.0,
description="Minimum normalized trust score (0-1) required to expose or route to a server",
),
] = DEFAULT_MCP_TRUST_SCORING_MIN_SCORE
cache_ttl_seconds: Annotated[
int,
Field(
ge=1,
description="Seconds to cache Dominion Observatory responses per server URL",
),
] = DEFAULT_MCP_TRUST_SCORING_CACHE_TTL
api_url: Annotated[
str,
Field(description="Dominion Observatory trust endpoint base URL"),
] = DEFAULT_MCP_TRUST_SCORING_API_URL
fail_open: Annotated[
bool,
Field(
description="When True, keep servers reachable if DO is down, rate-limited, or has no score yet"
),
] = True
api_key: Annotated[
str | None,
Field(description="Optional API key for paid Dominion Observatory tiers"),
] = None
class DominionObservatoryTrustResult(BaseModel):
model_config = ConfigDict(frozen=True)
server_url: str
found: bool
trust_score_normalized: float | None = Field(
default=None,
description="Trust score normalized to 0-1",
)
trust_score_raw: float | None = Field(
default=None,
description="Raw trust score from Dominion Observatory (0-100 scale)",
)
server_url: str # any-ok: required pydantic field, mypy exports the field symbol as Any but the runtime type is str
found: bool # any-ok: required pydantic field, mypy exports the field symbol as Any but the runtime type is bool
trust_score_normalized: Annotated[
float | None,
Field(description="Trust score normalized to 0-1"),
] = None
trust_score_raw: Annotated[
float | None,
Field(description="Raw trust score from Dominion Observatory (0-100 scale)"),
] = None
lookup_status: Literal["ok", "skipped", "error"] = "ok"
message: str | None = None
@ -90,12 +96,6 @@ class DominionObservatoryTrustResult(BaseModel):
return (0, -self.trust_score_normalized)
class _TrustCacheEntry(BaseModel):
model_config = ConfigDict(frozen=True)
payload_json: str
class _DominionObservatoryTrustApiResponse(BaseModel):
model_config = ConfigDict(extra="ignore", frozen=True)
@ -139,8 +139,8 @@ class _HttpClient(Protocol):
@dataclass(frozen=True, slots=True)
class _ScoredServer:
server: MCPServer
trust: DominionObservatoryTrustResult
server: MCPServer # any-ok: required dataclass field, mypy exports the field symbol as Any but the runtime type is MCPServer
trust: DominionObservatoryTrustResult # any-ok: required dataclass field, mypy exports the field symbol as Any but the runtime type is typed
def _normalize_server_url(server_url: str) -> str:
@ -170,7 +170,11 @@ def parse_trust_response(
server_url: str,
payload: object,
) -> DominionObservatoryTrustResult:
if not isinstance(payload, dict):
try:
parsed = TypeAdapter(_DominionObservatoryTrustApiResponse).validate_python(
payload
)
except ValidationError:
return DominionObservatoryTrustResult(
server_url=server_url,
found=False,
@ -178,8 +182,6 @@ def parse_trust_response(
message="Invalid Dominion Observatory response shape",
)
parsed = TypeAdapter(_DominionObservatoryTrustApiResponse).validate_python(payload)
return DominionObservatoryTrustResult(
server_url=server_url,
found=parsed.found,
@ -259,7 +261,9 @@ class MCPTrustScoringClient:
try:
shared_cache = self._cache_getter()
shared = await shared_cache.async_get_cache(cache_key)
shared = await shared_cache.async_get_cache( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is narrowed below
cache_key
)
except Exception:
shared = None
@ -270,16 +274,6 @@ class MCPTrustScoringClient:
ttl=self._config.cache_ttl_seconds,
)
return DominionObservatoryTrustResult.model_validate_json(shared)
if isinstance(shared, dict):
entry = _TrustCacheEntry.model_validate(shared)
self._local_cache.set_cache(
cache_key,
entry.payload_json,
ttl=self._config.cache_ttl_seconds,
)
return DominionObservatoryTrustResult.model_validate_json(
entry.payload_json
)
return None
async def _write_cached_result(
@ -296,9 +290,9 @@ class MCPTrustScoringClient:
)
try:
shared_cache = self._cache_getter()
await shared_cache.async_set_cache(
await shared_cache.async_set_cache( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the stored value is a typed JSON string
cache_key,
_TrustCacheEntry(payload_json=payload_json).model_dump(),
payload_json,
ttl=self._config.cache_ttl_seconds,
)
except Exception as exc:
@ -310,17 +304,25 @@ class MCPTrustScoringClient:
async def get_trust_score(self, server_url: str) -> DominionObservatoryTrustResult:
normalized_url = _normalize_server_url(server_url)
cached = await self._read_cached_result(normalized_url)
cached = await self._read_cached_result( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
normalized_url
)
if cached is not None:
return cached
async with self._lock_for_url(normalized_url):
cached = await self._read_cached_result(normalized_url)
cached = await self._read_cached_result( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
normalized_url
)
if cached is not None:
return cached
result = await self._fetch_trust_score(normalized_url)
await self._write_cached_result(normalized_url, result)
result = await self._fetch_trust_score( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
normalized_url
)
await self._write_cached_result( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and returns None
normalized_url, result
)
return result
async def _fetch_trust_score(
@ -332,7 +334,7 @@ class MCPTrustScoringClient:
try:
client = self._http_client_getter()
response = await client.get(
response = await client.get( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is an httpx.Response
self._config.api_url,
params={"url": server_url},
headers=headers or None,
@ -374,7 +376,9 @@ class MCPTrustScoringClient:
lookup_status="skipped",
message="Server has no HTTP URL to score",
)
return await self.get_trust_score(server.url)
return await self.get_trust_score( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
server.url
)
async def score_servers(
self,
@ -383,8 +387,13 @@ class MCPTrustScoringClient:
if not servers:
return ()
trust_results = await asyncio.gather(
*(self.get_trust_score_for_server(server) for server in servers)
trust_results = await asyncio.gather( # any-ok: awaited coroutine, gather's element Any is a typing artifact and each result is a typed DominionObservatoryTrustResult
*( # any-ok: generator of coroutines, the Coroutine Send/Recv Any is a typing artifact
self.get_trust_score_for_server( # any-ok: coroutine expression, Send/Recv Any is a typing artifact and the result is fully typed
server
)
for server in servers
)
)
return tuple(
_ScoredServer(server=server, trust=trust)
@ -398,7 +407,9 @@ class MCPTrustScoringClient:
if not self.enabled:
return tuple(servers)
scored = await self.score_servers(servers)
scored = await self.score_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
servers
)
return tuple(
entry.server
for entry in scored
@ -412,7 +423,9 @@ class MCPTrustScoringClient:
if not self.enabled:
return tuple(servers)
scored = await self.score_servers(servers)
scored = await self.score_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
servers
)
ordered = sorted(scored, key=lambda entry: entry.trust.sort_key())
return tuple(entry.server for entry in ordered)
@ -450,9 +463,9 @@ def _resolve_trust_scoring_config_values(
resolved = dict(config)
api_key = resolved.get("api_key")
if isinstance(api_key, str) and api_key.startswith("os.environ/"):
from litellm.secret_managers.main import get_secret # noqa: PLC0415
from litellm.secret_managers.main import get_secret_str # noqa: PLC0415
resolved["api_key"] = get_secret(api_key)
resolved["api_key"] = get_secret_str(api_key)
return resolved
@ -487,7 +500,9 @@ async def filter_mcp_servers_by_trust(
active_client = client or get_mcp_trust_scoring_client()
if active_client is None:
return tuple(servers)
return await active_client.filter_servers_by_trust(servers)
return await active_client.filter_servers_by_trust( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
servers
)
async def rank_mcp_servers_by_trust(
@ -498,7 +513,9 @@ async def rank_mcp_servers_by_trust(
active_client = client or get_mcp_trust_scoring_client()
if active_client is None:
return tuple(servers)
return await active_client.rank_servers_by_trust(servers)
return await active_client.rank_servers_by_trust( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
servers
)
async def apply_trust_filter_to_allowed_mcp_servers(
@ -506,4 +523,6 @@ async def apply_trust_filter_to_allowed_mcp_servers(
*,
client: MCPTrustScoringClient | None = None,
) -> tuple[MCPServer, ...]:
return await filter_mcp_servers_by_trust(servers, client=client)
return await filter_mcp_servers_by_trust( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
servers, client=client
)

18
litellm/proxy/_experimental/mcp_server/rest_endpoints.py
View file

@ -383,7 +383,9 @@ if MCP_AVAILABLE:
)
allowed_mcp_servers = list(
await apply_trust_filter_to_allowed_mcp_servers(allowed_mcp_servers)
await apply_trust_filter_to_allowed_mcp_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
allowed_mcp_servers
)
)
return allowed_mcp_servers, canonical_server_id
@ -477,7 +479,9 @@ if MCP_AVAILABLE:
)
return list(
await apply_trust_filter_to_allowed_mcp_servers(allowed_mcp_servers)
await apply_trust_filter_to_allowed_mcp_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
allowed_mcp_servers
)
)
async def _list_tools_for_single_server(
@ -666,12 +670,10 @@ if MCP_AVAILABLE:
)
is not None
]
allowed_server_ids = [
server.server_id
for server in await apply_trust_filter_to_allowed_mcp_servers(
allowed_mcp_server_objects
)
]
trusted_servers = await apply_trust_filter_to_allowed_mcp_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
allowed_mcp_server_objects
)
allowed_server_ids = [server.server_id for server in trusted_servers]
list_tools_result = []
error_message = None

11
litellm/proxy/_experimental/mcp_server/server.py
View file

@ -1717,7 +1717,9 @@ if MCP_AVAILABLE:
)
allowed_mcp_servers = list(
await apply_trust_filter_to_allowed_mcp_servers(allowed_mcp_servers)
await apply_trust_filter_to_allowed_mcp_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
allowed_mcp_servers
)
)
# Pre-fetch OAuth credentials only when at least one server uses OAuth2,
@ -2747,7 +2749,6 @@ if MCP_AVAILABLE:
raw_headers=raw_headers,
litellm_logging_obj=litellm_logging_obj,
host_progress_callback=host_progress_callback,
allowed_mcp_servers=allowed_mcp_servers,
)
# Fall back to local tool registry with original name (legacy support)
@ -2816,7 +2817,9 @@ if MCP_AVAILABLE:
)
allowed_mcp_servers = list(
await apply_trust_filter_to_allowed_mcp_servers(allowed_mcp_servers)
await apply_trust_filter_to_allowed_mcp_servers( # any-ok: awaited coroutine, Send/Recv Any is a typing artifact and the result is fully typed
allowed_mcp_servers
)
)
if not allowed_mcp_servers:
raise HTTPException(
@ -3004,7 +3007,6 @@ if MCP_AVAILABLE:
raw_headers: Optional[Dict[str, str]] = None,
litellm_logging_obj: Optional[Any] = None,
host_progress_callback: Optional[Callable] = None,
allowed_mcp_servers: list[MCPServer] | None = None,
) -> CallToolResult:
"""Handle tool execution for managed server tools"""
# Import here to avoid circular import
@ -3021,7 +3023,6 @@ if MCP_AVAILABLE:
raw_headers=raw_headers,
proxy_logging_obj=proxy_logging_obj,
host_progress_callback=host_progress_callback,
allowed_mcp_servers=allowed_mcp_servers,
)
verbose_logger.debug("CALL TOOL RESULT: %s", call_tool_result)
return call_tool_result

6
litellm/proxy/proxy_server.py
View file

@ -939,7 +939,7 @@ async def proxy_startup_event(app: FastAPI):
)
verbose_proxy_logger.debug("After semantic tool filter initialization")
ProxyStartupEvent._initialize_mcp_trust_scoring(
litellm_settings=_litellm_settings,
litellm_settings=_litellm_settings, # any-ok: proxy config state is untyped (Any), validated downstream by MCPTrustScoringConfig
)
except Exception as e:
verbose_proxy_logger.error(
@ -7608,7 +7608,9 @@ class ProxyStartupEvent:
)
initialize_mcp_trust_scoring_from_config(
litellm_settings.get("mcp_trust_scoring")
litellm_settings.get( # any-ok: proxy config state is untyped (Any), validated downstream by MCPTrustScoringConfig
"mcp_trust_scoring"
)
)
@classmethod