Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/mcp_management_endpoints.py: 56%
1259 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
21. Allow proxy admin to perform create, update, and delete operations on MCP servers in the db.
32. Allows users to view the mcp servers they have access to.
5Endpoints here:
6- GET `/v1/mcp/server` - Returns all of the configured mcp servers in the db filtered by requestor's access
7- GET `/v1/mcp/server/{server_id}` - Returns the the specific mcp server in the db given `server_id` filtered by requestor's access
8- POST `/v1/mcp/server` - Add a new external mcp server.
9- PUT `/v1/mcp/server` - Edits an existing mcp server.
10- DELETE `/v1/mcp/server/{server_id}` - Deletes the mcp server given `server_id`.
11- GET `/v1/mcp/tools - lists all the tools available for a key
12- GET `/v1/mcp/access_groups` - lists all available MCP access groups
13- GET `/v1/mcp/discover` - Returns curated list of well-known MCP servers for discovery UI
14- GET `/v1/mcp/openapi-registry` - Returns well-known OpenAPI APIs with OAuth 2.0 metadata
16"""
18import functools
19import importlib
20import json
21import os
22from collections.abc import Iterable, Mapping, Sequence
23from dataclasses import dataclass
24from datetime import datetime, timedelta, timezone
25from typing import (
26 TYPE_CHECKING,
27 Annotated,
28 Final,
29 Literal,
30 NoReturn,
31 Protocol,
32 cast, # noqa: TID251 # validated JSON values need explicit narrowing
33)
35from fastapi import (
36 APIRouter,
37 Depends,
38 Form,
39 Header,
40 HTTPException,
41 Query,
42 Request,
43 Response,
44 status,
45)
46from fastapi.responses import JSONResponse
47from typing_extensions import ReadOnly, TypedDict
49try:
50 from prisma.errors import RecordNotFoundError, UniqueViolationError
51except ImportError:
52 RecordNotFoundError = Exception
53 UniqueViolationError = Exception
55import litellm
56from litellm._logging import verbose_logger, verbose_proxy_logger
57from litellm._uuid import uuid
58from litellm.constants import LITELLM_PROXY_ADMIN_NAME, MCP_GATEWAY_SESSION_ID_PREFIX_LENGTH
59from litellm.proxy._experimental.mcp_server.utils import (
60 LITELLM_MCP_SERVER_DESCRIPTION,
61 LITELLM_MCP_SERVER_NAME,
62 McpServerPayloadLike,
63 build_env_var_setup_url,
64 collect_env_var_references,
65 get_server_prefix,
66 parse_admin_env_vars,
67)
68from litellm.proxy._experimental.mcp_server.utils import (
69 validate_and_normalize_mcp_server_payload as _base_validate_and_normalize_mcp_server_payload,
70)
71from litellm.proxy.common_utils.encrypt_decrypt_utils import (
72 decrypt_value_helper,
73 encrypt_value_helper,
74)
75from litellm.proxy.management_endpoints.sso.id_jag_assertion_capture import (
76 id_jag_assertion_capture_gap,
77)
78from litellm.proxy.management_helpers.audit_logs import (
79 get_audit_log_changed_by,
80 is_audit_logging_enabled,
81)
82from litellm.repositories.table_repositories import (
83 MCPServerRepository,
84 MCPUserCredentialsRepository,
85)
87router: Final = APIRouter(prefix="/v1/mcp", tags=["mcp"])
89MCP_AVAILABLE: bool = True
91TEMPORARY_MCP_SERVER_TTL_SECONDS: Final = 300
92TEMPORARY_MCP_SERVER_REDIS_KEY_PREFIX: Final = "litellm:mcp:temporary_server"
95class _HasServerId(Protocol):
96 server_id: str
99def does_mcp_server_exist(mcp_server_records: Iterable[_HasServerId], mcp_server_id: str) -> bool:
100 """
101 Check if the mcp server with the given id exists in the iterable of mcp servers.
103 Defined at module level (outside ``if MCP_AVAILABLE``) so it can be imported
104 on Python < 3.10 where the ``mcp`` package is unavailable.
105 """
106 for mcp_server_record in mcp_server_records:
107 if mcp_server_record.server_id == mcp_server_id:
108 return True
109 return False
112DEFAULT_MCP_REGISTRY_VERSION: Final = "1.0.0"
114if TYPE_CHECKING: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 from prisma import models as prisma_models
117 from litellm.proxy.utils import PrismaClient
119try:
120 importlib.import_module("mcp")
121except ImportError as e:
122 verbose_logger.debug("MCP module not found: %s", e)
123 MCP_AVAILABLE = False
125if MCP_AVAILABLE: 125 ↛ exitline 125 didn't exit the module because the condition on line 125 was always true
126 try:
127 from mcp.shared.tool_name_validation import (
128 validate_tool_name, # pyright: ignore[reportAssignmentType]
129 )
130 except ImportError:
131 from pydantic import BaseModel
133 class _ToolNameValidationResult(BaseModel):
134 is_valid: bool = True
135 warnings: list[str] = []
137 def validate_tool_name(name: str) -> _ToolNameValidationResult:
138 return _ToolNameValidationResult()
140 from litellm.proxy._experimental.mcp_server.db import (
141 McpIdentifierConflict,
142 approve_mcp_server,
143 create_draft_mcp_server,
144 create_mcp_server_if_identifier_free,
145 delete_mcp_server,
146 delete_user_credential,
147 delete_user_env_vars,
148 get_all_mcp_servers,
149 get_all_mcp_servers_for_user,
150 get_draft_mcp_server,
151 get_mcp_server,
152 get_mcp_servers,
153 get_mcp_submissions,
154 get_user_env_vars,
155 get_user_env_vars_bulk,
156 get_user_oauth_credential,
157 list_server_user_credentials,
158 list_user_oauth_credentials,
159 mcp_oauth_token_identity,
160 merge_user_env_vars,
161 purge_user_oauth_credentials_for_server,
162 reject_mcp_server,
163 store_user_credential,
164 store_user_oauth_credential,
165 update_mcp_server,
166 )
167 from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
168 _raise_if_not_oauth2,
169 authorize_with_server,
170 client_supplied_redirect_uris,
171 exchange_token_with_server,
172 get_request_base_url,
173 redeem_passthrough_authorization_code,
174 register_client_with_server,
175 resolve_ephemeral_dcr_client,
176 )
177 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
178 global_mcp_server_manager,
179 )
180 from litellm.proxy._experimental.mcp_server.ui_session_utils import (
181 admitted_user_context,
182 build_effective_auth_contexts,
183 can_access_mcp_server,
184 is_ui_session_credential,
185 )
186 from litellm.proxy._types import (
187 LiteLLM_MCPServerTable,
188 LitellmUserRoles,
189 MakeMCPServersPublicRequest,
190 MCPApprovalStatus,
191 MCPOAuthUserCredentialRequest,
192 MCPOAuthUserCredentialStatus,
193 MCPServerUserCredentialListItem,
194 MCPSubmissionsSummary,
195 MCPTransport,
196 MCPUserCredentialListItem,
197 MCPUserCredentialRequest,
198 MCPUserCredentialResponse,
199 MCPUserEnvVarSpec,
200 MCPUserEnvVarsRequest,
201 MCPUserEnvVarsStatus,
202 NewMCPServerRequest,
203 RejectMCPServerRequest,
204 SpecialMCPServerName,
205 UpdateMCPServerRequest,
206 UserAPIKeyAuth,
207 UserMCPManagementMode,
208 is_per_server_oauth_discovery_eligible,
209 )
210 from litellm.proxy.auth.user_api_key_auth import (
211 _user_api_key_auth_builder,
212 user_api_key_auth,
213 )
214 from litellm.proxy.common_utils.http_parsing_utils import (
215 _read_request_body,
216 populate_request_with_path_params,
217 )
218 from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
219 from litellm.proxy.management_endpoints.mcp_connector_import import (
220 ConnectorConversionError,
221 ConvertedConnector,
222 MCPConnectorImportFailure,
223 MCPConnectorImportRequest,
224 MCPConnectorImportResponse,
225 MCPConnectorImportResult,
226 MCPConnectorImportSkipped,
227 convert_connector_entries,
228 )
229 from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
230 from litellm.types.mcp import (
231 MCP_ADMIN_CONFIG_CREDENTIAL_KEYS,
232 MCPAuth,
233 MCPCredentials,
234 MCPGatewaySessionsResponse,
235 MCPGatewaySessionsTerminateResponse,
236 normalize_upstream_header_name,
237 )
238 from litellm.types.mcp_server.mcp_server_manager import MCPServer
240 @dataclass
241 class _TemporaryMCPServerEntry:
242 server: MCPServer
243 expires_at: datetime
245 def _validate_mcp_server_name_fields(payload: McpServerPayloadLike) -> None:
246 candidates: Final[list[tuple[str, str | None]]] = []
248 server_name: Final = getattr(payload, "server_name", None)
249 alias: Final = getattr(payload, "alias", None)
251 if server_name:
252 candidates.append(("server_name", server_name))
253 if alias:
254 candidates.append(("alias", alias))
256 for field_name, value in candidates:
257 if not value: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 continue
260 validation_result = validate_tool_name(value)
261 if validation_result.is_valid: 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 continue
264 error_messages_text = f"Invalid MCP tool prefix '{value}' provided via {field_name}"
265 if validation_result.warnings: 265 ↛ 267line 265 didn't jump to line 267 because the condition on line 265 was always true
266 error_messages_text = error_messages_text + "\n" + "\n".join(validation_result.warnings)
267 raise HTTPException(
268 status_code=status.HTTP_400_BAD_REQUEST,
269 detail={"error": error_messages_text},
270 )
272 def _validate_upstream_token_header(payload: McpServerPayloadLike) -> None:
273 credentials: Final = getattr(payload, "credentials", None)
274 raw: Final = credentials.get("upstream_token_header") if isinstance(credentials, dict) else None
275 if not isinstance(raw, str) or raw == "": 275 ↛ 277line 275 didn't jump to line 277 because the condition on line 275 was always true
276 return
277 if normalize_upstream_header_name(raw) is None:
278 raise HTTPException(
279 status_code=status.HTTP_400_BAD_REQUEST,
280 detail={
281 "error": (
282 f"Invalid upstream_token_header {raw!r}: must be a valid HTTP header name "
283 "(RFC 7230 token, e.g. 'esb-oauth')"
284 )
285 },
286 )
288 def validate_and_normalize_mcp_server_payload(payload: McpServerPayloadLike) -> None:
289 _base_validate_and_normalize_mcp_server_payload(payload)
290 _validate_mcp_server_name_fields(payload)
291 _validate_upstream_token_header(payload)
293 def mcp_identifier_conflict_message(conflict: McpIdentifierConflict) -> str:
294 return (
295 f"An MCP server with {conflict.field} '{conflict.value}' already exists "
296 f"(server_id={conflict.server_id}). "
297 "MCP server names and aliases must be unique, case-insensitive."
298 )
300 def raise_mcp_identifier_conflict(conflict: McpIdentifierConflict) -> NoReturn:
301 raise HTTPException(
302 status_code=status.HTTP_400_BAD_REQUEST,
303 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
304 "error": mcp_identifier_conflict_message(conflict)
305 },
306 )
308 def warn_if_id_jag_server_outruns_sso(server_id: str | None, auth_type: MCPAuth | str | None) -> None:
309 """Registering an ``oauth2_id_jag`` server under an SSO provider that captures no IdP
310 identity assertion is a dead configuration: nothing here fails, and then every ID-JAG call
311 fails for every user with a message that only ever tells them to sign in again. Say it once,
312 at the moment the admin can still act on it."""
313 if auth_type != MCPAuth.oauth2_id_jag: 313 ↛ 315line 313 didn't jump to line 315 because the condition on line 313 was always true
314 return
315 gap = id_jag_assertion_capture_gap()
316 if gap is None:
317 return
318 verbose_proxy_logger.warning(
319 "MCP server %s is registered with auth_type=oauth2_id_jag, but %s.",
320 server_id,
321 gap,
322 )
324 def stamp_omitted_oauth2_flow(payload: NewMCPServerRequest) -> None:
325 """Fallback only: fill in oauth2_flow when an oauth2 create omits it.
327 An explicit oauth2_flow from the caller (the dashboard's flow selector, a REST
328 body, config.yaml) always wins and is never touched. The shape check below runs
329 solely for oauth2 creates that leave the field unset, so those rows still
330 persist a flow instead of relying on read-time inference.
332 The create payload carries the plaintext credentials, so the M2M-vs-interactive
333 decision is reliable here in a way it is not at read time (credentials are
334 encrypted at rest and redacted in responses). The client_credentials shape
335 mirrors the legacy inference in MCPServerManager._resolve_oauth2_flow; every
336 other oauth2 configuration is the authorization_code grant, including
337 delegate_auth_to_upstream, where the client runs that grant upstream.
338 """
339 if payload.auth_type != MCPAuth.oauth2: 339 ↛ 341line 339 didn't jump to line 341 because the condition on line 339 was always true
340 return
341 if payload.oauth2_flow:
342 return
343 credentials: Final = payload.credentials or {}
344 has_m2m_shape: Final = bool(
345 payload.token_url
346 and credentials.get("client_id")
347 and credentials.get("client_secret")
348 and not payload.authorization_url
349 )
350 payload.oauth2_flow = "client_credentials" if has_m2m_shape else "authorization_code"
352 _VALID_MCP_REQUIRED_FIELDS: Final[frozenset] = frozenset(NewMCPServerRequest.model_fields)
354 def _validate_mcp_required_fields(payload: NewMCPServerRequest) -> None:
355 """Validate submission payload against admin-configured mcp_required_fields."""
356 from litellm.proxy.proxy_server import (
357 general_settings as proxy_general_settings,
358 )
360 required_fields: Final[list[str] | None] = proxy_general_settings.get("mcp_required_fields")
361 if not required_fields:
362 return
364 # Fail fast on unknown field names — a typo in the config would silently
365 # block every submission with a confusing "missing fields" error.
366 unknown: Final = [f for f in required_fields if f not in _VALID_MCP_REQUIRED_FIELDS]
367 if unknown:
368 raise HTTPException(
369 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
370 detail={
371 "error": f"mcp_required_fields contains unknown field names: {unknown}. "
372 "Check general_settings.mcp_required_fields in your proxy config."
373 },
374 )
376 # Mirror the UI's compliance checks (MCPStandardsSettings.tsx FIELD_GROUPS):
377 # auth_type requires a real value — "none" is treated as absent.
378 _AUTH_TYPE_SENTINEL: Final = "none"
380 def _field_present(field_name: str) -> bool:
381 value: Final = getattr(payload, field_name, None)
382 if value is None:
383 return False
384 # Treat empty string and empty list as absent (mirrors UI compliance check)
385 if isinstance(value, (str, list)) and not value:
386 return False
387 if field_name == "auth_type" and value == _AUTH_TYPE_SENTINEL:
388 return False
389 return True
391 missing: Final = [f for f in required_fields if not _field_present(f)]
392 if missing:
393 raise HTTPException(
394 status_code=status.HTTP_400_BAD_REQUEST,
395 detail={
396 "error": f"Submission is missing required fields: {missing}. "
397 "Configure required fields via general_settings.mcp_required_fields."
398 },
399 )
401 def _is_public_registry_enabled() -> bool:
402 from litellm.proxy.proxy_server import (
403 general_settings as proxy_general_settings,
404 )
406 return bool(proxy_general_settings.get("enable_mcp_registry"))
408 def _build_registry_remote_url(base_url: str, path: str) -> str:
409 normalized_base: Final = base_url.rstrip("/")
410 normalized_path: Final = path if path.startswith("/") else f"/{path}"
411 return f"{normalized_base}{normalized_path}"
413 def _build_mcp_registry_server_name(server: MCPServer) -> str:
414 if server.alias:
415 return server.alias
416 if server.server_name:
417 return server.server_name
418 return server.server_id
420 class _McpRegistryRemote(TypedDict):
421 type: ReadOnly[str]
422 url: ReadOnly[str]
424 class _McpRegistryEntry(TypedDict):
425 name: ReadOnly[str]
426 title: ReadOnly[str]
427 description: ReadOnly[str]
428 version: ReadOnly[str]
429 remotes: ReadOnly[Sequence[_McpRegistryRemote]]
431 def _build_mcp_registry_entry_for_server(server: MCPServer, base_url: str) -> _McpRegistryEntry:
432 server_name: Final = _build_mcp_registry_server_name(server)
433 title: Final = server_name
434 description: Final = server_name
435 version: Final = DEFAULT_MCP_REGISTRY_VERSION
437 server_prefix: Final = get_server_prefix(server)
438 if not server_prefix:
439 raise ValueError("MCP server prefix is missing")
440 remote_url: Final = _build_registry_remote_url(base_url, f"/{server_prefix}/mcp")
442 return {
443 "name": server_name,
444 "title": title,
445 "description": description,
446 "version": version,
447 "remotes": [
448 {
449 "type": "streamable-http",
450 "url": remote_url,
451 }
452 ],
453 }
455 def _build_builtin_registry_entry(base_url: str) -> _McpRegistryEntry:
456 remote_url: Final = _build_registry_remote_url(base_url, "/mcp")
457 return {
458 "name": LITELLM_MCP_SERVER_NAME,
459 "title": LITELLM_MCP_SERVER_NAME,
460 "description": LITELLM_MCP_SERVER_DESCRIPTION,
461 "version": DEFAULT_MCP_REGISTRY_VERSION,
462 "remotes": [
463 {
464 "type": "streamable-http",
465 "url": remote_url,
466 }
467 ],
468 }
470 _temporary_mcp_servers: Final[dict[str, _TemporaryMCPServerEntry]] = {}
472 def _prune_expired_temporary_mcp_servers() -> None:
473 if not _temporary_mcp_servers:
474 return
476 now: Final = datetime.utcnow()
477 expired_ids = [server_id for server_id, entry in _temporary_mcp_servers.items() if entry.expires_at <= now]
478 for server_id in expired_ids:
479 _temporary_mcp_servers.pop(server_id, None)
481 def _cache_temporary_mcp_server(server: MCPServer, ttl_seconds: int) -> MCPServer:
482 ttl_seconds = max(1, ttl_seconds)
483 _prune_expired_temporary_mcp_servers()
484 expires_at: Final = datetime.utcnow() + timedelta(seconds=ttl_seconds)
485 _temporary_mcp_servers[server.server_id] = _TemporaryMCPServerEntry(
486 server=server,
487 expires_at=expires_at,
488 )
489 return server
491 async def _cache_temporary_mcp_server_in_redis(server: MCPServer, ttl_seconds: int) -> None:
492 """
493 Best-effort write-through to Redis so temporary MCP OAuth sessions are
494 shared across proxy instances. Keep local in-memory cache as fallback.
495 """
496 if litellm.cache is None or not hasattr(litellm.cache, "cache"): 496 ↛ 497line 496 didn't jump to line 497 because the condition on line 496 was never true
497 return
498 cache_backend: Final = getattr(litellm.cache, "cache", None)
499 if cache_backend is None or not hasattr(cache_backend, "async_set_cache"): 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 return
502 payload: Final[dict[str, object]] = server.model_dump(mode="json")
503 payload_json: Final = json.dumps(payload)
504 try:
505 encrypted_payload: Final = encrypt_value_helper(payload_json)
506 except Exception as e:
507 verbose_proxy_logger.debug("Failed to encrypt temporary MCP server payload for Redis cache: %s", e)
508 return
510 if not isinstance(encrypted_payload, str): 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true
511 verbose_proxy_logger.debug("Encrypted temporary MCP payload is not a string; skipping Redis cache write")
512 return
514 try:
515 await cache_backend.async_set_cache(
516 key=f"{TEMPORARY_MCP_SERVER_REDIS_KEY_PREFIX}:{server.server_id}",
517 value=encrypted_payload,
518 ttl=max(1, ttl_seconds),
519 )
520 except Exception as e:
521 verbose_proxy_logger.debug("Failed to write temporary MCP server to Redis cache: %s", e)
523 async def _get_temporary_mcp_server_from_redis(
524 server_id: str,
525 ) -> MCPServer | None:
526 """
527 Best-effort read from Redis shared cache. Returns None on miss/errors.
529 Values must be encrypted strings (same contract as _cache_temporary_mcp_server_in_redis);
530 legacy plaintext dict payloads are rejected.
531 """
532 if litellm.cache is None or not hasattr(litellm.cache, "cache"):
533 return None
534 cache_backend: Final = getattr(litellm.cache, "cache", None)
535 if cache_backend is None or not hasattr(cache_backend, "async_get_cache"):
536 return None
538 try:
539 cached_server: Final = await cache_backend.async_get_cache(
540 key=f"{TEMPORARY_MCP_SERVER_REDIS_KEY_PREFIX}:{server_id}"
541 )
542 except Exception as e:
543 verbose_proxy_logger.debug("Failed reading temporary MCP server from Redis cache: %s", e)
544 return None
546 if not isinstance(cached_server, str):
547 verbose_proxy_logger.debug(
548 "Temporary MCP Redis cache value must be an encrypted string; rejecting non-string payload"
549 )
550 return None
552 decrypted_json: Final = decrypt_value_helper(
553 value=cached_server,
554 key="temporary_mcp_server",
555 exception_type="debug",
556 )
557 if decrypted_json is None:
558 return None
559 try:
560 loaded: Final = json.loads(decrypted_json)
561 except Exception as e:
562 verbose_proxy_logger.debug("Invalid decrypted temporary MCP payload in Redis cache: %s", e)
563 return None
564 if not isinstance(loaded, dict):
565 return None
566 payload_dict: Final[dict[str, object]] = loaded
568 try:
569 return MCPServer.model_validate(payload_dict)
570 except Exception as e:
571 verbose_proxy_logger.debug("Invalid temporary MCP server payload in Redis cache: %s", e)
572 return None
574 def _get_prisma_client_or_none() -> "PrismaClient | None":
575 """Non-throwing counterpart to ``get_prisma_client_or_throw`` for paths that degrade
576 gracefully: a proxy configured without a database keeps the in-memory OAuth session."""
577 from litellm.proxy.proxy_server import prisma_client
579 return prisma_client
581 async def _persist_draft_mcp_server(
582 payload: NewMCPServerRequest,
583 server_id: str,
584 created_by: str,
585 ) -> None:
586 """Write the draft row that makes the OAuth session resolvable from any worker.
588 A failure here is raised, not swallowed: without the shared row the flow degrades to
589 the per-process cache and fails intermittently, which is the defect being fixed.
590 """
591 prisma_client: Final = _get_prisma_client_or_none()
592 if prisma_client is None: 592 ↛ 593line 592 didn't jump to line 593 because the condition on line 592 was never true
593 return
594 await create_draft_mcp_server(
595 prisma_client,
596 payload,
597 created_by,
598 ttl_seconds=TEMPORARY_MCP_SERVER_TTL_SECONDS,
599 server_id=server_id,
600 )
602 async def _get_draft_mcp_server_as_mcp_server(server_id: str) -> MCPServer | None:
603 """Resolve a database-backed draft, which is the only lookup that works across workers."""
604 prisma_client: Final = _get_prisma_client_or_none()
605 if prisma_client is None:
606 return None
607 draft: Final = await get_draft_mcp_server(
608 prisma_client, server_id, ttl_seconds=TEMPORARY_MCP_SERVER_TTL_SECONDS
609 )
610 if draft is None:
611 return None
612 return await global_mcp_server_manager.build_mcp_server_from_table(draft)
614 async def get_cached_temporary_mcp_server(
615 server_id: str,
616 ) -> MCPServer | None:
617 _prune_expired_temporary_mcp_servers()
618 entry: Final = _temporary_mcp_servers.get(server_id)
619 if entry is not None:
620 return entry.server
622 # A miss here means either an expired session or, on a multi-worker or multi-replica
623 # proxy, that a different process served /session. The draft row is shared, so it
624 # resolves the second case; the in-memory hit above still serves single-process
625 # deployments with no database configured.
626 draft_server: Final = await _get_draft_mcp_server_as_mcp_server(server_id)
627 if draft_server is not None:
628 return draft_server
630 redis_server: Final = await _get_temporary_mcp_server_from_redis(server_id)
631 if redis_server is None:
632 return None
633 # Intentionally avoid repopulating local cache from Redis to prevent
634 # extending effective lifetime beyond the remaining Redis TTL.
635 return redis_server
637 def _redact_mcp_credentials(
638 mcp_server: LiteLLM_MCPServerTable,
639 ) -> LiteLLM_MCPServerTable:
640 """Return a copy with secret credentials removed, keeping only non-secret admin config so the
641 admin form can show and clear it. Non-admin and virtual-key views strip the whole blob."""
643 try:
644 redacted_server = mcp_server.model_copy(deep=True)
645 except AttributeError:
646 redacted_server = mcp_server.copy(deep=True)
648 if hasattr(redacted_server, "credentials"): 648 ↛ 651line 648 didn't jump to line 651 because the condition on line 648 was always true
649 setattr(redacted_server, "credentials", _preserved_admin_config_credentials(redacted_server.credentials))
651 return redacted_server
653 def _preserved_admin_config_credentials(
654 credentials: "MCPCredentials | str | None",
655 ) -> "dict[str, str | list[str]] | None": # mutable-ok: API response payload
656 """Keep non-secret admin-config keys and scopes, which are stored unencrypted so they lift out
657 as plaintext; every secret and minted-token key is dropped.
659 Total over every stored shape: a dict is read directly, a JSON-object string is parsed, and
660 anything else (a malformed or non-object JSON string, a scalar, ``None``) falls back to full
661 redaction rather than raising, because this runs on every admin list and get and one bad row
662 must not fail them all."""
663 parsed: object = credentials
664 if isinstance(credentials, str): 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true
665 try:
666 parsed = cast(object, json.loads(credentials)) # cast-ok: JSON parse result is validated below
667 except (ValueError, TypeError):
668 return None
669 if not isinstance(parsed, dict):
670 return None
671 parsed_credentials: Final = cast(Mapping[str, object], parsed) # cast-ok: dict shape validated above
672 scopes: Final[object] = parsed_credentials.get("scopes")
673 scopes_as_objects: Final = (
674 cast(Sequence[object], scopes) # cast-ok: list shape validated above
675 if isinstance(scopes, list)
676 else ()
677 )
678 preserved_scopes: Final = (
679 {"scopes": cast(list[str], scopes_as_objects)} # cast-ok: every scope is validated below
680 if scopes_as_objects and all(isinstance(scope, str) and scope for scope in scopes_as_objects)
681 else {}
682 )
683 preserved: Final = { # mutable-ok: API response payload
684 **{
685 key: value
686 for key in MCP_ADMIN_CONFIG_CREDENTIAL_KEYS
687 if isinstance((value := parsed_credentials.get(key)), str) and value
688 },
689 **preserved_scopes,
690 }
691 return preserved or None
693 def _redact_mcp_credentials_list(
694 mcp_servers: Iterable[LiteLLM_MCPServerTable],
695 ) -> list[LiteLLM_MCPServerTable]:
696 return [_redact_mcp_credentials(server) for server in mcp_servers]
698 def _user_is_full_admin(user_api_key_dict: UserAPIKeyAuth) -> bool:
699 """True only for ``PROXY_ADMIN``; ``PROXY_ADMIN_VIEW_ONLY`` returns False.
701 Global env var secrets pre-fill the admin edit form, so a full admin
702 must see them, but a read-only admin gets the same redacted view as
703 any other non-managing caller.
704 """
705 return user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
707 def _resolve_credential_target_user_id(user_api_key_dict: UserAPIKeyAuth, requested_user_id: str | None) -> str:
708 """The user whose stored MCP credential a request acts on.
710 Defaults to the caller. Naming another user is a revocation and needs
711 ``PROXY_ADMIN``; a read-only admin or a regular user gets 403.
712 """
713 caller_user_id: Final = user_api_key_dict.user_id or ""
714 if requested_user_id is not None and requested_user_id != caller_user_id:
715 if not _user_is_full_admin(user_api_key_dict): 715 ↛ 716line 715 didn't jump to line 716 because the condition on line 715 was never true
716 raise HTTPException(
717 status_code=status.HTTP_403_FORBIDDEN,
718 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
719 "error": "Proxy admin access required to revoke another user's MCP credential.",
720 },
721 )
722 return requested_user_id
723 if not caller_user_id: 723 ↛ 724line 723 didn't jump to line 724 because the condition on line 723 was never true
724 raise HTTPException(
725 status_code=status.HTTP_400_BAD_REQUEST,
726 detail={"error": "User ID not found in token"},
727 )
728 return caller_user_id
730 def _is_restricted_virtual_key_request(user_api_key_dict: UserAPIKeyAuth) -> bool:
731 """Best-effort detection for route-restricted virtual keys.
733 We treat a requestor as a "restricted" virtual key if `allowed_routes`
734 is a non-empty list. This matches the auth gate that blocks routes with
735 the error: "Virtual key is not allowed to call this route...".
736 """
738 allowed_routes: Final = getattr(user_api_key_dict, "allowed_routes", None)
739 return isinstance(allowed_routes, list) and len(allowed_routes) > 0
741 def _sanitize_mcp_server_for_non_admin(
742 mcp_server: LiteLLM_MCPServerTable,
743 ) -> LiteLLM_MCPServerTable:
744 """Strip credential-bearing fields for non-admin viewers.
746 Non-admin users may legitimately need to discover MCP servers
747 their team has access to (so they can pick one in the UI), but
748 they must never see fields that can carry bearer tokens or
749 upstream API keys. ``_redact_mcp_credentials`` already clears
750 the explicit ``credentials`` field; this layers on top to catch
751 the URL+headers+env vectors that the virtual-key sanitizer also
752 strips. Reset values match each field's declared default on
753 ``LiteLLM_MCPServerTable`` (``None`` for Optional fields,
754 ``[]``/``{}`` for required list/dict fields).
755 """
756 sanitized: Final = _redact_mcp_credentials(mcp_server)
757 sanitized.credentials = None
758 # URL is the highest-impact vector: many MCP integrations embed
759 # the upstream API key directly in the path. spec_path can carry
760 # similar tokens in the OpenAPI spec URL.
761 sanitized.url = None
762 sanitized.spec_path = None
763 sanitized.static_headers = None
764 sanitized.extra_headers = []
765 sanitized.env = {}
766 sanitized.command = None
767 sanitized.args = []
768 sanitized.issuer = None
769 sanitized.authorization_url = None
770 sanitized.token_url = None
771 sanitized.registration_url = None
772 sanitized.token_exchange_endpoint = None
773 sanitized.audience = None
774 sanitized.subject_token_type = None
775 sanitized.token_exchange_profile = None
776 # Drop env vars entirely rather than only blanking global values: the
777 # names alone (DB_PASSWORD, GITHUB_API_KEY, ...) leak what secrets the
778 # admin configured. Non-admins get the per-user vars they must fill in
779 # from the dedicated /user-env-vars/status endpoint instead.
780 sanitized.env_vars = None
781 return sanitized
783 def _sanitize_mcp_server_list_for_non_admin(
784 mcp_servers: Iterable[LiteLLM_MCPServerTable],
785 ) -> list[LiteLLM_MCPServerTable]:
786 return [_sanitize_mcp_server_for_non_admin(s) for s in mcp_servers]
788 def _sanitize_mcp_server_for_virtual_key(
789 mcp_server: LiteLLM_MCPServerTable,
790 ) -> LiteLLM_MCPServerTable:
791 """Return a minimally sufficient MCP server view for virtual keys.
793 Security model:
794 - Virtual keys should be able to *discover* accessible servers.
795 - They should NOT receive sensitive configuration details like upstream
796 URLs, env vars, headers, commands/args, access-group names, or
797 credentials.
798 """
800 sanitized: Final = _redact_mcp_credentials(mcp_server)
801 sanitized.credentials = None
803 # Remove potentially sensitive config + identity fields.
804 sanitized.url = None
805 sanitized.static_headers = None
806 sanitized.env = {}
807 sanitized.command = None
808 sanitized.args = []
809 sanitized.extra_headers = []
810 sanitized.allowed_tools = []
811 sanitized.mcp_access_groups = []
812 sanitized.teams = []
813 sanitized.env_vars = None
815 sanitized.issuer = None
816 sanitized.authorization_url = None
817 sanitized.token_url = None
818 sanitized.registration_url = None
819 sanitized.token_exchange_endpoint = None
820 sanitized.audience = None
821 sanitized.subject_token_type = None
822 sanitized.token_exchange_profile = None
824 sanitized.health_check_error = None
825 sanitized.last_health_check = None
827 sanitized.created_by = None
828 sanitized.updated_by = None
829 sanitized.created_at = None
830 sanitized.updated_at = None
832 # `mcp_info` is arbitrary metadata; keep only an explicit safe subset.
833 is_public = False
834 if isinstance(sanitized.mcp_info, dict):
835 is_public = bool(sanitized.mcp_info.get("is_public"))
836 sanitized.mcp_info = {"is_public": True} if is_public else None
838 return sanitized
840 def _sanitize_mcp_server_list_for_virtual_key(
841 mcp_servers: Iterable[LiteLLM_MCPServerTable],
842 ) -> list[LiteLLM_MCPServerTable]:
843 return [_sanitize_mcp_server_for_virtual_key(server) for server in mcp_servers]
845 # (server attribute, credentials key) a session server inherits from the server it derives from.
846 # Declared as a table rather than a chain of ifs, which is how upstream_resource was missed.
847 _INHERITED_CREDENTIAL_FIELDS: Final[tuple[tuple[str, str], ...]] = (
848 ("authentication_token", "auth_value"),
849 ("client_id", "client_id"),
850 ("client_secret", "client_secret"),
851 ("scopes", "scopes"),
852 ("aws_access_key_id", "aws_access_key_id"),
853 ("aws_secret_access_key", "aws_secret_access_key"),
854 ("aws_session_token", "aws_session_token"),
855 ("aws_region_name", "aws_region_name"),
856 ("aws_service_name", "aws_service_name"),
857 ("upstream_resource", "upstream_resource"),
858 ("upstream_token_header", "upstream_token_header"),
859 )
861 def _has_non_admin_config_credentials(credentials: "MCPCredentials | None") -> bool:
862 """Did the caller supply an actual credential? Admin config rides in the same blob but is not
863 one, so a form that round-trips it must not read as "credentials supplied"."""
864 if not credentials: 864 ↛ 866line 864 didn't jump to line 866 because the condition on line 864 was always true
865 return False
866 as_dict: Final[dict[str, object]] = dict(credentials)
867 return any(
868 value for key, value in as_dict.items() if key not in MCP_ADMIN_CONFIG_CREDENTIAL_KEYS and key != "scopes"
869 )
871 def _inherit_credentials_from_existing_server(
872 payload: NewMCPServerRequest,
873 ) -> NewMCPServerRequest:
874 if not payload.server_id or _has_non_admin_config_credentials(payload.credentials):
875 return payload
877 existing_server: Final = global_mcp_server_manager.get_mcp_server_by_id(payload.server_id)
878 if existing_server is None:
879 return payload
881 inherited_credentials: dict[str, object] = {
882 credential_key: value
883 for server_attr, credential_key in _INHERITED_CREDENTIAL_FIELDS
884 if (value := getattr(existing_server, server_attr, None))
885 }
886 # The gate above guarantees anything still supplied is admin config, which the admin just
887 # typed, so it wins over the stored value.
888 inherited_credentials = {**inherited_credentials, **dict(payload.credentials or {})}
890 if not inherited_credentials: 890 ↛ 893line 890 didn't jump to line 893 because the condition on line 890 was always true
891 return payload
893 try:
894 return payload.model_copy(update={"credentials": inherited_credentials})
895 except AttributeError:
896 pass
898 payload_dict: dict[str, object]
899 try:
900 payload_dict = payload.model_dump()
901 except AttributeError:
902 payload_dict = payload.dict()
903 payload_dict["credentials"] = inherited_credentials
904 return NewMCPServerRequest.model_validate(payload_dict)
906 async def _resolve_session_server_id(payload: NewMCPServerRequest) -> str:
907 """Decide the id an OAuth session runs under.
909 A caller-supplied id is honoured only when it names a server that really exists, which is
910 the edit form re-authorizing a saved server against its own id. Anything else gets a fresh
911 id, so two concurrent sessions can never land on one id and silently adopt each other's
912 URL or client credentials. Without a database there is nothing shared to collide over, so
913 the supplied id is kept and behaviour is unchanged.
914 """
915 supplied: Final = payload.server_id
916 if not supplied:
917 return str(uuid.uuid4())
918 if global_mcp_server_manager.get_mcp_server_by_id(supplied) is not None:
919 return supplied
920 prisma_client: Final = _get_prisma_client_or_none()
921 if prisma_client is None: 921 ↛ 922line 921 didn't jump to line 922 because the condition on line 921 was never true
922 return supplied
923 # A draft is another session's row, not a saved server, so re-supplying an id this
924 # endpoint previously handed back must not let a later session adopt its configuration.
925 existing: Final = await get_mcp_server(prisma_client, supplied)
926 if existing is None or existing.approval_status == MCPApprovalStatus.draft: 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 return str(uuid.uuid4())
928 return supplied
930 def _build_temporary_mcp_server_record(
931 payload: NewMCPServerRequest,
932 created_by: str | None,
933 server_id: str,
934 ) -> LiteLLM_MCPServerTable:
935 now: Final = datetime.utcnow()
936 server_name: Final = payload.server_name or payload.alias or server_id
937 return LiteLLM_MCPServerTable(
938 server_id=server_id,
939 server_name=server_name,
940 alias=payload.alias,
941 description=payload.description,
942 url=payload.url,
943 transport=payload.transport,
944 auth_type=payload.auth_type,
945 credentials=payload.credentials,
946 created_at=now,
947 updated_at=now,
948 created_by=created_by,
949 updated_by=created_by,
950 teams=[],
951 mcp_access_groups=payload.mcp_access_groups,
952 allowed_tools=payload.allowed_tools or [],
953 extra_headers=payload.extra_headers or [],
954 mcp_info=payload.mcp_info,
955 static_headers=payload.static_headers,
956 command=payload.command,
957 args=payload.args,
958 env=payload.env,
959 issuer=payload.issuer,
960 authorization_url=payload.authorization_url,
961 token_url=payload.token_url,
962 registration_url=payload.registration_url,
963 allow_all_keys=payload.allow_all_keys,
964 available_on_public_internet=payload.available_on_public_internet,
965 timeout=payload.timeout,
966 max_concurrent_requests=payload.max_concurrent_requests,
967 )
969 def get_prisma_client_or_throw(message: str):
970 from litellm.proxy.proxy_server import prisma_client
972 if prisma_client is None: 972 ↛ 973line 972 didn't jump to line 973 because the condition on line 972 was never true
973 raise HTTPException(
974 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
975 detail={"error": message},
976 )
977 return prisma_client
979 # Router to fetch all MCP tools available for the current key
981 @router.get(
982 "/tools",
983 tags=["mcp"],
984 dependencies=[Depends(user_api_key_auth)],
985 )
986 async def get_mcp_tools(
987 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
988 ):
989 """
990 Get all MCP tools available for the current key, including those from access groups
991 """
992 from litellm.proxy._experimental.mcp_server.server import _list_mcp_tools
994 listing: Final = await _list_mcp_tools(
995 user_api_key_auth=user_api_key_dict,
996 mcp_auth_header=None,
997 mcp_servers=None,
998 mcp_server_auth_headers=None,
999 )
1000 tools: Final = listing.tools
1001 dumped_tools: Final = [tool.model_dump(by_alias=True) for tool in tools]
1003 return {"tools": dumped_tools}
1005 @router.get(
1006 "/access_groups",
1007 tags=["mcp"],
1008 dependencies=[Depends(user_api_key_auth)],
1009 )
1010 async def get_mcp_access_groups(
1011 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1012 ):
1013 """
1014 Get all available MCP access groups from the database AND config
1015 """
1016 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
1017 global_mcp_server_manager,
1018 )
1019 from litellm.proxy.proxy_server import prisma_client
1021 access_groups: Final = set()
1023 # Get from config-loaded servers
1024 for server in global_mcp_server_manager.config_mcp_servers.values(): 1024 ↛ 1025line 1024 didn't jump to line 1025 because the loop on line 1024 never started
1025 if server.access_groups:
1026 access_groups.update(server.access_groups)
1028 # Get from DB
1029 if prisma_client is not None: 1029 ↛ 1041line 1029 didn't jump to line 1041 because the condition on line 1029 was always true
1030 try:
1031 mcp_servers: Final[Sequence[prisma_models.LiteLLM_MCPServerTable]] = await MCPServerRepository(
1032 prisma_client
1033 ).table.find_many()
1034 for server in mcp_servers:
1035 if hasattr(server, "mcp_access_groups") and server.mcp_access_groups:
1036 access_groups.update(server.mcp_access_groups)
1037 except Exception as e:
1038 verbose_proxy_logger.debug("Error getting MCP access groups: %s", e)
1040 # Convert to sorted list
1041 access_groups_list: Final = sorted(list(access_groups))
1042 return {"access_groups": access_groups_list}
1044 @router.get(
1045 "/network/client-ip",
1046 tags=["mcp"],
1047 dependencies=[Depends(user_api_key_auth)],
1048 description="Returns the caller's IP address as seen by the proxy.",
1049 )
1050 async def get_client_ip(request: Request):
1051 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
1053 client_ip: Final = IPAddressUtils.get_mcp_client_ip(request)
1054 return {"ip": client_ip}
1056 @router.get(
1057 "/registry.json",
1058 tags=["mcp"],
1059 description="MCP registry endpoint. Spec: https://github.com/modelcontextprotocol/registry",
1060 )
1061 async def get_mcp_registry(request: Request):
1062 if not _is_public_registry_enabled(): 1062 ↛ 1068line 1062 didn't jump to line 1068 because the condition on line 1062 was always true
1063 raise HTTPException(
1064 status_code=status.HTTP_404_NOT_FOUND,
1065 detail="MCP registry is not enabled",
1066 )
1068 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
1070 client_ip: Final = IPAddressUtils.get_mcp_client_ip(request)
1072 verbose_proxy_logger.debug("MCP registry request from IP=%s", client_ip)
1074 base_url: Final = get_request_base_url(request)
1075 registry_servers: Final[list[dict[str, _McpRegistryEntry]]] = []
1076 registry_servers.append({"server": _build_builtin_registry_entry(base_url)})
1078 # Centralized IP-based filtering: external callers only see public servers
1079 registered_servers: Final = list(global_mcp_server_manager.get_filtered_registry(client_ip).values())
1081 registered_servers.sort(key=_build_mcp_registry_server_name)
1083 for server in registered_servers:
1084 try:
1085 entry = _build_mcp_registry_entry_for_server(server, base_url)
1086 except Exception as e:
1087 verbose_proxy_logger.debug(
1088 "Skipping MCP server %s in registry: %s", getattr(server, "server_id", "unknown"), e
1089 )
1090 continue
1091 registry_servers.append({"server": entry})
1093 return {"servers": registry_servers}
1095 ## FastAPI Routes
1096 def _mcp_server_display_order(server: LiteLLM_MCPServerTable) -> tuple[str, str]:
1097 return ((server.server_name or server.alias or server.server_id).lower(), server.server_id)
1099 def _get_user_mcp_management_mode() -> UserMCPManagementMode:
1100 from litellm.proxy.proxy_server import (
1101 general_settings as proxy_general_settings,
1102 )
1104 mode: Final = proxy_general_settings.get("user_mcp_management_mode")
1105 if mode == "view_all": 1105 ↛ 1106line 1105 didn't jump to line 1106 because the condition on line 1105 was never true
1106 return "view_all"
1107 return "restricted"
1109 async def _get_team_scoped_mcp_server_list(
1110 team_id: str,
1111 ) -> list[LiteLLM_MCPServerTable]:
1112 """
1113 Return MCP servers scoped to a team: team's allowed servers + allow_all_keys servers.
1114 Used by the Create Key UI to populate the MCP server dropdown.
1115 """
1116 from litellm.proxy.auth.auth_checks import get_team_object
1117 from litellm.proxy.management_helpers.object_permission_utils import (
1118 _get_allow_all_keys_server_ids,
1119 _get_team_allowed_mcp_servers,
1120 )
1121 from litellm.proxy.proxy_server import prisma_client, user_api_key_cache
1123 team_obj: Final = await get_team_object(
1124 team_id=team_id,
1125 prisma_client=prisma_client,
1126 user_api_key_cache=user_api_key_cache,
1127 check_db_only=True,
1128 )
1130 team_server_ids: Final = await _get_team_allowed_mcp_servers(team_obj)
1131 allow_all_server_ids: Final = _get_allow_all_keys_server_ids()
1132 all_allowed_ids: Final = team_server_ids | allow_all_server_ids
1134 if not all_allowed_ids: 1134 ↛ 1138line 1134 didn't jump to line 1138 because the condition on line 1134 was always true
1135 return []
1137 # Collect servers from registry
1138 servers: Final[list[LiteLLM_MCPServerTable]] = []
1139 for server_id in all_allowed_ids:
1140 server = global_mcp_server_manager.get_mcp_server_by_id(server_id)
1141 if server is not None:
1142 mcp_server_table = global_mcp_server_manager._build_mcp_server_table(server)
1143 servers.append(mcp_server_table)
1145 return _redact_mcp_credentials_list(servers)
1147 async def _resolve_accessible_mcp_servers(
1148 user_api_key_dict: UserAPIKeyAuth,
1149 ) -> list[LiteLLM_MCPServerTable]:
1150 """The server set the dashboard grid shows (GET /v1/mcp/server, no team
1151 filter), returned unredacted. Callers that surface this to a client must
1152 apply their own redaction; the per-user env-var status endpoint relies on
1153 the raw env_vars and only ever returns is_set booleans, never secrets.
1155 Sharing this resolution keeps the red "missing user fields" card status
1156 aligned with the cards actually rendered: an admin in view_all mode sees
1157 every server even when their key carries no per-server MCP grant.
1158 """
1159 if _get_user_mcp_management_mode() == "view_all" and not _is_restricted_virtual_key_request(user_api_key_dict): 1159 ↛ 1160line 1159 didn't jump to line 1160 because the condition on line 1159 was never true
1160 return await global_mcp_server_manager.get_all_mcp_servers_unfiltered()
1162 aggregated: Final[dict[str, LiteLLM_MCPServerTable]] = {}
1163 for auth_context in await build_effective_auth_contexts(user_api_key_dict):
1164 for server in await global_mcp_server_manager.get_all_allowed_mcp_servers(user_api_key_auth=auth_context):
1165 aggregated.setdefault(server.server_id, server)
1166 return list(aggregated.values())
1168 async def _connected_app_reachable_server_ids(user_api_key_dict: UserAPIKeyAuth) -> frozenset[str]:
1169 """Server ids a connected app authorized by this dashboard user is served on the aggregate
1170 MCP endpoint, resolved through the one owner of the admitted subject so the page and the
1171 session cannot drift. Empty when that identity cannot be built, which is the true answer:
1172 the same user cannot open a gateway session either."""
1173 admitted: Final = await admitted_user_context(user_api_key_dict)
1174 if admitted is None:
1175 return frozenset()
1176 return frozenset(await global_mcp_server_manager.get_allowed_mcp_servers(admitted))
1178 @router.get(
1179 "/server",
1180 description="Returns the mcp server list with associated teams",
1181 dependencies=[Depends(user_api_key_auth)],
1182 response_model=list[LiteLLM_MCPServerTable],
1183 )
1184 async def fetch_all_mcp_servers(
1185 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1186 team_id: str | None = Query(
1187 None,
1188 description="Filter MCP servers by team scope. When provided, returns only "
1189 "servers the team has access to plus globally available (allow_all_keys) servers. "
1190 "Used by the Create Key UI to show team-scoped MCP servers.",
1191 ),
1192 connected_app_view: bool = Query(
1193 False,
1194 description="Annotate each returned server with connected_app_reachable: whether a "
1195 "connected app authorized by the calling user (a gateway OAuth session) is served "
1196 "this server on the aggregate MCP endpoint.",
1197 ),
1198 ):
1199 """
1200 Get all of the configured mcp servers for the user in the db with their associated teams
1201 ```
1202 curl --location 'http://localhost:4000/v1/mcp/server' \
1203 --header 'Authorization: Bearer your_api_key_here'
1205 # Filter by team scope (for Create Key UI)
1206 curl --location 'http://localhost:4000/v1/mcp/server?team_id=team-123' \
1207 --header 'Authorization: Bearer your_api_key_here'
1208 ```
1209 """
1211 # If team_id is provided, return team-scoped servers + allow_all_keys servers
1212 is_restricted_virtual_key: Final = _is_restricted_virtual_key_request(user_api_key_dict)
1213 if team_id is not None and isinstance(team_id, str) and team_id.strip():
1214 # Restricted virtual keys must not use the team_id filter to
1215 # bypass their own access limitations.
1216 if is_restricted_virtual_key: 1216 ↛ 1217line 1216 didn't jump to line 1217 because the condition on line 1216 was never true
1217 raise HTTPException(
1218 status_code=403,
1219 detail="Restricted virtual keys cannot query team-scoped MCP servers.",
1220 )
1222 # Only proxy admins may query another team's MCP servers.
1223 # Non-admins must belong to the requested team.
1224 sanitized_team_id: Final = team_id.strip()
1225 is_admin: Final = _user_has_admin_view(user_api_key_dict)
1226 if not is_admin: 1226 ↛ 1227line 1226 didn't jump to line 1227 because the condition on line 1226 was never true
1227 from litellm.proxy.auth.auth_checks import get_team_object
1228 from litellm.proxy.proxy_server import (
1229 prisma_client,
1230 user_api_key_cache,
1231 )
1233 team_obj: Final = await get_team_object(
1234 team_id=sanitized_team_id,
1235 prisma_client=prisma_client,
1236 user_api_key_cache=user_api_key_cache,
1237 check_db_only=True,
1238 )
1239 user_in_team: Final = any(
1240 m.user_id is not None and m.user_id == user_api_key_dict.user_id
1241 for m in team_obj.members_with_roles
1242 )
1243 if not user_in_team:
1244 raise HTTPException(
1245 status_code=403,
1246 detail="You do not have permission to view MCP servers for this team.",
1247 )
1249 redacted_mcp_servers = sorted(
1250 await _get_team_scoped_mcp_server_list(sanitized_team_id), key=_mcp_server_display_order
1251 )
1252 else:
1253 servers: Final = await _resolve_accessible_mcp_servers(user_api_key_dict)
1254 redacted_mcp_servers = sorted(_redact_mcp_credentials_list(servers), key=_mcp_server_display_order)
1256 if connected_app_view is True and is_ui_session_credential(user_api_key_dict): 1256 ↛ 1257line 1256 didn't jump to line 1257 because the condition on line 1256 was never true
1257 reachable_ids: Final = await _connected_app_reachable_server_ids(user_api_key_dict)
1258 for server in redacted_mcp_servers:
1259 server.connected_app_reachable = server.server_id in reachable_ids
1261 # augment the mcp servers with public status
1262 if litellm.public_mcp_servers is not None: 1262 ↛ 1270line 1262 didn't jump to line 1270 because the condition on line 1262 was always true
1263 for server in redacted_mcp_servers:
1264 if server.server_id in litellm.public_mcp_servers: 1264 ↛ 1265line 1264 didn't jump to line 1265 because the condition on line 1264 was never true
1265 if server.mcp_info is None:
1266 server.mcp_info = {}
1267 server.mcp_info["is_public"] = True
1269 # Annotate has_user_credential for BYOK servers (single batched query)
1270 from litellm.proxy.proxy_server import prisma_client as _byok_prisma_client
1272 user_id: Final = user_api_key_dict.user_id or ""
1273 if user_id and _byok_prisma_client is not None: 1273 ↛ 1287line 1273 didn't jump to line 1287 because the condition on line 1273 was always true
1274 byok_server_ids: Final = [s.server_id for s in redacted_mcp_servers if getattr(s, "is_byok", False)]
1275 if byok_server_ids:
1276 cred_rows: Final[
1277 Sequence[prisma_models.LiteLLM_MCPUserCredentials]
1278 ] = await MCPUserCredentialsRepository(_byok_prisma_client).table.find_many(
1279 where={"user_id": user_id, "server_id": {"in": byok_server_ids}}
1280 )
1281 cred_set: Final = {r.server_id for r in cred_rows}
1282 for server in redacted_mcp_servers:
1283 if getattr(server, "is_byok", False):
1284 server.has_user_credential = server.server_id in cred_set
1286 # Virtual keys only get a sanitized discovery view.
1287 if is_restricted_virtual_key: 1287 ↛ 1288line 1287 didn't jump to line 1288 because the condition on line 1287 was never true
1288 return _sanitize_mcp_server_list_for_virtual_key(redacted_mcp_servers)
1290 # only a full PROXY_ADMIN sees credential-bearing fields; everyone else
1291 # goes through the non-admin sanitizer
1292 if not _user_is_full_admin(user_api_key_dict): 1292 ↛ 1293line 1292 didn't jump to line 1293 because the condition on line 1292 was never true
1293 return _sanitize_mcp_server_list_for_non_admin(redacted_mcp_servers)
1295 return redacted_mcp_servers
1297 @router.get(
1298 "/server/health",
1299 description="Health check for MCP servers",
1300 dependencies=[Depends(user_api_key_auth)],
1301 )
1302 async def health_check_servers(
1303 server_ids: list[str] | None = Query(
1304 None,
1305 description="Server IDs to check. If not provided, checks all accessible servers.",
1306 ),
1307 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1308 ):
1309 """
1310 Perform health checks on one or more MCP servers.
1312 Parameters:
1313 - server_ids: Optional list of server IDs. If not provided, checks all accessible servers.
1315 Returns:
1316 - Health check results for requested servers
1318 ```
1319 # Check all accessible servers
1320 curl --location 'http://localhost:4000/v1/mcp/server/health' \
1321 --header 'Authorization: Bearer your_api_key_here'
1323 # Check specific servers
1324 curl --location 'http://localhost:4000/v1/mcp/server/health?server_ids=server-1&server_ids=server-2' \
1325 --header 'Authorization: Bearer your_api_key_here'
1326 ```
1327 """
1328 user_mcp_management_mode: Final = _get_user_mcp_management_mode()
1330 if user_mcp_management_mode == "view_all" and not _is_restricted_virtual_key_request(user_api_key_dict): 1330 ↛ 1331line 1330 didn't jump to line 1331 because the condition on line 1330 was never true
1331 servers = await global_mcp_server_manager.get_all_mcp_servers_with_health_unfiltered(server_ids=server_ids)
1332 return [{"server_id": server.server_id, "status": server.status} for server in servers]
1334 auth_contexts: Final = await build_effective_auth_contexts(user_api_key_dict)
1336 server_status_map: Final[dict[str, Literal["healthy", "unhealthy", "unknown"] | None]] = {}
1337 for auth_context in auth_contexts:
1338 servers = await global_mcp_server_manager.get_all_mcp_servers_with_health_and_teams(
1339 user_api_key_auth=auth_context,
1340 server_ids=server_ids,
1341 )
1342 for server in servers:
1343 if server.server_id not in server_status_map: 1343 ↛ 1342line 1343 didn't jump to line 1342 because the condition on line 1343 was always true
1344 server_status_map[server.server_id] = server.status
1346 return [{"server_id": server_id, "status": status} for server_id, status in server_status_map.items()]
1348 @router.post(
1349 "/server/register",
1350 description="Submit a new MCP server for admin review (non-admin users). Mirrors POST /guardrails/register.",
1351 dependencies=[Depends(user_api_key_auth)],
1352 response_model=LiteLLM_MCPServerTable,
1353 status_code=status.HTTP_201_CREATED,
1354 )
1355 @management_endpoint_wrapper
1356 async def register_mcp_server(
1357 payload: NewMCPServerRequest,
1358 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1359 ):
1360 """
1361 Allow team members to submit an MCP server for admin review.
1362 Creates the server with approval_status=pending_review.
1363 Requires a team-scoped API key.
1364 """
1365 if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN: 1365 ↛ 1373line 1365 didn't jump to line 1373 because the condition on line 1365 was always true
1366 raise HTTPException(
1367 status_code=status.HTTP_403_FORBIDDEN,
1368 detail={
1369 "error": "PROXY_ADMIN users should use POST /v1/mcp/server to create servers directly instead of the submission workflow."
1370 },
1371 )
1373 if not user_api_key_dict.team_id:
1374 raise HTTPException(
1375 status_code=status.HTTP_400_BAD_REQUEST,
1376 detail={"error": "Registration requires an API key associated with a team. Use a team-scoped key."},
1377 )
1379 # stdio servers spawn a local subprocess on the proxy host with the
1380 # configured command + args, so accepting them from non-admin callers
1381 # would let a team member propose a server config that an admin could
1382 # rubber-stamp into local code execution. Restrict stdio submission to
1383 # the admin POST /v1/mcp/server path or to config.yaml.
1384 if payload.transport == MCPTransport.stdio:
1385 raise HTTPException(
1386 status_code=status.HTTP_400_BAD_REQUEST,
1387 detail={
1388 "error": (
1389 "stdio MCP servers cannot be submitted via the user "
1390 "registration workflow. Ask a proxy admin to add this "
1391 "server via POST /v1/mcp/server or to declare it in "
1392 "config.yaml."
1393 )
1394 },
1395 )
1397 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1399 validate_and_normalize_mcp_server_payload(payload)
1400 stamp_omitted_oauth2_flow(payload)
1401 _validate_mcp_required_fields(payload)
1403 payload.approval_status = MCPApprovalStatus.pending_review
1404 payload.submitted_by = user_api_key_dict.user_id
1405 payload.submitted_at = datetime.now(timezone.utc)
1407 try:
1408 new_mcp_server: Final = await create_mcp_server_if_identifier_free(
1409 prisma_client,
1410 payload,
1411 touched_by=user_api_key_dict.user_id or user_api_key_dict.team_id,
1412 )
1413 except Exception as e:
1414 verbose_proxy_logger.exception("Error registering mcp server: %s", e)
1415 raise HTTPException(
1416 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
1417 detail={"error": f"Error registering mcp server: {e}"},
1418 )
1419 if isinstance(new_mcp_server, McpIdentifierConflict):
1420 raise_mcp_identifier_conflict(new_mcp_server)
1421 # Do NOT add to runtime registry — pending servers are not active
1422 return _redact_mcp_credentials(new_mcp_server)
1424 @router.get(
1425 "/sessions",
1426 description="Live stateful MCP gateway sessions on this proxy worker, grouped by AI client and by user.",
1427 dependencies=(Depends(user_api_key_auth),),
1428 response_model=MCPGatewaySessionsResponse,
1429 )
1430 @management_endpoint_wrapper
1431 async def get_mcp_gateway_sessions(
1432 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
1433 ) -> MCPGatewaySessionsResponse:
1434 if user_api_key_dict.user_role not in ( 1434 ↛ 1438line 1434 didn't jump to line 1438 because the condition on line 1434 was never true
1435 LitellmUserRoles.PROXY_ADMIN,
1436 LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
1437 ):
1438 raise HTTPException(
1439 status_code=status.HTTP_403_FORBIDDEN,
1440 detail={ # mutable-ok: HTTPException detail must be a plain mapping to keep this route's {"error": ...} response shape
1441 "error": "Admin access required to view MCP gateway sessions."
1442 },
1443 )
1444 from litellm.proxy._experimental.mcp_server.server import (
1445 get_mcp_gateway_sessions_report,
1446 )
1448 return get_mcp_gateway_sessions_report()
1450 @router.delete(
1451 "/sessions",
1452 description=(
1453 "Force-close live stateful MCP gateway sessions on this proxy worker, selected by session id prefix "
1454 "and/or by the LiteLLM user that opened them (proxy admin only)."
1455 ),
1456 dependencies=(Depends(user_api_key_auth),),
1457 response_model=MCPGatewaySessionsTerminateResponse,
1458 )
1459 @management_endpoint_wrapper
1460 async def delete_mcp_gateway_sessions(
1461 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
1462 session_id_prefix: Annotated[str | None, Query(min_length=MCP_GATEWAY_SESSION_ID_PREFIX_LENGTH)] = None,
1463 user_id: Annotated[str | None, Query(min_length=1)] = None,
1464 ) -> MCPGatewaySessionsTerminateResponse:
1465 if not _user_is_full_admin(user_api_key_dict): 1465 ↛ 1466line 1465 didn't jump to line 1466 because the condition on line 1465 was never true
1466 raise HTTPException(
1467 status_code=status.HTTP_403_FORBIDDEN,
1468 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
1469 "error": "Proxy admin access required to terminate MCP gateway sessions.",
1470 },
1471 )
1472 if session_id_prefix is None and user_id is None:
1473 raise HTTPException(
1474 status_code=status.HTTP_400_BAD_REQUEST,
1475 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
1476 "error": "Provide session_id_prefix and/or user_id to select the sessions to terminate.",
1477 },
1478 )
1479 from litellm.proxy._experimental.mcp_server.server import (
1480 terminate_mcp_gateway_sessions,
1481 )
1483 return await terminate_mcp_gateway_sessions(session_id_prefix=session_id_prefix, user_id=user_id)
1485 @router.get(
1486 "/server/submissions",
1487 description="Returns all MCP servers submitted by non-admin users (admin review queue). Mirrors GET /guardrails/submissions.",
1488 dependencies=[Depends(user_api_key_auth)],
1489 response_model=MCPSubmissionsSummary,
1490 )
1491 @management_endpoint_wrapper
1492 async def get_mcp_server_submissions(
1493 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1494 ):
1495 """
1496 Admin-only endpoint to view all user-submitted MCP servers pending review.
1497 """
1498 if user_api_key_dict.user_role not in ( 1498 ↛ 1502line 1498 didn't jump to line 1502 because the condition on line 1498 was never true
1499 LitellmUserRoles.PROXY_ADMIN,
1500 LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
1501 ):
1502 raise HTTPException(
1503 status_code=status.HTTP_403_FORBIDDEN,
1504 detail={"error": "Admin access required to view MCP server submissions."},
1505 )
1507 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1509 submissions: Final = await get_mcp_submissions(prisma_client)
1510 submissions.items = _redact_mcp_credentials_list(submissions.items)
1511 if not _user_is_full_admin(user_api_key_dict): 1511 ↛ 1512line 1511 didn't jump to line 1512 because the condition on line 1511 was never true
1512 submissions.items = _sanitize_mcp_server_list_for_non_admin(submissions.items)
1513 return submissions
1515 @router.put(
1516 "/server/{server_id}/approve",
1517 description="Approve a pending MCP server submission (admin only). Mirrors PUT /guardrails/{id}/approve.",
1518 dependencies=[Depends(user_api_key_auth)],
1519 response_model=LiteLLM_MCPServerTable,
1520 )
1521 @management_endpoint_wrapper
1522 async def approve_mcp_server_submission(
1523 server_id: str,
1524 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1525 ):
1526 """
1527 Admin approves a pending or previously-rejected MCP server — sets approval_status=active and loads it into the runtime registry.
1528 """
1529 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 1529 ↛ 1530line 1529 didn't jump to line 1530 because the condition on line 1529 was never true
1530 raise HTTPException(
1531 status_code=status.HTTP_403_FORBIDDEN,
1532 detail={"error": "Admin access required to approve MCP server submissions."},
1533 )
1535 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1537 existing: Final = await get_mcp_server(prisma_client, server_id)
1538 if existing is None:
1539 raise HTTPException(
1540 status_code=status.HTTP_404_NOT_FOUND,
1541 detail={"error": f"MCP server '{server_id}' not found."},
1542 )
1543 if existing.approval_status == MCPApprovalStatus.active:
1544 raise HTTPException(
1545 status_code=status.HTTP_400_BAD_REQUEST,
1546 detail={"error": "MCP server is already active."},
1547 )
1549 approved: Final = await approve_mcp_server(
1550 prisma_client,
1551 server_id,
1552 touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
1553 )
1554 await global_mcp_server_manager.invalidate_byom_submitted_servers_cache(approved.submitted_by)
1555 await global_mcp_server_manager.reload_servers_from_database()
1557 return _redact_mcp_credentials(approved)
1559 @router.put(
1560 "/server/{server_id}/reject",
1561 description="Reject a pending MCP server submission (admin only). Mirrors PUT /guardrails/{id}/reject.",
1562 dependencies=[Depends(user_api_key_auth)],
1563 response_model=LiteLLM_MCPServerTable,
1564 )
1565 @management_endpoint_wrapper
1566 async def reject_mcp_server_submission(
1567 server_id: str,
1568 payload: RejectMCPServerRequest,
1569 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1570 ):
1571 """
1572 Admin rejects a pending MCP server — sets approval_status=rejected with optional review_notes.
1573 """
1574 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 1574 ↛ 1575line 1574 didn't jump to line 1575 because the condition on line 1574 was never true
1575 raise HTTPException(
1576 status_code=status.HTTP_403_FORBIDDEN,
1577 detail={"error": "Admin access required to reject MCP server submissions."},
1578 )
1580 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1582 existing: Final = await get_mcp_server(prisma_client, server_id)
1583 if existing is None:
1584 raise HTTPException(
1585 status_code=status.HTTP_404_NOT_FOUND,
1586 detail={"error": f"MCP server '{server_id}' not found."},
1587 )
1588 if existing.approval_status == MCPApprovalStatus.rejected:
1589 raise HTTPException(
1590 status_code=status.HTTP_400_BAD_REQUEST,
1591 detail={"error": "MCP server is already rejected."},
1592 )
1594 was_active: Final = existing.approval_status == MCPApprovalStatus.active
1595 rejected: Final = await reject_mcp_server(
1596 prisma_client,
1597 server_id,
1598 touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
1599 review_notes=payload.review_notes,
1600 )
1601 # Only evict from the runtime registry if the server was previously active
1602 if was_active: 1602 ↛ 1604line 1602 didn't jump to line 1604 because the condition on line 1602 was always true
1603 await global_mcp_server_manager.reload_servers_from_database()
1604 return _redact_mcp_credentials(rejected)
1606 @router.get(
1607 "/server/{server_id}",
1608 description="Returns the mcp server info",
1609 dependencies=[Depends(user_api_key_auth)],
1610 response_model=LiteLLM_MCPServerTable,
1611 )
1612 async def fetch_mcp_server(
1613 request: Request,
1614 server_id: str,
1615 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1616 ):
1617 """
1618 Get the info on the mcp server specified by the `server_id`
1619 Parameters:
1620 - server_id: str - Required. The unique identifier of the mcp server to get info on.
1621 ```
1622 curl --location 'http://localhost:4000/v1/mcp/server/server_id' \
1623 --header 'Authorization: Bearer your_api_key_here'
1624 ```
1625 """
1626 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1628 # check to see if server exists (DB first, then registry for config-based servers)
1629 mcp_server = await get_mcp_server(prisma_client, server_id)
1630 from_db: Final = mcp_server is not None
1632 if mcp_server is None: 1632 ↛ 1634line 1632 didn't jump to line 1634 because the condition on line 1632 was never true
1633 # Fallback: check registry (config-based servers) - list endpoint uses get_registry()
1634 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
1636 client_ip: Final = IPAddressUtils.get_mcp_client_ip(request)
1637 registry_server = global_mcp_server_manager.get_mcp_server_by_id(server_id)
1638 if registry_server is not None and not global_mcp_server_manager._is_server_accessible_from_ip(
1639 registry_server, client_ip
1640 ):
1641 registry_server = None
1642 if registry_server is None:
1643 # Try lookup by server_name or alias (client may use display name in URL)
1644 registry_server = global_mcp_server_manager.get_mcp_server_by_name(server_id, client_ip=client_ip)
1645 if registry_server is not None:
1646 mcp_server = global_mcp_server_manager._build_mcp_server_table(registry_server)
1648 if mcp_server is None: 1648 ↛ 1649line 1648 didn't jump to line 1649 because the condition on line 1648 was never true
1649 raise HTTPException(
1650 status_code=status.HTTP_404_NOT_FOUND,
1651 detail={"error": f"MCP Server with id {server_id} not found"},
1652 )
1654 # Implement authz restriction from requested user
1655 is_admin_view: Final = _user_has_admin_view(user_api_key_dict)
1656 is_restricted_virtual_key: Final = _is_restricted_virtual_key_request(user_api_key_dict)
1658 if not is_admin_view: 1658 ↛ 1661line 1658 didn't jump to line 1661 because the condition on line 1658 was never true
1659 # Perform authz check BEFORE any health check (avoid side-effects for
1660 # unauthorized callers).
1661 if from_db:
1662 mcp_server_records: Final = await get_all_mcp_servers_for_user(prisma_client, user_api_key_dict)
1663 exists = does_mcp_server_exist(mcp_server_records, server_id)
1664 else:
1665 # Registry/config server: use same access logic as list endpoint
1666 allowed_server_ids: Final = await global_mcp_server_manager.get_allowed_mcp_servers(user_api_key_dict)
1667 exists = mcp_server.server_id in allowed_server_ids
1669 if not exists:
1670 raise HTTPException(
1671 status_code=status.HTTP_403_FORBIDDEN,
1672 detail={
1673 "error": (
1674 f"User does not have permission to view mcp server with id {server_id}. "
1675 "You can only view mcp servers that you have access to."
1676 )
1677 },
1678 )
1680 # At this point caller is authorized to view the server.
1681 if from_db: 1681 ↛ 1685line 1681 didn't jump to line 1685 because the condition on line 1681 was always true
1682 await global_mcp_server_manager.add_server(mcp_server)
1684 # Perform health check on the server using server manager
1685 try:
1686 health_result: Final = await global_mcp_server_manager.health_check_server(server_id)
1687 # Update the server object with health check results
1688 mcp_server.status = health_result.status if health_result.status else "unknown"
1689 mcp_server.last_health_check = health_result.last_health_check
1690 mcp_server.health_check_error = health_result.health_check_error
1691 except Exception as e:
1692 verbose_proxy_logger.debug("Error performing health check on server %s: %s", server_id, e)
1693 mcp_server.status = "unknown"
1694 mcp_server.last_health_check = datetime.now()
1695 mcp_server.health_check_error = str(e)
1697 redacted: Final = _redact_mcp_credentials(mcp_server)
1698 if is_restricted_virtual_key: 1698 ↛ 1699line 1698 didn't jump to line 1699 because the condition on line 1698 was never true
1699 return _sanitize_mcp_server_for_virtual_key(redacted)
1700 # only a full PROXY_ADMIN sees credential-bearing fields; everyone else
1701 # goes through the non-admin sanitizer
1702 if not _user_is_full_admin(user_api_key_dict): 1702 ↛ 1703line 1702 didn't jump to line 1703 because the condition on line 1702 was never true
1703 return _sanitize_mcp_server_for_non_admin(redacted)
1704 return redacted
1706 @router.post(
1707 "/server",
1708 description="Allows creation of mcp servers",
1709 dependencies=[Depends(user_api_key_auth)],
1710 response_model=LiteLLM_MCPServerTable,
1711 status_code=status.HTTP_201_CREATED,
1712 )
1713 @management_endpoint_wrapper
1714 async def add_mcp_server(
1715 payload: NewMCPServerRequest,
1716 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1717 litellm_changed_by: str | None = Header(
1718 None,
1719 description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
1720 ),
1721 ):
1722 """
1723 Allow users to add a new external mcp server.
1724 """
1725 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1727 # Validate and normalize payload fields
1728 validate_and_normalize_mcp_server_payload(payload)
1729 stamp_omitted_oauth2_flow(payload)
1731 # AuthZ - restrict only proxy admins to create mcp servers
1732 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 1732 ↛ 1733line 1732 didn't jump to line 1733 because the condition on line 1732 was never true
1733 raise HTTPException(
1734 status_code=status.HTTP_403_FORBIDDEN,
1735 detail={
1736 "error": "User does not have permission to create mcp servers. You can only create mcp servers if you are a PROXY_ADMIN."
1737 },
1738 )
1740 # Block reserved special server IDs
1741 if ( 1741 ↛ 1745line 1741 didn't jump to line 1745 because the condition on line 1741 was never true
1742 SpecialMCPServerName.all_team_servers == payload.server_id
1743 or SpecialMCPServerName.all_proxy_servers == payload.server_id
1744 ):
1745 raise HTTPException(
1746 status_code=status.HTTP_400_BAD_REQUEST,
1747 detail={"error": f"MCP Server with id {payload.server_id} is special and cannot be used."},
1748 )
1750 if payload.server_id is not None:
1751 # fail if the mcp server with id already exists
1752 mcp_server: Final = await get_mcp_server(prisma_client, payload.server_id)
1753 if mcp_server is not None:
1754 raise HTTPException(
1755 status_code=status.HTTP_400_BAD_REQUEST,
1756 detail={"error": f"MCP Server with id {payload.server_id} already exists. Cannot create another."},
1757 )
1759 # TODO: audit log for create
1761 # Admin-created servers are always active — clear any submission lifecycle
1762 # fields the caller may have provided to prevent fake entries appearing in
1763 # the submissions queue.
1764 payload.approval_status = MCPApprovalStatus.active
1765 payload.submitted_by = None
1766 payload.submitted_at = None
1768 # The database write is the commit point: if it fails nothing was
1769 # persisted and the request is a genuine failure.
1770 try:
1771 new_mcp_server: Final = await create_mcp_server_if_identifier_free(
1772 prisma_client,
1773 payload,
1774 touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
1775 )
1776 except Exception as e:
1777 verbose_proxy_logger.exception("Error creating mcp server: %s", e)
1778 raise HTTPException(
1779 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
1780 detail={"error": f"Error creating mcp server: {e}"},
1781 )
1782 if isinstance(new_mcp_server, McpIdentifierConflict): 1782 ↛ 1783line 1782 didn't jump to line 1783 because the condition on line 1782 was never true
1783 raise_mcp_identifier_conflict(new_mcp_server)
1785 warn_if_id_jag_server_outruns_sso(new_mcp_server.server_id, new_mcp_server.auth_type)
1787 # Registry refresh is best-effort: the row is already committed, so a
1788 # failure here (e.g. an unrelated malformed row in the table) must not
1789 # surface as a 500 and orphan the created server, which would push the
1790 # caller to retry and create duplicates.
1791 try:
1792 await global_mcp_server_manager.add_server(new_mcp_server)
1793 await global_mcp_server_manager.reload_servers_from_database()
1794 except Exception as e:
1795 verbose_proxy_logger.exception(
1796 "MCP server %s created but in-memory registry refresh failed: %s", new_mcp_server.server_id, e
1797 )
1799 return _redact_mcp_credentials(new_mcp_server)
1801 @router.post(
1802 "/server/import",
1803 description="Bulk-import MCP connectors from Anthropic mcpServers or mcp_servers JSON",
1804 dependencies=(Depends(user_api_key_auth),),
1805 response_model=MCPConnectorImportResponse,
1806 status_code=status.HTTP_200_OK,
1807 )
1808 @management_endpoint_wrapper
1809 async def import_mcp_servers(
1810 payload: MCPConnectorImportRequest,
1811 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), # noqa: B008 # FastAPI dependency injection
1812 ):
1813 """
1814 Bulk-import MCP connectors. Accepts the Claude Desktop / Claude Code
1815 ``mcpServers`` mapping or the Anthropic Messages API ``mcp_servers``
1816 array, creates each entry as a LiteLLM MCP server, and returns
1817 per-entry results so partial imports are visible to the caller.
1818 """
1819 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
1821 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 1821 ↛ 1822line 1821 didn't jump to line 1822 because the condition on line 1821 was never true
1822 raise HTTPException(
1823 status_code=status.HTTP_403_FORBIDDEN,
1824 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
1825 "error": "User does not have permission to import mcp servers. You can only import mcp servers if you are a PROXY_ADMIN."
1826 },
1827 )
1829 conversions: Final = convert_connector_entries(payload)
1830 existing_servers: Final = await get_all_mcp_servers(prisma_client)
1831 existing_names: Final = frozenset(
1832 name.lower() for server in existing_servers for name in (server.alias, server.server_name) if name
1833 )
1835 def _classify(
1836 index: int, conversion: ConvertedConnector | ConnectorConversionError
1837 ) -> ConvertedConnector | ConnectorConversionError | MCPConnectorImportSkipped:
1838 if isinstance(conversion, ConnectorConversionError): 1838 ↛ 1840line 1838 didn't jump to line 1840 because the condition on line 1838 was always true
1839 return conversion
1840 alias: Final = conversion.request.alias or ""
1841 if alias.lower() in existing_names:
1842 return MCPConnectorImportSkipped(
1843 name=conversion.name, reason=f"An MCP server named '{alias}' already exists."
1844 )
1845 earlier_aliases: Final = frozenset(
1846 (earlier.request.alias or "").lower()
1847 for earlier in conversions[:index]
1848 if isinstance(earlier, ConvertedConnector)
1849 )
1850 if alias.lower() in earlier_aliases:
1851 return MCPConnectorImportSkipped(
1852 name=conversion.name, reason=f"Duplicate connector name '{alias}' in the import payload."
1853 )
1854 return conversion
1856 async def _create(
1857 conversion: ConvertedConnector,
1858 ) -> MCPConnectorImportResult | MCPConnectorImportFailure | MCPConnectorImportSkipped:
1859 try:
1860 validate_and_normalize_mcp_server_payload(conversion.request)
1861 except HTTPException as e:
1862 error_text: Final = (
1863 str(e.detail.get("error", e.detail)) if isinstance(e.detail, dict) else str(e.detail)
1864 )
1865 return MCPConnectorImportFailure(name=conversion.name, error=error_text)
1866 try:
1867 created: Final = await create_mcp_server_if_identifier_free(
1868 prisma_client,
1869 conversion.request,
1870 touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
1871 )
1872 except Exception as e: # noqa: BLE001 # any create failure must become a per-entry error, not a 500
1873 verbose_proxy_logger.exception("Error importing mcp server %s: %s", conversion.name, e)
1874 return MCPConnectorImportFailure(name=conversion.name, error=str(e))
1875 if isinstance(created, McpIdentifierConflict):
1876 return MCPConnectorImportSkipped(name=conversion.name, reason=mcp_identifier_conflict_message(created))
1877 try:
1878 await global_mcp_server_manager.add_server(created)
1879 except Exception as e: # noqa: BLE001 # the row is committed; the reload after the loop retries registration
1880 verbose_proxy_logger.exception(
1881 "Imported mcp server %s committed but in-memory registration failed: %s", conversion.name, e
1882 )
1883 return MCPConnectorImportResult(
1884 name=conversion.name, server_id=created.server_id, alias=created.alias or ""
1885 )
1887 classified: Final = tuple(_classify(index, conversion) for index, conversion in enumerate(conversions))
1888 outcomes: Final = tuple(
1889 [await _create(entry) if isinstance(entry, ConvertedConnector) else entry for entry in classified]
1890 )
1892 imported: Final = tuple(entry for entry in outcomes if isinstance(entry, MCPConnectorImportResult))
1893 if imported: 1893 ↛ 1894line 1893 didn't jump to line 1894 because the condition on line 1893 was never true
1894 try:
1895 await global_mcp_server_manager.reload_servers_from_database()
1896 except Exception as e: # noqa: BLE001 # rows are committed; a refresh failure must not surface as a 500
1897 verbose_proxy_logger.exception("MCP connector import committed but registry refresh failed: %s", e)
1899 return MCPConnectorImportResponse(
1900 imported=imported,
1901 skipped=tuple(entry for entry in outcomes if isinstance(entry, MCPConnectorImportSkipped)),
1902 errors=tuple(
1903 MCPConnectorImportFailure(name=entry.name, error=entry.error)
1904 if isinstance(entry, ConnectorConversionError)
1905 else entry
1906 for entry in outcomes
1907 if isinstance(entry, (ConnectorConversionError, MCPConnectorImportFailure))
1908 ),
1909 )
1911 @router.post(
1912 "/server/oauth/session",
1913 description="Temporarily cache an MCP server in memory without writing to the database",
1914 dependencies=[Depends(user_api_key_auth)],
1915 status_code=status.HTTP_200_OK,
1916 )
1917 @management_endpoint_wrapper
1918 async def add_session_mcp_server(
1919 payload: NewMCPServerRequest,
1920 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
1921 litellm_changed_by: str | None = Header(
1922 None,
1923 description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
1924 ),
1925 ):
1926 """
1927 Cache MCP server info in memory for a short duration (~5 minutes).
1929 This endpoint does not write to the database. If the same server_id is provided
1930 again while the cache entry is active, it will refresh the cached data + TTL.
1931 """
1933 # Validate and normalize payload fields (alias/server name rules)
1934 validate_and_normalize_mcp_server_payload(payload)
1935 stamp_omitted_oauth2_flow(payload)
1937 # Restrict to proxy admins similar to the persistent create endpoint
1938 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 1938 ↛ 1939line 1938 didn't jump to line 1939 because the condition on line 1938 was never true
1939 raise HTTPException(
1940 status_code=status.HTTP_403_FORBIDDEN,
1941 detail={
1942 "error": "User does not have permission to create temporary mcp servers. You can only create temporary mcp servers if you are a PROXY_ADMIN."
1943 },
1944 )
1946 created_by: Final = user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME
1947 payload_with_credentials: Final = _inherit_credentials_from_existing_server(payload)
1948 temp_record: Final = _build_temporary_mcp_server_record(
1949 payload_with_credentials,
1950 created_by,
1951 await _resolve_session_server_id(payload_with_credentials),
1952 )
1954 try:
1955 temporary_server: Final = await global_mcp_server_manager.build_mcp_server_from_table(
1956 temp_record,
1957 credentials_are_encrypted=False,
1958 )
1959 _cache_temporary_mcp_server(
1960 temporary_server,
1961 ttl_seconds=TEMPORARY_MCP_SERVER_TTL_SECONDS,
1962 )
1963 await _persist_draft_mcp_server(
1964 payload_with_credentials,
1965 temp_record.server_id,
1966 created_by,
1967 )
1968 await _cache_temporary_mcp_server_in_redis(
1969 temporary_server,
1970 ttl_seconds=TEMPORARY_MCP_SERVER_TTL_SECONDS,
1971 )
1972 except Exception as e:
1973 verbose_proxy_logger.exception("Error caching temporary mcp server: %s", e)
1974 raise HTTPException(
1975 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
1976 detail={"error": f"Error caching temporary mcp server: {e}"},
1977 )
1979 return _redact_mcp_credentials(temp_record)
1981 async def _mcp_oauth_user_api_key_auth(request: Request) -> UserAPIKeyAuth:
1982 """
1983 Auth dependency for MCP OAuth browser-navigation endpoints (/authorize, /token).
1985 Tries the Authorization header first. Falls back to decoding the UI
1986 'token' session cookie (set by SSO login) to extract the API key, which
1987 allows browser-based OAuth redirects to work without an explicit
1988 Authorization header.
1989 """
1990 import jwt as _jwt
1992 from litellm.proxy.proxy_server import master_key
1994 auth_header: Final = request.headers.get("Authorization", "")
1995 api_key = auth_header # _get_bearer_token will strip "Bearer " prefix
1997 if not api_key:
1998 token_cookie: Final = request.cookies.get("token")
1999 if token_cookie and master_key:
2000 try:
2001 decoded: Final = _jwt.decode(
2002 token_cookie,
2003 master_key,
2004 algorithms=["HS256"],
2005 # UI session cookies may omit exp; don't require it.
2006 options={"verify_exp": False, "verify_aud": False},
2007 )
2008 if decoded.get("login_method") in ("sso", "username_password"):
2009 cookie_key: Final[str] = decoded.get("key", "")
2010 if cookie_key:
2011 api_key = f"Bearer {cookie_key}"
2012 except _jwt.InvalidTokenError:
2013 pass
2015 # For delegate_auth_to_upstream servers the entire PKCE handshake
2016 # (both /authorize browser redirect and /token authorization_code
2017 # exchange) must work without a LiteLLM session. /authorize is opened
2018 # in a VS Code webview that may have no cookie; /token is a programmatic
2019 # POST from VS Code. PKCE security (code_verifier) guarantees the
2020 # authorization_code exchange cannot be replayed, so anonymous access
2021 # is safe for that grant only.
2022 #
2023 # Importantly, NOT safe for refresh_token grants: ``mcp_token`` will
2024 # forward the request to the upstream issuer with LiteLLM's stored
2025 # ``client_secret`` attached, so any caller holding a refresh token
2026 # issued to this client could mint fresh upstream access tokens through
2027 # us. Require normal LiteLLM auth for those.
2028 if not api_key:
2029 from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( # noqa: PLC0415
2030 global_mcp_server_manager,
2031 )
2032 from litellm.proxy.auth.auth_utils import ( # noqa: PLC0415
2033 get_request_route,
2034 )
2036 server_id: Final[str] = request.path_params.get("server_id", "")
2037 if server_id:
2038 _s = global_mcp_server_manager.get_mcp_server_by_id(server_id)
2039 if not _s:
2040 _s = global_mcp_server_manager.get_mcp_server_by_name(server_id)
2041 if (
2042 _s
2043 and getattr(_s, "auth_type", None) == MCPAuth.oauth2
2044 and getattr(_s, "delegate_auth_to_upstream", False) is True
2045 # M2M servers fetch tokens with stored credentials; never
2046 # expose their /authorize or /token endpoints anonymously.
2047 and not _s.has_client_credentials
2048 ):
2049 # For /token, require PKCE authorization_code; refresh_token
2050 # grants must NOT bypass auth (see comment above).
2051 path_lower: Final = get_request_route(request).rstrip("/").lower()
2052 if path_lower.endswith("/token"):
2053 body_data: Final = await _read_request_body(request=request)
2054 grant_type: Final = (body_data or {}).get("grant_type", "")
2055 if grant_type != "authorization_code":
2056 # Fall through to normal LiteLLM auth (will 401 if
2057 # no key supplied).
2058 pass
2059 else:
2060 return UserAPIKeyAuth()
2061 else:
2062 # /authorize and other PKCE-flow GETs are safe to
2063 # bypass: PKCE binds the upstream issuer's ``code``
2064 # to the original ``code_challenge`` so no anonymous
2065 # token can be minted via the redirect alone.
2066 return UserAPIKeyAuth()
2068 request_data = await _read_request_body(request=request)
2069 request_data = populate_request_with_path_params(request_data=request_data, request=request)
2071 return await _user_api_key_auth_builder(
2072 request=request,
2073 api_key=api_key,
2074 azure_api_key_header="",
2075 anthropic_api_key_header=None,
2076 google_ai_studio_api_key_header=None,
2077 azure_apim_header=None,
2078 request_data=request_data,
2079 )
2081 async def _get_cached_temporary_mcp_server_or_404(
2082 server_id: str,
2083 user_api_key_dict: UserAPIKeyAuth,
2084 request: Request | None = None,
2085 ) -> MCPServer:
2086 server = await get_cached_temporary_mcp_server(server_id)
2087 resolved_from_temp_cache: Final = server is not None
2088 if server is None:
2089 # Fall back to real DB/config server (e.g. for the user-side OAuth flow
2090 # which calls these endpoints with a real server_id, not a temp session id).
2091 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
2093 client_ip: Final = IPAddressUtils.get_mcp_client_ip(request) if request else None
2094 server = global_mcp_server_manager.get_mcp_server_by_id(
2095 server_id
2096 ) or global_mcp_server_manager.get_mcp_server_by_name(server_id, client_ip=client_ip)
2097 if server is None:
2098 raise HTTPException(
2099 status_code=status.HTTP_404_NOT_FOUND,
2100 detail={"error": f"MCP server {server_id} not found"},
2101 )
2103 # Per-server access policy mirrors `fetch_mcp_server`: admin-view
2104 # callers are unrestricted; non-admins must have the server in their
2105 # allowed-servers set. Temporary cached servers come from the
2106 # admin-only `/server/oauth/session` setup flow and are not exposed
2107 # to non-admins.
2108 if not _user_has_admin_view(user_api_key_dict):
2109 if resolved_from_temp_cache:
2110 raise HTTPException(
2111 status_code=status.HTTP_403_FORBIDDEN,
2112 detail={"error": f"Access denied to MCP server {server_id}"},
2113 )
2114 allowed_server_ids: Final[set[str]] = set()
2115 for auth_context in await build_effective_auth_contexts(user_api_key_dict):
2116 allowed_server_ids.update(await global_mcp_server_manager.get_allowed_mcp_servers(auth_context))
2117 if server.server_id not in allowed_server_ids:
2118 raise HTTPException(
2119 status_code=status.HTTP_403_FORBIDDEN,
2120 detail={"error": f"Access denied to MCP server {server_id}"},
2121 )
2122 return server
2124 @router.get(
2125 "/server/oauth/{server_id}/authorize",
2126 include_in_schema=False,
2127 dependencies=[Depends(_mcp_oauth_user_api_key_auth)],
2128 )
2129 async def mcp_authorize(
2130 request: Request,
2131 server_id: str,
2132 user_api_key_dict: UserAPIKeyAuth = Depends(_mcp_oauth_user_api_key_auth),
2133 client_id: str | None = None,
2134 redirect_uri: str = Query(...),
2135 state: str = "",
2136 code_challenge: str | None = None,
2137 code_challenge_method: str | None = None,
2138 response_type: str | None = None,
2139 scope: str | None = None,
2140 ):
2141 mcp_server: Final = await _get_cached_temporary_mcp_server_or_404(server_id, user_api_key_dict, request=request)
2142 _raise_if_not_oauth2(mcp_server)
2143 # Use the server's stored client_id when the caller doesn't supply one
2144 stored_or_supplied_client_id: Final = mcp_server.client_id or client_id or ""
2145 ephemeral_dcr_client: Final = (
2146 await resolve_ephemeral_dcr_client(
2147 request=request,
2148 mcp_server=mcp_server,
2149 code_challenge=code_challenge,
2150 code_challenge_method=code_challenge_method,
2151 redirect_uri=redirect_uri,
2152 )
2153 if not stored_or_supplied_client_id
2154 else None
2155 )
2156 resolved_client_id: Final = stored_or_supplied_client_id or (
2157 ephemeral_dcr_client.client_id if ephemeral_dcr_client else ""
2158 )
2159 if not resolved_client_id:
2160 raise HTTPException(
2161 status_code=status.HTTP_400_BAD_REQUEST,
2162 detail={
2163 "error": "missing_client_id",
2164 "message": (
2165 "No client_id available for this MCP server. "
2166 "Either configure the server with a client_id or supply one in the request."
2167 ),
2168 },
2169 )
2170 return await authorize_with_server(
2171 request=request,
2172 mcp_server=mcp_server,
2173 client_id=resolved_client_id,
2174 redirect_uri=redirect_uri,
2175 state=state,
2176 code_challenge=code_challenge,
2177 code_challenge_method=code_challenge_method,
2178 response_type=response_type,
2179 scope=scope,
2180 ephemeral_dcr_client=ephemeral_dcr_client,
2181 )
2183 @router.post(
2184 "/server/oauth/{server_id}/token",
2185 include_in_schema=False,
2186 dependencies=[Depends(_mcp_oauth_user_api_key_auth)],
2187 )
2188 async def mcp_token(
2189 request: Request,
2190 server_id: str,
2191 user_api_key_dict: UserAPIKeyAuth = Depends(_mcp_oauth_user_api_key_auth),
2192 grant_type: str = Form(...),
2193 code: str | None = Form(None),
2194 redirect_uri: str | None = Form(None),
2195 client_id: str | None = Form(None),
2196 client_secret: str | None = Form(None),
2197 code_verifier: str | None = Form(None),
2198 refresh_token: str | None = Form(None),
2199 scope: str | None = Form(None),
2200 ):
2201 mcp_server: Final = await _get_cached_temporary_mcp_server_or_404(server_id, user_api_key_dict, request=request)
2202 _raise_if_not_oauth2(mcp_server)
2203 # Sealed passthrough codes exist only for the authorization_code grant. A refresh_token
2204 # grant must never open one: the minted client is unrecoverable after the single flow by
2205 # contract, so an expired browser-held token re-runs authorize instead.
2206 sealed_code: Final = (
2207 redeem_passthrough_authorization_code(code=code, mcp_server=mcp_server, code_verifier=code_verifier)
2208 if grant_type == "authorization_code"
2209 else None
2210 )
2211 resolved_code: Final = sealed_code.upstream_code if sealed_code else code
2212 # A sealed flow ran the gateway /callback as its upstream redirect (bridge short-circuit
2213 # or plain flow alike), so the exchange must present that binding, not the browser page.
2214 resolved_redirect_uri: Final = f"{get_request_base_url(request)}/callback" if sealed_code else redirect_uri
2215 caller_client_id: Final = sealed_code.client_id if sealed_code else client_id
2216 caller_client_secret: Final = sealed_code.client_secret if sealed_code else client_secret
2217 resolved_client_id: Final = mcp_server.client_id or caller_client_id or ""
2218 if not resolved_client_id:
2219 raise HTTPException(
2220 status_code=status.HTTP_400_BAD_REQUEST,
2221 detail={
2222 "error": "missing_client_id",
2223 "message": (
2224 "No client_id available for this MCP server. "
2225 "Either configure the server with a client_id or supply one in the request."
2226 ),
2227 },
2228 )
2229 return await exchange_token_with_server(
2230 request=request,
2231 mcp_server=mcp_server,
2232 grant_type=grant_type,
2233 code=resolved_code,
2234 redirect_uri=resolved_redirect_uri,
2235 client_id=resolved_client_id,
2236 client_secret=caller_client_secret,
2237 code_verifier=code_verifier,
2238 refresh_token=refresh_token,
2239 scope=scope,
2240 client_token_endpoint_auth_method=sealed_code.token_endpoint_auth_method if sealed_code else None,
2241 )
2243 @router.post(
2244 "/server/oauth/{server_id}/register",
2245 include_in_schema=False,
2246 dependencies=[Depends(user_api_key_auth)],
2247 )
2248 async def mcp_register(
2249 request: Request,
2250 server_id: str,
2251 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2252 ):
2253 mcp_server: Final = await _get_cached_temporary_mcp_server_or_404(server_id, user_api_key_dict, request=request)
2254 request_data: Final = await _read_request_body(request=request)
2255 data: Final[dict] = {**request_data}
2256 client_redirect_uris: Final = client_supplied_redirect_uris(data.get("redirect_uris"))
2258 return await register_client_with_server(
2259 request=request,
2260 mcp_server=mcp_server,
2261 client_name=data.get("client_name", ""),
2262 grant_types=data.get("grant_types", []),
2263 response_types=data.get("response_types", []),
2264 token_endpoint_auth_method=data.get("token_endpoint_auth_method", ""),
2265 fallback_client_id=server_id,
2266 persist_credentials=_user_is_full_admin(user_api_key_dict),
2267 client_redirect_uris=client_redirect_uris,
2268 )
2270 @router.delete(
2271 "/server/{server_id}",
2272 description="Allows deleting mcp serves in the db",
2273 dependencies=[Depends(user_api_key_auth)],
2274 response_class=JSONResponse,
2275 status_code=status.HTTP_202_ACCEPTED,
2276 )
2277 @management_endpoint_wrapper
2278 async def remove_mcp_server(
2279 server_id: str,
2280 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2281 litellm_changed_by: str | None = Header(
2282 None,
2283 description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
2284 ),
2285 ):
2286 """
2287 Delete MCP Server from db and associated MCP related server entities.
2289 Parameters:
2290 - server_id: str - Required. The unique identifier of the mcp server to delete.
2291 ```
2292 curl -X "DELETE" --location 'http://localhost:4000/v1/mcp/server/server_id' \
2293 --header 'Authorization: Bearer your_api_key_here'
2294 ```
2295 """
2296 prisma_client: Final = get_prisma_client_or_throw(
2297 "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
2298 )
2300 # Authz - restrict only admins to delete mcp servers
2301 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 2301 ↛ 2302line 2301 didn't jump to line 2302 because the condition on line 2301 was never true
2302 raise HTTPException(
2303 status_code=status.HTTP_403_FORBIDDEN,
2304 detail={
2305 "error": "Call not allowed to delete MCP server. User is not a proxy admin. route={}".format(
2306 "DELETE /v1/mcp/server"
2307 )
2308 },
2309 )
2311 # try to delete the mcp server
2312 mcp_server_record_deleted: Final = await delete_mcp_server(prisma_client, server_id)
2314 if mcp_server_record_deleted is None:
2315 raise HTTPException(
2316 status_code=status.HTTP_404_NOT_FOUND,
2317 detail={"error": f"MCP Server not found, passed server_id={server_id}"},
2318 )
2319 global_mcp_server_manager.remove_server(mcp_server_record_deleted)
2321 # Ensure registry is up to date by reloading from database
2322 await global_mcp_server_manager.reload_servers_from_database()
2324 # TODO: Enterprise: Finish audit log trail
2325 if is_audit_logging_enabled(): 2325 ↛ 2326line 2325 didn't jump to line 2326 because the condition on line 2325 was never true
2326 pass
2328 # TODO: Delete from virtual keys
2330 # TODO: Delete from teams
2332 # Update from global mcp store
2334 return Response(status_code=status.HTTP_202_ACCEPTED)
2336 @router.post(
2337 "/server/{server_id}/user-credential",
2338 description="Store or update the calling user's API key for a BYOK MCP server",
2339 dependencies=[Depends(user_api_key_auth)],
2340 response_model=MCPUserCredentialResponse,
2341 )
2342 @management_endpoint_wrapper
2343 async def store_mcp_user_credential(
2344 server_id: str,
2345 payload: MCPUserCredentialRequest,
2346 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2347 ):
2348 """Store a BYOK credential for the calling user."""
2349 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2350 mcp_server: Final = await _authorize_and_fetch_mcp_server(prisma_client, user_api_key_dict, server_id)
2351 if not getattr(mcp_server, "is_byok", False): 2351 ↛ 2356line 2351 didn't jump to line 2356 because the condition on line 2351 was always true
2352 raise HTTPException(
2353 status_code=status.HTTP_400_BAD_REQUEST,
2354 detail={"error": "This MCP server does not support BYOK credentials"},
2355 )
2356 user_id: Final = user_api_key_dict.user_id or ""
2357 if not user_id:
2358 raise HTTPException(
2359 status_code=status.HTTP_400_BAD_REQUEST,
2360 detail={"error": "User ID not found in token"},
2361 )
2362 if payload.save:
2363 await store_user_credential(prisma_client, user_id, server_id, payload.credential)
2364 from litellm.proxy._experimental.mcp_server.server import (
2365 _invalidate_byok_cred_cache,
2366 )
2368 await _invalidate_byok_cred_cache(user_id, server_id)
2369 return MCPUserCredentialResponse(server_id=server_id, has_credential=True)
2370 # save=False: credential not persisted
2371 return MCPUserCredentialResponse(server_id=server_id, has_credential=False)
2373 @router.delete(
2374 "/server/{server_id}/user-credential",
2375 description=(
2376 "Delete the calling user's stored API key for a BYOK MCP server. "
2377 "A proxy admin may pass user_id to revoke another user's stored key."
2378 ),
2379 dependencies=[Depends(user_api_key_auth)],
2380 response_model=MCPUserCredentialResponse,
2381 )
2382 @management_endpoint_wrapper
2383 async def delete_mcp_user_credential(
2384 server_id: str,
2385 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2386 user_id: Annotated[str | None, Query(min_length=1)] = None,
2387 ):
2388 """Remove the target user's BYOK credential (the caller unless an admin names another user)."""
2389 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2390 target_user_id: Final = _resolve_credential_target_user_id(user_api_key_dict, user_id)
2391 try:
2392 await delete_user_credential(prisma_client, target_user_id, server_id)
2393 except RecordNotFoundError:
2394 pass # Already deleted or didn't exist
2395 from litellm.proxy._experimental.mcp_server.server import (
2396 _invalidate_byok_cred_cache,
2397 )
2399 await _invalidate_byok_cred_cache(target_user_id, server_id)
2400 return MCPUserCredentialResponse(server_id=server_id, has_credential=False)
2402 # ── OAuth2 user-credential endpoints ──────────────────────────────────────
2404 @router.post(
2405 "/server/{server_id}/oauth-user-credential",
2406 description="Store the calling user's OAuth2 token for an OpenAPI MCP server",
2407 dependencies=[Depends(user_api_key_auth)],
2408 response_model=MCPOAuthUserCredentialStatus,
2409 )
2410 @management_endpoint_wrapper
2411 async def store_mcp_oauth_user_credential(
2412 server_id: str,
2413 payload: MCPOAuthUserCredentialRequest,
2414 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2415 ):
2416 """Persist the OAuth2 access token obtained by the calling user."""
2417 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2418 await _authorize_and_fetch_mcp_server(prisma_client, user_api_key_dict, server_id)
2419 from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( # noqa: PLC0415 # keep manager import lazy
2420 global_mcp_server_manager as _manager,
2421 )
2423 # This endpoint accepts an opaque token with no upstream identity validation, so it must be
2424 # closed for identity-bound servers or it becomes a bypass of the token-relay binding check.
2425 registry_server: Final = _manager.get_mcp_server_by_id(server_id)
2426 binding: Final = registry_server.oauth_identity_binding if registry_server else None
2427 if binding is not None and binding.mode == "enforce": 2427 ↛ 2428line 2427 didn't jump to line 2428 because the condition on line 2427 was never true
2428 raise HTTPException(
2429 status_code=status.HTTP_403_FORBIDDEN,
2430 detail={ # mutable-ok: FastAPI exception detail requires a JSON-serializable dictionary
2431 "error": "oauth_identity_binding_enforced",
2432 "error_description": (
2433 "Direct credential storage is disabled for this server: its OAuth identity "
2434 "binding is enforced and this endpoint cannot validate the token's principal. "
2435 "Complete the OAuth flow through the gateway instead."
2436 ),
2437 "server_id": server_id,
2438 "credential_stored": False,
2439 },
2440 )
2441 user_id: Final = user_api_key_dict.user_id or ""
2442 if not user_id: 2442 ↛ 2443line 2442 didn't jump to line 2443 because the condition on line 2442 was never true
2443 raise HTTPException(
2444 status_code=status.HTTP_400_BAD_REQUEST,
2445 detail={"error": "User ID not found in token"},
2446 )
2447 await store_user_oauth_credential(
2448 prisma_client,
2449 user_id,
2450 server_id,
2451 payload.access_token,
2452 refresh_token=payload.refresh_token,
2453 expires_in=payload.expires_in,
2454 scopes=payload.scopes,
2455 )
2456 from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( # noqa: PLC0415
2457 global_mcp_server_manager,
2458 )
2460 await global_mcp_server_manager.invalidate_user_oauth_token_cache(user_id, server_id)
2461 # Read back the persisted record so the response reflects the stored
2462 # expires_at rather than recomputing it here (which could diverge by
2463 # milliseconds or if the storage logic ever adds a grace period).
2464 stored: Final = await get_user_oauth_credential(prisma_client, user_id, server_id)
2465 expires_at: Final[str | None] = stored.get("expires_at") if stored else None
2466 return MCPOAuthUserCredentialStatus(
2467 server_id=server_id,
2468 has_credential=True,
2469 expires_at=expires_at,
2470 is_expired=False,
2471 )
2473 @router.delete(
2474 "/server/{server_id}/oauth-user-credential",
2475 description=(
2476 "Revoke the calling user's stored OAuth2 token for an MCP server. "
2477 "A proxy admin may pass user_id to revoke another user's stored token."
2478 ),
2479 dependencies=[Depends(user_api_key_auth)],
2480 response_model=MCPOAuthUserCredentialStatus,
2481 )
2482 @management_endpoint_wrapper
2483 async def delete_mcp_oauth_user_credential(
2484 server_id: str,
2485 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2486 user_id: Annotated[str | None, Query(min_length=1)] = None,
2487 ):
2488 """Revoke the target user's OAuth2 credential (the caller unless an admin names another user)."""
2489 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2490 target_user_id: Final = _resolve_credential_target_user_id(user_api_key_dict, user_id)
2491 # Only delete if the stored credential is actually an OAuth2 token.
2492 # This prevents accidentally deleting a BYOK credential if one exists
2493 # for the same (user_id, server_id) pair.
2494 cred_to_delete: Final = await get_user_oauth_credential(prisma_client, target_user_id, server_id)
2495 if cred_to_delete is not None:
2496 try:
2497 await delete_user_credential(prisma_client, target_user_id, server_id)
2498 except RecordNotFoundError:
2499 pass # Already gone — treat as a successful delete
2500 from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( # noqa: PLC0415
2501 global_mcp_server_manager,
2502 )
2504 await global_mcp_server_manager.invalidate_user_oauth_token_cache(target_user_id, server_id)
2505 return MCPOAuthUserCredentialStatus(
2506 server_id=server_id,
2507 has_credential=False,
2508 is_expired=False,
2509 )
2511 @router.get(
2512 "/server/{server_id}/oauth-user-credential/status",
2513 description="Check whether the calling user has a stored OAuth2 credential for this MCP server",
2514 dependencies=[Depends(user_api_key_auth)],
2515 response_model=MCPOAuthUserCredentialStatus,
2516 )
2517 @management_endpoint_wrapper
2518 async def get_mcp_oauth_user_credential_status(
2519 server_id: str,
2520 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2521 ):
2522 """Return credential status (has_credential, expiry) without exposing the token."""
2523 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2524 user_id: Final = user_api_key_dict.user_id or ""
2525 if not user_id: 2525 ↛ 2526line 2525 didn't jump to line 2526 because the condition on line 2525 was never true
2526 raise HTTPException(
2527 status_code=status.HTTP_400_BAD_REQUEST,
2528 detail={"error": "User ID not found in token"},
2529 )
2530 cred: Final = await get_user_oauth_credential(prisma_client, user_id, server_id)
2531 if cred is None:
2532 return MCPOAuthUserCredentialStatus(server_id=server_id, has_credential=False, is_expired=False)
2533 expires_at: Final[str | None] = cred.get("expires_at")
2534 is_expired = False
2535 if expires_at: 2535 ↛ 2536line 2535 didn't jump to line 2536 because the condition on line 2535 was never true
2536 try:
2537 exp: Final = datetime.fromisoformat(expires_at)
2538 is_expired = exp < datetime.now(timezone.utc)
2539 except Exception:
2540 pass
2541 return MCPOAuthUserCredentialStatus(
2542 server_id=server_id,
2543 has_credential=True,
2544 expires_at=expires_at,
2545 is_expired=is_expired,
2546 connected_at=cred.get("connected_at"),
2547 )
2549 @router.get(
2550 "/user-credentials",
2551 description="List all OAuth2 MCP credentials stored for the calling user",
2552 dependencies=[Depends(user_api_key_auth)],
2553 response_model=list[MCPUserCredentialListItem],
2554 )
2555 @management_endpoint_wrapper
2556 async def list_mcp_user_credentials(
2557 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2558 ):
2559 """Return all servers the calling user has connected via OAuth2."""
2560 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2561 user_id: Final = user_api_key_dict.user_id or ""
2562 if not user_id: 2562 ↛ 2563line 2562 didn't jump to line 2563 because the condition on line 2562 was never true
2563 raise HTTPException(
2564 status_code=status.HTTP_400_BAD_REQUEST,
2565 detail={"error": "User ID not found in token"},
2566 )
2567 oauth_creds: Final = await list_user_oauth_credentials(prisma_client, user_id)
2568 if not oauth_creds:
2569 return []
2570 # Fetch server metadata for display names — single batch query instead of N+1.
2571 server_ids: Final = [c["server_id"] for c in oauth_creds if "server_id" in c]
2572 servers: Final = {srv.server_id: srv for srv in await get_mcp_servers(prisma_client, server_ids)}
2573 items: Final[list[MCPUserCredentialListItem]] = []
2574 for cred in oauth_creds:
2575 if "server_id" not in cred: 2575 ↛ 2576line 2575 didn't jump to line 2576 because the condition on line 2575 was never true
2576 continue
2577 sid = cred["server_id"]
2578 srv = servers.get(sid)
2579 expires_at: str | None = cred.get("expires_at")
2580 items.append(
2581 MCPUserCredentialListItem(
2582 server_id=sid,
2583 server_name=getattr(srv, "server_name", None) if srv else None,
2584 alias=getattr(srv, "alias", None) if srv else None,
2585 credential_type="oauth2",
2586 has_credential=True,
2587 expires_at=expires_at, # always pass the raw timestamp; client computes expiry state
2588 connected_at=cred.get("connected_at"),
2589 )
2590 )
2591 return items
2593 @router.get(
2594 "/server/{server_id}/user-credentials",
2595 description="List every user's stored BYOK or OAuth2 credential for an MCP server (admin only, no secrets)",
2596 dependencies=(Depends(user_api_key_auth),),
2597 response_model=list[MCPServerUserCredentialListItem],
2598 )
2599 @management_endpoint_wrapper
2600 async def list_mcp_server_user_credentials(
2601 server_id: str,
2602 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
2603 ) -> tuple[MCPServerUserCredentialListItem, ...]:
2604 if user_api_key_dict.user_role not in ( 2604 ↛ 2608line 2604 didn't jump to line 2608 because the condition on line 2604 was never true
2605 LitellmUserRoles.PROXY_ADMIN,
2606 LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
2607 ):
2608 raise HTTPException(
2609 status_code=status.HTTP_403_FORBIDDEN,
2610 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
2611 "error": "Admin access required to view MCP server user credentials.",
2612 },
2613 )
2614 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2615 return await list_server_user_credentials(prisma_client, server_id)
2617 # ── Per-user MCP env var endpoints ────────────────────────────────────────
2619 async def _authorize_and_fetch_mcp_server(
2620 prisma_client,
2621 user_api_key_dict: UserAPIKeyAuth,
2622 server_id: str,
2623 ) -> LiteLLM_MCPServerTable:
2624 """Resolve the MCP server a caller may manage their own per-user state for.
2626 Looks the server up in the DB, then the in-memory registry, so a
2627 config-defined server (which never gets a DB row) resolves too. Admins
2628 may reach any server and get a 404 for an unknown id. A non-admin may
2629 only reach a server in their allowed set and otherwise gets 403 (never
2630 404, so server ids can't be enumerated), using the same allowed-server
2631 resolution the MCP gateway enforces on tool calls.
2632 """
2633 server = await get_mcp_server(prisma_client, server_id)
2634 if server is None:
2635 registry_server: Final = global_mcp_server_manager.get_mcp_server_by_id(server_id)
2636 if registry_server is not None: 2636 ↛ 2637line 2636 didn't jump to line 2637 because the condition on line 2636 was never true
2637 server = global_mcp_server_manager._build_mcp_server_table(registry_server)
2639 if _user_has_admin_view(user_api_key_dict): 2639 ↛ 2647line 2639 didn't jump to line 2647 because the condition on line 2639 was always true
2640 if server is None:
2641 raise HTTPException(
2642 status_code=status.HTTP_404_NOT_FOUND,
2643 detail={"error": f"MCP Server {server_id} not found"},
2644 )
2645 return server
2647 if server is None or not await can_access_mcp_server(
2648 user_api_key_dict,
2649 server.server_id,
2650 global_mcp_server_manager.get_allowed_mcp_servers,
2651 ):
2652 raise HTTPException(
2653 status_code=status.HTTP_403_FORBIDDEN,
2654 detail={
2655 "error": (
2656 f"User does not have permission to access mcp server with id {server_id}. "
2657 "You can only manage mcp servers that you have access to."
2658 )
2659 },
2660 )
2661 return server
2663 def _compute_user_env_var_status(
2664 *,
2665 server: LiteLLM_MCPServerTable,
2666 stored_values: dict[str, str],
2667 ) -> MCPUserEnvVarsStatus:
2668 """Build a status object for one server given the user's stored values.
2670 Stored credentials are write-only: the response reports only whether
2671 each value ``is_set`` and never echoes the decrypted secret back, so a
2672 leaked token can't be used to exfiltrate the raw upstream credential.
2673 """
2674 global_values, user_specs = parse_admin_env_vars(getattr(server, "env_vars", None))
2675 # An empty-valued global is not a usable fallback, so it must not mark a
2676 # referenced per-user var as covered, matching the empty-global filter in
2677 # _resolve_static_headers_with_env_vars. Otherwise this endpoint reports no
2678 # credential needed for a var every tool call still 412s on.
2679 global_values = {name: value for name, value in global_values.items() if value}
2681 # A var only blocks when it's referenced by static_headers and has no
2682 # admin global fallback, mirroring _resolve_static_headers_with_env_vars
2683 # (globals win the merge) so the status endpoint never asks the user for
2684 # credentials a tool call wouldn't actually require.
2685 static_headers = getattr(server, "static_headers", None) or {}
2686 if isinstance(static_headers, str): 2686 ↛ 2687line 2686 didn't jump to line 2687 because the condition on line 2686 was never true
2687 try:
2688 static_headers = json.loads(static_headers) or {}
2689 except (ValueError, TypeError):
2690 static_headers = {}
2691 referenced: Final = collect_env_var_references(strings=static_headers.values())
2692 user_var_names: Final = {spec["name"] for spec in user_specs}
2693 blocking: Final = {name for name in (referenced & user_var_names) if name not in global_values}
2695 required: Final[list[MCPUserEnvVarSpec]] = []
2696 missing_count = 0
2697 for spec in user_specs: 2697 ↛ 2698line 2697 didn't jump to line 2698 because the loop on line 2697 never started
2698 name: str = spec["name"]
2699 if name not in blocking:
2700 continue
2701 value = stored_values.get(name)
2702 is_set = bool(value)
2703 if not is_set:
2704 missing_count += 1
2705 required.append(
2706 MCPUserEnvVarSpec(
2707 name=name,
2708 description=spec.get("description"),
2709 is_set=is_set,
2710 )
2711 )
2713 return MCPUserEnvVarsStatus(
2714 server_id=server.server_id,
2715 server_name=getattr(server, "server_name", None),
2716 alias=getattr(server, "alias", None),
2717 required=required,
2718 missing_count=missing_count,
2719 setup_url=build_env_var_setup_url(server.server_id) if required else None,
2720 )
2722 @router.get(
2723 "/server/{server_id}/user-env-vars",
2724 description="Return the calling user's per-user MCP env var status for this server.",
2725 dependencies=[Depends(user_api_key_auth)],
2726 response_model=MCPUserEnvVarsStatus,
2727 )
2728 @management_endpoint_wrapper
2729 async def get_mcp_user_env_vars(
2730 server_id: str,
2731 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2732 ) -> MCPUserEnvVarsStatus:
2733 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2734 user_id: Final = user_api_key_dict.user_id or ""
2735 if not user_id: 2735 ↛ 2736line 2735 didn't jump to line 2736 because the condition on line 2735 was never true
2736 raise HTTPException(
2737 status_code=status.HTTP_400_BAD_REQUEST,
2738 detail={"error": "User ID not found in token"},
2739 )
2740 server: Final = await _authorize_and_fetch_mcp_server(prisma_client, user_api_key_dict, server_id)
2741 stored: Final = await get_user_env_vars(prisma_client, user_id, server_id)
2742 return _compute_user_env_var_status(server=server, stored_values=stored)
2744 @router.post(
2745 "/server/{server_id}/user-env-vars",
2746 description=(
2747 "Store the calling user's per-user MCP env var values for this "
2748 "server. Submitted values are merged over any previously stored "
2749 "values, so you only send the fields you want to set or change; a "
2750 "variable omitted (or sent empty) keeps its stored value. Use "
2751 "DELETE to clear all stored values."
2752 ),
2753 dependencies=[Depends(user_api_key_auth)],
2754 response_model=MCPUserEnvVarsStatus,
2755 )
2756 @management_endpoint_wrapper
2757 async def store_mcp_user_env_vars(
2758 server_id: str,
2759 payload: MCPUserEnvVarsRequest,
2760 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2761 ) -> MCPUserEnvVarsStatus:
2762 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2763 user_id: Final = user_api_key_dict.user_id or ""
2764 if not user_id: 2764 ↛ 2765line 2764 didn't jump to line 2765 because the condition on line 2764 was never true
2765 raise HTTPException(
2766 status_code=status.HTTP_400_BAD_REQUEST,
2767 detail={"error": "User ID not found in token"},
2768 )
2769 server: Final = await _authorize_and_fetch_mcp_server(prisma_client, user_api_key_dict, server_id)
2770 # Only known per-user var names declared by the admin are accepted —
2771 # never persist arbitrary keys the user invents. Submitted values are
2772 # merged over the existing set so a user updating one credential does
2773 # not have to re-enter the others (which are write-only and never shown
2774 # back); an omitted/empty field keeps its stored value.
2775 _, user_specs = parse_admin_env_vars(getattr(server, "env_vars", None))
2776 allowed_names: Final = {spec["name"] for spec in user_specs}
2777 updates: Final = {k: v for k, v in payload.values.items() if k in allowed_names and v != ""}
2778 merged: Final = await merge_user_env_vars(prisma_client, user_id, server_id, updates, allowed_names)
2779 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
2780 invalidate_user_env_vars_cache,
2781 )
2783 invalidate_user_env_vars_cache(user_id, server_id)
2784 return _compute_user_env_var_status(server=server, stored_values=merged)
2786 @router.delete(
2787 "/server/{server_id}/user-env-vars",
2788 description="Clear the calling user's per-user MCP env var values for this server.",
2789 dependencies=[Depends(user_api_key_auth)],
2790 response_model=MCPUserEnvVarsStatus,
2791 )
2792 @management_endpoint_wrapper
2793 async def clear_mcp_user_env_vars(
2794 server_id: str,
2795 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2796 ) -> MCPUserEnvVarsStatus:
2797 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2798 user_id: Final = user_api_key_dict.user_id or ""
2799 if not user_id: 2799 ↛ 2800line 2799 didn't jump to line 2800 because the condition on line 2799 was never true
2800 raise HTTPException(
2801 status_code=status.HTTP_400_BAD_REQUEST,
2802 detail={"error": "User ID not found in token"},
2803 )
2804 server: Final = await _authorize_and_fetch_mcp_server(prisma_client, user_api_key_dict, server_id)
2805 await delete_user_env_vars(prisma_client, user_id, server_id)
2806 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
2807 invalidate_user_env_vars_cache,
2808 )
2810 invalidate_user_env_vars_cache(user_id, server_id)
2811 return _compute_user_env_var_status(server=server, stored_values={})
2813 @router.get(
2814 "/user-env-vars/status",
2815 description="Per-user MCP env var status across every server the user can access. "
2816 "Used by the dashboard to highlight servers with missing per-user vars.",
2817 dependencies=[Depends(user_api_key_auth)],
2818 response_model=list[MCPUserEnvVarsStatus],
2819 )
2820 @management_endpoint_wrapper
2821 async def list_mcp_user_env_var_status(
2822 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2823 ) -> list[MCPUserEnvVarsStatus]:
2824 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
2825 user_id: Final = user_api_key_dict.user_id or ""
2826 if not user_id: 2826 ↛ 2827line 2826 didn't jump to line 2827 because the condition on line 2826 was never true
2827 return []
2828 accessible: Final = await _resolve_accessible_mcp_servers(user_api_key_dict)
2829 if not accessible:
2830 return []
2831 server_ids: Final = [s.server_id for s in accessible]
2832 stored_bulk: Final = await get_user_env_vars_bulk(prisma_client, user_id, server_ids)
2833 statuses: Final[list[MCPUserEnvVarsStatus]] = []
2834 for server in accessible:
2835 stored = stored_bulk.get(server.server_id, {})
2836 status_obj = _compute_user_env_var_status(server=server, stored_values=stored)
2837 if status_obj.required: 2837 ↛ 2838line 2837 didn't jump to line 2838 because the condition on line 2837 was never true
2838 statuses.append(status_obj)
2839 return statuses
2841 @router.put(
2842 "/server",
2843 description="Allows deleting mcp serves in the db",
2844 dependencies=[Depends(user_api_key_auth)],
2845 response_model=LiteLLM_MCPServerTable,
2846 status_code=status.HTTP_202_ACCEPTED,
2847 )
2848 @management_endpoint_wrapper
2849 async def edit_mcp_server(
2850 payload: UpdateMCPServerRequest,
2851 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
2852 litellm_changed_by: str | None = Header(
2853 None,
2854 description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
2855 ),
2856 ):
2857 """
2858 Updates the MCP Server in the db.
2860 Partial update: a field left out of the payload keeps its stored value, and a field sent as null is cleared.
2862 Parameters:
2863 - payload: UpdateMCPServerRequest - Required. The updated mcp server data.
2864 ```
2865 curl -X "PUT" --location 'http://localhost:4000/v1/mcp/server' \
2866 --header 'Authorization: Bearer your_api_key_here'
2867 ```
2868 """
2869 prisma_client: Final = get_prisma_client_or_throw(
2870 "Database not connected. Connect a database to your proxy - https://docs.litellm.ai/docs/simple_proxy#managing-auth---virtual-keys"
2871 )
2873 payload_fields_set: Final = set(payload.fields_set())
2875 # Validate and normalize payload fields
2876 validate_and_normalize_mcp_server_payload(payload)
2878 # Authz - restrict only admins to delete mcp servers
2879 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 2879 ↛ 2880line 2879 didn't jump to line 2880 because the condition on line 2879 was never true
2880 raise HTTPException(
2881 status_code=status.HTTP_403_FORBIDDEN,
2882 detail={
2883 "error": "Call not allowed to update MCP server. User is not a proxy admin. route={}".format(
2884 "PUT /v1/mcp/server"
2885 )
2886 },
2887 )
2889 # Snapshot the pre-update identity so we can detect a mint-relevant change below. The read is
2890 # advisory (it only feeds the stale-token purge decision), so a failure skips the purge with a
2891 # warning instead of failing the edit, whose primary job is the update itself.
2892 try:
2893 old_server_record = await get_mcp_server(prisma_client, payload.server_id)
2894 old_server_record_read_failed = False
2895 except Exception as exc: # noqa: BLE001 - advisory read; invalidation is best-effort end-to-end
2896 verbose_logger.warning(
2897 "MCP server %s: could not snapshot the pre-update record; skipping the stale-token check: %s",
2898 payload.server_id,
2899 exc,
2900 )
2901 old_server_record = None
2902 old_server_record_read_failed = True
2904 if payload.per_server_oauth_discovery and (old_server_record is not None or old_server_record_read_failed): 2904 ↛ 2905line 2904 didn't jump to line 2905 because the condition on line 2904 was never true
2905 relay_eligible: Final = old_server_record is not None and is_per_server_oauth_discovery_eligible(
2906 payload.auth_type if "auth_type" in payload_fields_set else old_server_record.auth_type,
2907 payload.oauth2_flow if "oauth2_flow" in payload_fields_set else old_server_record.oauth2_flow,
2908 (
2909 payload.delegate_auth_to_upstream
2910 if "delegate_auth_to_upstream" in payload_fields_set
2911 else old_server_record.delegate_auth_to_upstream
2912 ),
2913 )
2914 if not relay_eligible:
2915 raise HTTPException(
2916 status_code=status.HTTP_400_BAD_REQUEST,
2917 detail={ # mutable-ok: FastAPI HTTPException detail requires a plain dict
2918 "error": (
2919 "per_server_oauth_discovery is only supported for auth_type oauth2 with oauth2_flow "
2920 "authorization_code and without delegate_auth_to_upstream."
2921 )
2922 },
2923 )
2925 if ( 2925 ↛ 2930line 2925 didn't jump to line 2930 because the condition on line 2925 was never true
2926 payload.dcr_bridge
2927 and payload.auth_type is None
2928 and (old_server_record is not None or old_server_record_read_failed)
2929 ):
2930 stored_auth_type: Final = old_server_record.auth_type if old_server_record else None
2931 stored_auth_type_name: Final = getattr(stored_auth_type, "value", stored_auth_type)
2932 if stored_auth_type not in (MCPAuth.true_passthrough, MCPAuth.oauth_delegate):
2933 raise HTTPException(
2934 status_code=status.HTTP_400_BAD_REQUEST,
2935 detail={
2936 "error": (
2937 "dcr_bridge is only supported for auth_type true_passthrough or "
2938 f"oauth_delegate (stored auth_type: {stored_auth_type_name!r}). Include "
2939 "the server's auth_type in the update payload or configure one of the "
2940 "client-forwarded token modes first."
2941 )
2942 },
2943 )
2945 # try to update the mcp server
2946 mcp_server_record_updated: Final = await update_mcp_server(
2947 prisma_client,
2948 payload,
2949 touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
2950 fields_set=payload_fields_set,
2951 )
2953 if isinstance(mcp_server_record_updated, McpIdentifierConflict): 2953 ↛ 2954line 2953 didn't jump to line 2954 because the condition on line 2953 was never true
2954 raise_mcp_identifier_conflict(mcp_server_record_updated)
2956 if mcp_server_record_updated is None:
2957 raise HTTPException(
2958 status_code=status.HTTP_404_NOT_FOUND,
2959 detail={"error": f"MCP Server not found, passed server_id={payload.server_id}"},
2960 )
2961 warn_if_id_jag_server_outruns_sso(mcp_server_record_updated.server_id, mcp_server_record_updated.auth_type)
2962 await global_mcp_server_manager.update_server(mcp_server_record_updated)
2964 # Ensure registry is up to date by reloading from database
2965 await global_mcp_server_manager.reload_servers_from_database()
2967 # If a field that determines which upstream OAuth token gets minted changed (url/audience, OAuth
2968 # mode/grant, authorization-server endpoints, or the OAuth client + scopes), every stored per-user
2969 # token was minted for the old configuration and is stale. Purge them (DB + cache) so the next
2970 # tool call re-authorizes instead of forwarding a token for a resource/AS/client that no longer
2971 # matches. Best-effort: a purge failure must not fail the update, whose primary job already
2972 # succeeded.
2973 if old_server_record is not None and mcp_oauth_token_identity(old_server_record) != mcp_oauth_token_identity(
2974 mcp_server_record_updated
2975 ):
2976 try:
2977 purged: Final = await purge_user_oauth_credentials_for_server(prisma_client, payload.server_id)
2978 if purged: 2978 ↛ 2979line 2978 didn't jump to line 2979 because the condition on line 2978 was never true
2979 verbose_logger.info(
2980 "MCP server %s: purged %d stale per-user OAuth token(s) after a mint-relevant config change",
2981 payload.server_id,
2982 purged,
2983 )
2984 except Exception as exc: # noqa: BLE001 - purge is best-effort; the server update already succeeded
2985 verbose_logger.warning(
2986 "MCP server %s: failed to purge stale per-user OAuth tokens after config change: %s",
2987 payload.server_id,
2988 exc,
2989 )
2991 # TODO: Enterprise: Finish audit log trail
2992 if is_audit_logging_enabled(): 2992 ↛ 2993line 2992 didn't jump to line 2993 because the condition on line 2992 was never true
2993 pass
2995 return _redact_mcp_credentials(mcp_server_record_updated)
2997 @router.post(
2998 "/make_public",
2999 description="Allows making MCP servers public for AI Hub",
3000 dependencies=[Depends(user_api_key_auth)],
3001 status_code=status.HTTP_202_ACCEPTED,
3002 )
3003 async def make_mcp_servers_public(
3004 request: MakeMCPServersPublicRequest,
3005 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3006 ):
3007 """
3008 Make MCP servers public for AI Hub
3009 """
3010 try:
3011 # Update the public model groups
3012 import litellm
3013 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
3014 global_mcp_server_manager,
3015 )
3016 from litellm.proxy.proxy_server import proxy_config
3018 # Load existing config
3019 config: Final = await proxy_config.get_config()
3020 # Check if user has admin permissions
3021 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 3021 ↛ 3022line 3021 didn't jump to line 3022 because the condition on line 3021 was never true
3022 raise HTTPException(
3023 status_code=403,
3024 detail={
3025 "error": f"Only proxy admins can update public mcp servers. Your role={user_api_key_dict.user_role}"
3026 },
3027 )
3029 if litellm.public_mcp_servers is None:
3030 litellm.public_mcp_servers = []
3032 for server_id in request.mcp_server_ids:
3033 server = global_mcp_server_manager.get_mcp_server_by_id(server_id=server_id)
3034 if server is None: 3034 ↛ 3032line 3034 didn't jump to line 3032 because the condition on line 3034 was always true
3035 raise HTTPException(
3036 status_code=404,
3037 detail=f"MCP Server with ID {server_id} not found",
3038 )
3040 litellm.public_mcp_servers = request.mcp_server_ids
3042 # Update config with new settings
3043 if "litellm_settings" not in config or config["litellm_settings"] is None: 3043 ↛ 3044line 3043 didn't jump to line 3044 because the condition on line 3043 was never true
3044 config["litellm_settings"] = {}
3046 config["litellm_settings"]["public_mcp_servers"] = litellm.public_mcp_servers
3048 # Save the updated config
3049 await proxy_config.save_config(new_config=config)
3051 verbose_proxy_logger.debug(
3052 "Updated public mcp servers to: %s by user: %s", litellm.public_mcp_servers, user_api_key_dict.user_id
3053 )
3055 return {
3056 "message": "Successfully updated public mcp servers",
3057 "public_mcp_servers": litellm.public_mcp_servers,
3058 "updated_by": user_api_key_dict.user_id,
3059 }
3060 except HTTPException:
3061 raise
3062 except Exception as e:
3063 verbose_proxy_logger.exception("Error making agent public: %s", e)
3064 raise HTTPException(status_code=500, detail=str(e))
3066 # --- MCP Discovery ---
3068 _MCP_REGISTRY_PATH: Final = os.path.join(
3069 os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
3070 "mcp_registry.json",
3071 )
3073 _mcp_registry_cache: Mapping[str, Sequence[Mapping[str, str]]] | None = None
3075 def _load_mcp_registry() -> Mapping[str, Sequence[Mapping[str, str]]]:
3076 """Load the curated MCP registry from disk. Cached after first read."""
3077 global _mcp_registry_cache
3078 if _mcp_registry_cache is not None:
3079 return _mcp_registry_cache
3080 try:
3081 with open(_MCP_REGISTRY_PATH, "r") as f:
3082 data: Mapping[str, Sequence[Mapping[str, str]]] = json.load(f)
3083 except Exception as e:
3084 verbose_proxy_logger.warning("Failed to load MCP registry from %s: %s", _MCP_REGISTRY_PATH, e)
3085 data = {"servers": []}
3086 _mcp_registry_cache = data
3087 return data
3089 @router.get(
3090 "/discover",
3091 description="Returns a curated list of well-known MCP servers for discovery UI",
3092 dependencies=[Depends(user_api_key_auth)],
3093 )
3094 async def discover_mcp_servers(
3095 query: str | None = Query(None, description="Search filter for server names and descriptions"),
3096 category: str | None = Query(None, description="Filter by category"),
3097 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3098 ):
3099 """
3100 Returns a curated list of well-known MCP servers that can be added to the proxy.
3102 Used by the UI to show a discovery grid when adding new MCP servers.
3103 """
3104 # Admin Viewer follows the read-parity rule.
3105 if not _user_has_admin_view(user_api_key_dict): 3105 ↛ 3106line 3105 didn't jump to line 3106 because the condition on line 3105 was never true
3106 raise HTTPException(
3107 status_code=403,
3108 detail={
3109 "error": f"Only proxy admins can access MCP discovery. Your role={user_api_key_dict.user_role}"
3110 },
3111 )
3113 registry: Final = _load_mcp_registry()
3114 servers = registry.get("servers", [])
3116 # Apply query filter
3117 if query:
3118 query_lower: Final = query.lower()
3119 servers = [
3120 s
3121 for s in servers
3122 if query_lower in s.get("name", "").lower()
3123 or query_lower in s.get("title", "").lower()
3124 or query_lower in s.get("description", "").lower()
3125 ]
3127 # Apply category filter
3128 if category:
3129 servers = [s for s in servers if s.get("category", "") == category]
3131 # Extract unique categories from the full list (before filtering)
3132 all_servers: Final = registry.get("servers", [])
3133 categories: Final = sorted(set(s.get("category", "Other") for s in all_servers))
3135 return {
3136 "servers": servers,
3137 "categories": categories,
3138 }
3140 # --- OpenAPI Registry ---
3142 _OPENAPI_REGISTRY_PATH: Final = os.path.join(
3143 os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
3144 "openapi_registry.json",
3145 )
3147 @functools.lru_cache(maxsize=1)
3148 def _load_openapi_registry() -> dict[str, object]:
3149 with open(_OPENAPI_REGISTRY_PATH, "r") as f:
3150 data: Final[dict[str, object]] = json.load(f)
3151 return data
3153 @router.get(
3154 "/openapi-registry",
3155 description="Returns well-known OpenAPI APIs with OAuth 2.0 metadata for the OpenAPI MCP picker",
3156 )
3157 async def get_openapi_registry(
3158 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3159 ):
3160 # Admin Viewer follows the read-parity rule.
3161 if not _user_has_admin_view(user_api_key_dict): 3161 ↛ 3162line 3161 didn't jump to line 3162 because the condition on line 3161 was never true
3162 raise HTTPException(
3163 status_code=403,
3164 detail={
3165 "error": f"Only proxy admins can access the OpenAPI registry. Your role={user_api_key_dict.user_role}"
3166 },
3167 )
3168 try:
3169 return _load_openapi_registry()
3170 except Exception as e:
3171 verbose_proxy_logger.warning("Failed to load OpenAPI registry from %s: %s", _OPENAPI_REGISTRY_PATH, e)
3172 return {"apis": []}
3174 # ---------------------------------------------------------------------------
3175 # MCP Toolset endpoints
3176 # ---------------------------------------------------------------------------
3178 from litellm.proxy._experimental.mcp_server.toolset_db import (
3179 create_mcp_toolset,
3180 delete_mcp_toolset,
3181 get_mcp_toolset,
3182 list_mcp_toolsets,
3183 update_mcp_toolset,
3184 )
3185 from litellm.types.mcp_server.mcp_toolset import (
3186 NewMCPToolsetRequest,
3187 UpdateMCPToolsetRequest,
3188 )
3190 @router.post(
3191 "/toolset",
3192 description="Create a new MCP toolset (admin only)",
3193 status_code=status.HTTP_201_CREATED,
3194 )
3195 @management_endpoint_wrapper
3196 async def add_mcp_toolset(
3197 payload: NewMCPToolsetRequest,
3198 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3199 litellm_changed_by: str | None = Header(None),
3200 ):
3201 """Create a named toolset — a curated selection of {server_id, tool_name} pairs."""
3202 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
3203 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 3203 ↛ 3204line 3203 didn't jump to line 3204 because the condition on line 3203 was never true
3204 raise HTTPException(
3205 status_code=status.HTTP_403_FORBIDDEN,
3206 detail={"error": "Only proxy admins can create MCP toolsets."},
3207 )
3208 touched_by: Final = (
3209 get_audit_log_changed_by(
3210 litellm_changed_by=litellm_changed_by,
3211 user_api_key_dict=user_api_key_dict,
3212 litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME,
3213 )
3214 or LITELLM_PROXY_ADMIN_NAME
3215 )
3216 try:
3217 result: Final = await create_mcp_toolset(prisma_client, payload, touched_by)
3218 except UniqueViolationError:
3219 raise HTTPException(
3220 status_code=status.HTTP_409_CONFLICT,
3221 detail={"error": f"A toolset named '{payload.toolset_name}' already exists."},
3222 )
3223 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
3224 global_mcp_server_manager,
3225 )
3227 global_mcp_server_manager.invalidate_toolset_cache()
3228 return result
3230 @router.get(
3231 "/toolset",
3232 description="List MCP toolsets accessible to the calling key",
3233 )
3234 @management_endpoint_wrapper
3235 async def fetch_mcp_toolsets(
3236 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3237 ):
3238 """Return toolsets the calling key is allowed to access."""
3239 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
3240 is_admin: Final = _user_has_admin_view(user_api_key_dict)
3241 op: Final = user_api_key_dict.object_permission
3242 # mcp_toolsets=None or [] both mean "not restricted by toolsets".
3243 # For admins: either value → no restriction → return all.
3244 # For non-admins: either value → no toolsets explicitly granted → return nothing.
3245 # (An admin whose DB row has mcp_toolsets=[] should still see all toolsets.)
3246 raw_toolsets: Final = getattr(op, "mcp_toolsets", None) if op else None
3247 if not raw_toolsets: 3247 ↛ 3251line 3247 didn't jump to line 3251 because the condition on line 3247 was always true
3248 if is_admin: 3248 ↛ 3250line 3248 didn't jump to line 3250 because the condition on line 3248 was always true
3249 return await list_mcp_toolsets(prisma_client)
3250 return []
3251 return await list_mcp_toolsets(prisma_client, toolset_ids=raw_toolsets)
3253 @router.get(
3254 "/toolset/{toolset_id}",
3255 description="Get a specific MCP toolset by ID",
3256 )
3257 @management_endpoint_wrapper
3258 async def fetch_mcp_toolset(
3259 toolset_id: str,
3260 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3261 ):
3262 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
3263 # Non-admin keys may only fetch toolsets they've been explicitly granted.
3264 if not _user_has_admin_view(user_api_key_dict): 3264 ↛ 3265line 3264 didn't jump to line 3265 because the condition on line 3264 was never true
3265 op: Final = user_api_key_dict.object_permission
3266 granted: Final = getattr(op, "mcp_toolsets", None) if op else None
3267 if granted is None or toolset_id not in granted:
3268 raise HTTPException(
3269 status_code=status.HTTP_403_FORBIDDEN,
3270 detail={"error": "API key does not have access to this toolset."},
3271 )
3272 toolset: Final = await get_mcp_toolset(prisma_client, toolset_id)
3273 if toolset is None:
3274 raise HTTPException(
3275 status_code=status.HTTP_404_NOT_FOUND,
3276 detail={"error": f"Toolset '{toolset_id}' not found."},
3277 )
3278 return toolset
3280 @router.put(
3281 "/toolset",
3282 description="Update an existing MCP toolset (admin only)",
3283 )
3284 @management_endpoint_wrapper
3285 async def edit_mcp_toolset(
3286 payload: UpdateMCPToolsetRequest,
3287 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3288 litellm_changed_by: str | None = Header(None),
3289 ):
3290 """Partial update: a field left out keeps its stored value, and a field sent as null is cleared, except
3291 ``toolset_name`` and ``tools``, which a toolset always has; empty the tool selection with an explicit []."""
3292 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
3293 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 3293 ↛ 3294line 3293 didn't jump to line 3294 because the condition on line 3293 was never true
3294 raise HTTPException(
3295 status_code=status.HTTP_403_FORBIDDEN,
3296 detail={"error": "Only proxy admins can update MCP toolsets."},
3297 )
3298 touched_by: Final = (
3299 get_audit_log_changed_by(
3300 litellm_changed_by=litellm_changed_by,
3301 user_api_key_dict=user_api_key_dict,
3302 litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME,
3303 )
3304 or LITELLM_PROXY_ADMIN_NAME
3305 )
3306 try:
3307 result: Final = await update_mcp_toolset(prisma_client, payload, touched_by)
3308 except UniqueViolationError:
3309 raise HTTPException(
3310 status_code=status.HTTP_409_CONFLICT,
3311 detail={
3312 "error": (
3313 f"A toolset named '{payload.toolset_name}' already exists."
3314 if payload.toolset_name
3315 else "A toolset with that name already exists."
3316 )
3317 },
3318 )
3319 if result is None:
3320 raise HTTPException(
3321 status_code=status.HTTP_404_NOT_FOUND,
3322 detail={"error": f"Toolset '{payload.toolset_id}' not found."},
3323 )
3324 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
3325 global_mcp_server_manager,
3326 )
3328 global_mcp_server_manager.invalidate_toolset_cache(getattr(payload, "toolset_id", None))
3329 return result
3331 @router.delete(
3332 "/toolset/{toolset_id}",
3333 description="Delete an MCP toolset (admin only)",
3334 status_code=status.HTTP_202_ACCEPTED,
3335 )
3336 @management_endpoint_wrapper
3337 async def remove_mcp_toolset(
3338 toolset_id: str,
3339 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
3340 litellm_changed_by: str | None = Header(None),
3341 ):
3342 prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
3343 if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role: 3343 ↛ 3344line 3343 didn't jump to line 3344 because the condition on line 3343 was never true
3344 raise HTTPException(
3345 status_code=status.HTTP_403_FORBIDDEN,
3346 detail={"error": "Only proxy admins can delete MCP toolsets."},
3347 )
3348 deleted: Final = await delete_mcp_toolset(prisma_client, toolset_id)
3349 if deleted is None: 3349 ↛ 3350line 3349 didn't jump to line 3350 because the condition on line 3349 was never true
3350 raise HTTPException(
3351 status_code=status.HTTP_404_NOT_FOUND,
3352 detail={"error": f"Toolset '{toolset_id}' not found."},
3353 )
3354 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
3355 global_mcp_server_manager,
3356 )
3358 global_mcp_server_manager.invalidate_toolset_cache(toolset_id)
3359 return Response(status_code=status.HTTP_202_ACCEPTED)