import re import traceback from collections.abc import Iterable, Mapping, Sequence from datetime import datetime from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypeAlias, TypedDict, overload from openai.types.chat import ChatCompletionToolParam from openai.types.responses.function_tool_param import FunctionToolParam from litellm._logging import verbose_logger from litellm.constants import MAXIMUM_TRACEBACK_LINES_TO_LOG from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.proxy._experimental.mcp_server.utils import ( logging_safe_mcp_headers, split_server_prefix_from_name, strip_known_server_prefix, ) from litellm.responses.main import aresponses from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator from litellm.types.llms.openai import ( ResponseInputParam, ResponsesAPIResponse, ResponsesAPIStreamingResponse, ) from litellm.types.llms.openai import ToolParam as ResponsesToolParam from litellm.types.utils import ( CallTypes, ChatCompletionMessageCustomToolCall, ChatCompletionMessageToolCall, Choices, Message, ModelResponse, StandardLoggingMCPToolCall, ) from litellm.utils import Rules, function_setup if TYPE_CHECKING: from mcp.types import CallToolResult from mcp.types import Tool as MCPTool from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.utils import ProxyLogging else: MCPTool = Any # NOTE: We intentionally keep ToolParam as a broad type here to avoid tight coupling ToolParam: TypeAlias = Mapping[str, object] class MCPToolResult(TypedDict): tool_call_id: str | None result: str name: str | None LITELLM_PROXY_MCP_SERVER_URL: Final = "litellm_proxy" LITELLM_PROXY_MCP_SERVER_URL_PREFIX: Final = f"{LITELLM_PROXY_MCP_SERVER_URL}/mcp/" # Matches any URL whose path ends with /mcp/ — covers both root-path # (http://host:port/mcp/name) and sub-path (http://host/base/mcp/name) proxy deployments. # A false-positive match (e.g. an external URL that happens to end with /mcp/) results # in a "server not found" error from the internal gateway, not a silent failure or data leak, # so this broad pattern is intentional and preferred over anchoring to localhost only. _PROXY_MCP_PATH_RE: Final = re.compile(r"^https?://.+/mcp/([^/]+)$") class LiteLLM_Proxy_MCP_Handler: """ Helper class with static methods for MCP integration with Responses API. This handles when a user passes mcp server_url="litellm_proxy" in their tools. """ @staticmethod def _get_parent_request_tags(kwargs: dict[str, Any] | None) -> list[str]: """Tags from the parent LLM request, using the same extraction logic as standard logging (incl. User-Agent).""" if not kwargs: return [] from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup litellm_params: Final = kwargs.get("litellm_params") or kwargs proxy_server_request = litellm_params.get("proxy_server_request") or kwargs.get("proxy_server_request") or {} return StandardLoggingPayloadSetup._get_request_tags( litellm_params=litellm_params, proxy_server_request=proxy_server_request, ) @staticmethod def _should_use_litellm_mcp_gateway(tools: Iterable[ToolParam] | None) -> bool: """ Returns True if any MCP tool should be handled via the litellm proxy MCP gateway. This includes tools with server_url="litellm_proxy" as well as URLs ending in /mcp/. """ if tools: for tool in tools: if isinstance(tool, dict) and tool.get("type") == "mcp": server_url = tool.get("server_url", "") if isinstance(server_url, str) and server_url.startswith(LITELLM_PROXY_MCP_SERVER_URL): return True if isinstance(server_url, str) and _PROXY_MCP_PATH_RE.match(server_url): return True return False @staticmethod def _parse_mcp_tools( tools: Iterable[Mapping[str, object]] | None, ) -> tuple[list[ToolParam], list[Any]]: """ Parse tools and separate MCP tools with litellm_proxy from other tools. Returns: Tuple of (mcp_tools_with_litellm_proxy, other_tools) """ mcp_tools_with_litellm_proxy: Final[list[ToolParam]] = [] other_tools: Final[list[Any]] = [] if tools: for tool in tools: if isinstance(tool, dict) and tool.get("type") == "mcp": server_url = tool.get("server_url", "") if isinstance(server_url, str) and server_url.startswith(LITELLM_PROXY_MCP_SERVER_URL): mcp_tools_with_litellm_proxy.append(tool) elif isinstance(server_url, str): # Also intercept URLs like http://localhost:4000/mcp/atlassian_test # by rewriting them to the internal litellm_proxy format. m = _PROXY_MCP_PATH_RE.match(server_url) if m: rewritten = { **tool, "server_url": f"{LITELLM_PROXY_MCP_SERVER_URL_PREFIX}{m.group(1)}", } mcp_tools_with_litellm_proxy.append(rewritten) else: other_tools.append(tool) else: other_tools.append(tool) else: other_tools.append(tool) return mcp_tools_with_litellm_proxy, other_tools @staticmethod async def _apply_toolset_permissions( resolved_toolset_ids: list[str], resolved_mcp_servers: list[str], user_api_key_auth: "UserAPIKeyAuth", ) -> "UserAPIKeyAuth": """Apply resolved toolset permissions to user_api_key_auth and return updated auth.""" from litellm.proxy._types import LiteLLM_ObjectPermissionTable try: from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( global_mcp_server_manager, ) tool_permissions: Final = await global_mcp_server_manager.resolve_toolset_tool_permissions( toolset_ids=resolved_toolset_ids ) all_server_ids: Final = list(set(tool_permissions.keys()) | set(resolved_mcp_servers)) existing_op: Final = user_api_key_auth.object_permission if existing_op is not None: merged_tool_perms: Final = dict(existing_op.mcp_tool_permissions or {}) for server_id, tool_names in tool_permissions.items(): existing_tools = merged_tool_perms.get(server_id, []) merged_tool_perms[server_id] = list(set(existing_tools) | set(tool_names)) updated_op = existing_op.model_copy( update={ "mcp_servers": all_server_ids, "mcp_tool_permissions": merged_tool_perms, "mcp_toolsets": [], } ) else: updated_op = LiteLLM_ObjectPermissionTable( object_permission_id="toolset-scope", mcp_servers=all_server_ids, mcp_tool_permissions=tool_permissions, ) return user_api_key_auth.model_copy(update={"object_permission": updated_op}) except Exception as _e: verbose_logger.debug("Could not apply toolset permissions: %s", _e) return user_api_key_auth @staticmethod async def _get_mcp_tools_from_manager( user_api_key_auth: "UserAPIKeyAuth | None", mcp_tools_with_litellm_proxy: Iterable[Mapping[str, object]] | None, litellm_trace_id: str | None = None, mcp_auth_header: str | None = None, mcp_server_auth_headers: dict[str, dict[str, str]] | None = None, request_tags: list[str] | None = None, ) -> tuple[list[MCPTool], list[str]]: """ Get available tools from the MCP server manager. Args: user_api_key_auth: User authentication info for access control mcp_tools_with_litellm_proxy: ToolParam objects with server_url starting with "litellm_proxy" mcp_auth_header: Optional deprecated auth header for MCP servers mcp_server_auth_headers: Optional server-specific auth headers (e.g. from x-mcp-{alias}-*) Returns: List of MCP tools List names of allowed MCP servers """ from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( global_mcp_server_manager, ) from litellm.proxy._experimental.mcp_server.server import ( _get_allowed_mcp_servers_from_mcp_server_names, _get_tools_from_mcp_servers, ) mcp_servers: Final = [ server_url.split("/")[-1] for _tool in (mcp_tools_with_litellm_proxy or ()) for server_url in (_tool.get("server_url", "") if isinstance(_tool, dict) else "",) if isinstance(server_url, str) and server_url.startswith(LITELLM_PROXY_MCP_SERVER_URL_PREFIX) ] # Resolve toolset names: collect all toolset IDs first, then apply their # combined permissions in a single pass so multiple toolsets are unioned # rather than the last one overwriting the others. resolved_mcp_servers: Final[list[str]] = [] resolved_toolset_ids: Final[list[str]] = [] for name in mcp_servers: if not global_mcp_server_manager.get_mcp_server_by_name(name): try: from litellm.proxy.proxy_server import prisma_client if prisma_client is not None: toolset = await global_mcp_server_manager.get_toolset_by_name_cached(prisma_client, name) if toolset is not None: # Access control: only allow if the key explicitly grants this toolset. if user_api_key_auth is not None: from litellm.proxy.management_endpoints.common_utils import ( _user_has_admin_view, ) is_admin = _user_has_admin_view(user_api_key_auth) if not is_admin: op = user_api_key_auth.object_permission granted = getattr(op, "mcp_toolsets", None) if op else None # None means no grants configured → deny (consistent with # fetch_mcp_toolsets which returns [] for unconfigured keys) if granted is None or toolset.toolset_id not in granted: verbose_logger.debug( "Key does not have access to toolset '%s', skipping.", name ) continue resolved_toolset_ids.append(toolset.toolset_id) # Don't add to resolved_mcp_servers — toolset scope # restricts via object_permission, not server name filter. continue except Exception as _e: verbose_logger.debug("Could not resolve '%s' as toolset: %s", name, _e) resolved_mcp_servers.append(name) # Apply all resolved toolsets at once (union), avoiding permission overwrite. if resolved_toolset_ids and user_api_key_auth is not None: user_api_key_auth = await LiteLLM_Proxy_MCP_Handler._apply_toolset_permissions( resolved_toolset_ids=resolved_toolset_ids, resolved_mcp_servers=resolved_mcp_servers, user_api_key_auth=user_api_key_auth, ) # When toolsets were resolved we updated object_permission.mcp_servers to the # full union (toolset server IDs + direct server names). Passing a name-based # filter here would exclude those toolset server IDs (which are UUIDs, not # names), so use None and let the auth object's mcp_servers do the filtering. effective_server_filter: Final = None if resolved_toolset_ids else (resolved_mcp_servers or None) listing: Final = await _get_tools_from_mcp_servers( user_api_key_auth=user_api_key_auth, mcp_auth_header=mcp_auth_header, mcp_servers=effective_server_filter, mcp_server_auth_headers=mcp_server_auth_headers, log_list_tools_to_spendlogs=True, list_tools_log_source="responses", litellm_trace_id=litellm_trace_id, request_tags=request_tags, ) tools: Final = listing.tools allowed_mcp_server_ids: Final = await global_mcp_server_manager.get_allowed_mcp_servers(user_api_key_auth) allowed_mcp_servers = global_mcp_server_manager.get_mcp_servers_from_ids(allowed_mcp_server_ids) allowed_mcp_servers = await _get_allowed_mcp_servers_from_mcp_server_names( mcp_servers=effective_server_filter, allowed_mcp_servers=allowed_mcp_servers, ) server_names: Final = [ server_name for server in allowed_mcp_servers if server is not None for server_name in ( getattr(server, "server_name", None) or getattr(server, "alias", None) or getattr(server, "name", None), ) if isinstance(server_name, str) ] return tools, server_names @staticmethod def _deduplicate_mcp_tools( mcp_tools: list[MCPTool], allowed_mcp_servers: list[str] ) -> tuple[list[MCPTool], dict[str, str]]: """ Deduplicate MCP tools by name, keeping the first occurrence of each tool. Args: mcp_tools: List of MCP tools that may contain duplicates Returns: List of deduplicated MCP tools The returned dictionary maps each tool_name to the server_name """ seen_names: Final[set[str]] = set() deduplicated_tools: Final[list[MCPTool]] = [] tool_server_map: Final[dict[str, str]] = {} for tool in mcp_tools: if isinstance(tool, dict): tool_name = tool.get("name") else: tool_name = getattr(tool, "name", None) if tool_name and tool_name not in seen_names: seen_names.add(tool_name) deduplicated_tools.append(tool) if len(allowed_mcp_servers) == 1: tool_server_map[tool_name] = allowed_mcp_servers[0] else: _, tool_server_map[tool_name] = split_server_prefix_from_name(tool_name) return deduplicated_tools, tool_server_map @staticmethod def _filter_mcp_tools_by_allowed_tools( mcp_tools: list[MCPTool], mcp_tools_with_litellm_proxy: Sequence[Mapping[str, object]] ) -> list[MCPTool]: """Filter MCP tools based on allowed_tools parameter from the original tool configs.""" # Collect all allowed tool names from all MCP tool configs allowed_tool_names: Final[set[str]] = set() for tool_config in mcp_tools_with_litellm_proxy: if isinstance(tool_config, dict) and "allowed_tools" in tool_config: allowed_tools = tool_config.get("allowed_tools", []) if isinstance(allowed_tools, list): allowed_tool_names.update(allowed_tools) # If no allowed_tools specified, return all tools if not allowed_tool_names: return mcp_tools # Filter tools based on allowed names filtered_tools: Final = [ mcp_tool for mcp_tool in mcp_tools for tool_name in (mcp_tool.get("name") if isinstance(mcp_tool, dict) else getattr(mcp_tool, "name", None),) if tool_name and (tool_name in allowed_tool_names or split_server_prefix_from_name(tool_name)[0] in allowed_tool_names) ] return filtered_tools @staticmethod async def _process_mcp_tools_to_openai_format( user_api_key_auth: "UserAPIKeyAuth | None", mcp_tools_with_litellm_proxy: Sequence[Mapping[str, object]], litellm_trace_id: str | None = None, request_tags: list[str] | None = None, ) -> tuple[list[Any], dict[str, str]]: """ Centralized method to process MCP tools through the complete pipeline. Args: user_api_key_auth: User authentication info for access control mcp_tools_with_litellm_proxy: ToolParam objects with server_url starting with "litellm_proxy" litellm_trace_id: Optional trace ID for linking list_mcp_tools spend logs to parent request Returns: List of tools in OpenAI format ready to be sent to the LLM The returned dictionary maps each tool_name to the server_name """ ( deduplicated_mcp_tools, tool_server_map, ) = await LiteLLM_Proxy_MCP_Handler._process_mcp_tools_without_openai_transform( user_api_key_auth, mcp_tools_with_litellm_proxy, litellm_trace_id=litellm_trace_id, request_tags=request_tags, ) openai_tools: Final = LiteLLM_Proxy_MCP_Handler._transform_mcp_tools_to_openai(deduplicated_mcp_tools) return openai_tools, tool_server_map @staticmethod async def _process_mcp_tools_without_openai_transform( user_api_key_auth: Any, mcp_tools_with_litellm_proxy: Sequence[Mapping[str, object]], litellm_trace_id: str | None = None, mcp_auth_header: str | None = None, mcp_server_auth_headers: dict[str, dict[str, str]] | None = None, request_tags: list[str] | None = None, ) -> tuple[list[MCPTool], dict[str, str]]: """ Process MCP tools through filtering and deduplication pipeline without OpenAI transformation. This is useful for cases where we need the original MCP tool objects (e.g., for events). Args: user_api_key_auth: User authentication info for access control mcp_tools_with_litellm_proxy: ToolParam objects with server_url starting with "litellm_proxy" mcp_auth_header: Optional deprecated auth header for MCP servers mcp_server_auth_headers: Optional server-specific auth headers (e.g. from x-mcp-{alias}-*) Returns: List of filtered and deduplicated MCP tools in their original format """ if not mcp_tools_with_litellm_proxy: return [], {} typed_user_api_key_auth: Final[UserAPIKeyAuth | None] = user_api_key_auth # Step 1: Fetch MCP tools from manager ( mcp_tools_fetched, allowed_mcp_servers, ) = await LiteLLM_Proxy_MCP_Handler._get_mcp_tools_from_manager( user_api_key_auth=typed_user_api_key_auth, mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy, litellm_trace_id=litellm_trace_id, mcp_auth_header=mcp_auth_header, mcp_server_auth_headers=mcp_server_auth_headers, request_tags=request_tags, ) # Step 2: Filter tools based on allowed_tools parameter filtered_mcp_tools: Final = LiteLLM_Proxy_MCP_Handler._filter_mcp_tools_by_allowed_tools( mcp_tools=mcp_tools_fetched, mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy, ) # Step 3: Deduplicate tools after filtering ( deduplicated_mcp_tools, tool_server_map, ) = LiteLLM_Proxy_MCP_Handler._deduplicate_mcp_tools(filtered_mcp_tools, allowed_mcp_servers) return deduplicated_mcp_tools, tool_server_map @overload @staticmethod def _transform_mcp_tools_to_openai( mcp_tools: Sequence[MCPTool], target_format: Literal["responses"] = ..., ) -> list[FunctionToolParam]: ... @overload @staticmethod def _transform_mcp_tools_to_openai( mcp_tools: Sequence[MCPTool], target_format: Literal["chat"], ) -> list[ChatCompletionToolParam]: ... @staticmethod def _transform_mcp_tools_to_openai( mcp_tools: Sequence[MCPTool], target_format: Literal["responses", "chat"] = "responses", ) -> Sequence[FunctionToolParam | ChatCompletionToolParam]: """Transform MCP tools to OpenAI-compatible format.""" from litellm.experimental_mcp_client.tools import ( transform_mcp_tool_to_openai_responses_api_tool, transform_mcp_tool_to_openai_tool, ) openai_tools: Final = [ transform_mcp_tool_to_openai_tool(mcp_tool) if target_format == "chat" else transform_mcp_tool_to_openai_responses_api_tool(mcp_tool) for mcp_tool in mcp_tools ] return openai_tools @staticmethod def _should_auto_execute_tools( mcp_tools_with_litellm_proxy: Sequence[Mapping[str, object]], ) -> bool: """Check if we should auto-execute tool calls. Auto-execution requires EVERY MCP reference to opt in with ``require_approval="never"``. A single reference that requires approval ("always", "manual", the object form, or an unset value) disables auto-execution for the whole request. This fails closed: when an approval-required reference shares a request with a "never" one, the model's tool calls are returned to the caller instead of being run, so an approval-gated tool can never be invoked without approval. Returns False for an empty list. """ references: Final = list(mcp_tools_with_litellm_proxy or []) if not references: return False for tool in references: approval = ( tool.get("require_approval") if isinstance(tool, dict) else getattr(tool, "require_approval", None) ) if approval != "never": return False return True @staticmethod def _extract_tool_calls_from_response(response: ResponsesAPIResponse) -> list[object]: """Extract tool calls from the response output.""" tool_calls: Final[list[object]] = [] for output_item in response.output: # Check if this is a function call output item if isinstance(output_item, dict) and output_item.get("type") == "function_call": tool_calls.append(output_item) elif hasattr(output_item, "type") and getattr(output_item, "type") == "function_call": # Handle pydantic model case tool_calls.append(output_item) return tool_calls @staticmethod def _extract_tool_calls_from_chat_response(response: ModelResponse) -> list[object]: """Extract tool calls from a chat completion response.""" tool_calls: Final[list[object]] = [] try: for choice in response.choices: message: Message | None = getattr(choice, "message", None) if message is None: continue tool_call_entries: ( Sequence[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None ) = getattr(message, "tool_calls", None) if tool_call_entries: for tool_call in tool_call_entries: if hasattr(tool_call, "model_dump"): tool_calls.append(tool_call.model_dump()) else: tool_calls.append(tool_call) except Exception: verbose_logger.exception("Failed to extract tool calls from chat completion response") return tool_calls @staticmethod def _extract_tool_call_details( tool_call: object, ) -> tuple[str | None, str | None, str | None]: """Extract tool name, arguments, and call_id from a tool call.""" if isinstance(tool_call, dict): tool_call_id = tool_call.get("call_id") or tool_call.get("id") # OpenAI chat completions wrap tool info under a `function` block function_block: Final = tool_call.get("function") if isinstance(function_block, dict): tool_name = function_block.get("name") tool_arguments = function_block.get("arguments") else: tool_name = tool_call.get("name") # Anthropic tool_use blocks carry the arguments under `input` tool_arguments = tool_call.get("arguments") if tool_arguments is None: tool_arguments = tool_call.get("input") else: tool_call_id = getattr(tool_call, "call_id", None) or getattr(tool_call, "id", None) function_obj: Final[object] = getattr(tool_call, "function", None) if function_obj is not None: tool_name = getattr(function_obj, "name", None) tool_arguments = getattr(function_obj, "arguments", None) else: tool_name = getattr(tool_call, "name", None) tool_arguments = getattr(tool_call, "arguments", None) if tool_arguments is None: tool_arguments = getattr(tool_call, "input", None) return tool_name, tool_arguments, tool_call_id @staticmethod def _parse_tool_arguments(tool_arguments: str | None) -> dict[str, object]: """Parse tool arguments, handling both string and dict formats.""" import json if isinstance(tool_arguments, str): try: return json.loads(tool_arguments) except json.JSONDecodeError: return {} else: return tool_arguments or {} @staticmethod def _parse_mcp_result(result: "CallToolResult") -> str: """Parse MCP tool call result and extract meaningful content.""" if not result or not hasattr(result, "content") or not result.content: return "Tool executed successfully" # Import MCP content types for isinstance checks try: from mcp.types import EmbeddedResource, ImageContent, TextContent except ImportError: # Fallback to generic handling if MCP types not available return "Tool executed successfully" text_parts: Final = [ str(content_item.text) for content_item in result.content if isinstance(content_item, TextContent) ] other_content_types: Final = [ "Image" if isinstance(content_item, ImageContent) else "EmbeddedResource" if isinstance(content_item, EmbeddedResource) else type(content_item).__name__ for content_item in result.content if not isinstance(content_item, TextContent) ] # Combine text parts if any result_text = " ".join(text_parts) if text_parts else "" # Add info about other content types if other_content_types: other_info: Final = f"[Generated {', '.join(other_content_types)}]" result_text = f"{result_text} {other_info}".strip() return result_text or "Tool executed successfully" @staticmethod async def _execute_tool_calls( tool_server_map: dict[str, str], tool_calls: Sequence[object], user_api_key_auth: Any, mcp_auth_header: str | None = None, mcp_server_auth_headers: dict[str, dict[str, str]] | None = None, oauth2_headers: dict[str, str] | None = None, raw_headers: dict[str, str] | None = None, litellm_call_id: str | None = None, litellm_trace_id: str | None = None, request_tags: list[str] | None = None, ) -> list[MCPToolResult]: """Execute tool calls and return results.""" from fastapi import HTTPException from litellm._uuid import uuid from litellm.exceptions import BlockedPiiEntityError, GuardrailRaisedException from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( global_mcp_server_manager, ) from litellm.proxy._experimental.mcp_server.server import ( _resolve_display_name_to_original, ) from litellm.proxy.proxy_server import proxy_logging_obj tool_results: Final[list[MCPToolResult]] = [] tool_call_id: str | None = None rules_obj: Final = Rules() logging_safe_headers: Final = logging_safe_mcp_headers(raw_headers) typed_user_api_key_auth: Final[UserAPIKeyAuth | None] = user_api_key_auth for tool_call in tool_calls: logging_request_data: dict[str, object] = {} tool_name: str | None = None try: ( tool_name, tool_arguments, tool_call_id, ) = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call) if not tool_name: verbose_logger.warning("Tool call missing name: %s", tool_call) continue parsed_arguments = LiteLLM_Proxy_MCP_Handler._parse_tool_arguments(tool_arguments) # Import here to avoid circular import from litellm.proxy.proxy_server import proxy_logging_obj server_name = tool_server_map[tool_name] mcp_server = global_mcp_server_manager.get_mcp_server_by_name( server_name ) or global_mcp_server_manager._get_mcp_server_from_tool_name(tool_name) resolved_tool_name = ( _resolve_display_name_to_original(tool_name, [mcp_server]) if mcp_server else tool_name ) sanitized_tool_name = strip_known_server_prefix(resolved_tool_name, mcp_server) start_time = datetime.now() logging_input: Sequence[Mapping[str, object]] = [ { "role": "tool", "content": { "tool_name": sanitized_tool_name, "arguments": parsed_arguments, }, } ] tool_logging_call_id = litellm_call_id or str(uuid.uuid4()) logging_metadata: dict[str, object] = { "tool_call_id": tool_call_id, "tool_name": sanitized_tool_name, "server_name": server_name, "headers": logging_safe_headers, } logging_request_data = { "model": f"MCP: {tool_name}", "metadata": logging_metadata, "input": logging_input, "call_type": CallTypes.call_mcp_tool.value, "litellm_call_id": tool_logging_call_id, # Add proxy_server_request with arguments for callback logging "proxy_server_request": { "url": "/mcp/tools/call", "method": "POST", "headers": logging_safe_headers, "body": { "name": sanitized_tool_name, "arguments": parsed_arguments, }, }, } if litellm_trace_id: logging_request_data["litellm_trace_id"] = litellm_trace_id if request_tags: logging_metadata["tags"] = request_tags if typed_user_api_key_auth is not None: from litellm.proxy.litellm_pre_call_utils import ( LiteLLMProxyRequestSetup, ) LiteLLMProxyRequestSetup.add_user_api_key_auth_to_request_metadata( data=logging_request_data, user_api_key_dict=typed_user_api_key_auth, _metadata_variable_name="metadata", ) user_identifier = getattr(typed_user_api_key_auth, "end_user_id", None) or getattr( typed_user_api_key_auth, "user_id", None ) if user_identifier: logging_request_data["user"] = user_identifier litellm_logging_obj: LiteLLMLoggingObj | None = None try: litellm_logging_obj, _ = function_setup( original_function="call_mcp_tool", rules_obj=rules_obj, start_time=start_time, **logging_request_data, ) except Exception as logging_error: verbose_logger.debug( "Failed to initialize logging for MCP tool call %s: %s", tool_name, logging_error, ) litellm_logging_obj = None logging_request_data["litellm_logging_obj"] = litellm_logging_obj logging_request_data["arguments"] = parsed_arguments if litellm_logging_obj: try: litellm_logging_obj.pre_call( input=logging_input, api_key="", ) except Exception: verbose_logger.exception("Failed to run pre_call for MCP tool logging") standard_logging_mcp_tool_call: StandardLoggingMCPToolCall = { "name": sanitized_tool_name, "arguments": parsed_arguments, "namespaced_tool_name": tool_name, } if mcp_server: mcp_info = mcp_server.mcp_info or {} standard_logging_mcp_tool_call["mcp_server_name"] = ( mcp_info.get("server_name") or getattr(mcp_server, "server_name", None) or server_name ) logo_url = mcp_info.get("logo_url") if logo_url: standard_logging_mcp_tool_call["mcp_server_logo_url"] = logo_url cost_info = mcp_info.get("mcp_server_cost_info") if cost_info: standard_logging_mcp_tool_call["mcp_server_cost_info"] = cost_info if litellm_logging_obj: litellm_logging_obj.model_call_details["mcp_tool_call_metadata"] = standard_logging_mcp_tool_call litellm_logging_obj.model = f"MCP: {tool_name}" litellm_logging_obj.call_type = CallTypes.call_mcp_tool.value result = await global_mcp_server_manager.call_tool( server_name=server_name, name=sanitized_tool_name, arguments=parsed_arguments, user_api_key_auth=typed_user_api_key_auth, mcp_auth_header=mcp_auth_header, mcp_server_auth_headers=mcp_server_auth_headers, oauth2_headers=oauth2_headers, raw_headers=raw_headers, proxy_logging_obj=proxy_logging_obj, litellm_logging_obj=litellm_logging_obj, ) if proxy_logging_obj: result = await proxy_logging_obj.post_mcp_call_hook( response=result, request_data=( litellm_logging_obj.model_call_details if litellm_logging_obj else {"mcp_tool_name": tool_name} ), user_api_key_dict=typed_user_api_key_auth, ) if litellm_logging_obj: try: litellm_logging_obj.post_call(original_response=result) await litellm_logging_obj.async_post_mcp_tool_call_hook( kwargs=litellm_logging_obj.model_call_details, response_obj=result, start_time=start_time, end_time=datetime.now(), ) except Exception: verbose_logger.exception("Failed to run post-call logging for MCP tool call %s", tool_name) try: await litellm_logging_obj.async_success_handler( result=result, start_time=start_time, end_time=datetime.now(), ) except Exception: verbose_logger.exception("Failed to log MCP tool call success for %s", tool_name) # Format result for inclusion in response result_text = LiteLLM_Proxy_MCP_Handler._parse_mcp_result(result) tool_results.append( { "tool_call_id": tool_call_id, "result": result_text, "name": tool_name, } ) except BlockedPiiEntityError as e: await LiteLLM_Proxy_MCP_Handler._log_mcp_tool_failure( proxy_logging_obj=proxy_logging_obj, user_api_key_auth=typed_user_api_key_auth, request_data=logging_request_data, error=e, ) verbose_logger.error("BlockedPiiEntityError in MCP tool call: %s", e) error_message = f"Tool call blocked: PII entity '{getattr(e, 'entity_type', 'unknown')}' detected by guardrail '{getattr(e, 'guardrail_name', 'unknown')}'. {e}" tool_results.append( { "tool_call_id": tool_call_id, "result": error_message, "name": tool_name, } ) except GuardrailRaisedException as e: await LiteLLM_Proxy_MCP_Handler._log_mcp_tool_failure( proxy_logging_obj=proxy_logging_obj, user_api_key_auth=typed_user_api_key_auth, request_data=logging_request_data, error=e, ) verbose_logger.error("GuardrailRaisedException in MCP tool call: %s", e) error_message = ( f"Tool call blocked: Guardrail '{getattr(e, 'guardrail_name', 'unknown')}' violation. {e}" ) tool_results.append( { "tool_call_id": tool_call_id, "result": error_message, "name": tool_name, } ) except HTTPException as e: await LiteLLM_Proxy_MCP_Handler._log_mcp_tool_failure( proxy_logging_obj=proxy_logging_obj, user_api_key_auth=typed_user_api_key_auth, request_data=logging_request_data, error=e, ) verbose_logger.error("HTTPException in MCP tool call: %s", e) error_message = f"Tool call failed: {str(e.detail) if hasattr(e, 'detail') else str(e)}" tool_results.append( { "tool_call_id": tool_call_id, "result": error_message, "name": tool_name, } ) except Exception as e: await LiteLLM_Proxy_MCP_Handler._log_mcp_tool_failure( proxy_logging_obj=proxy_logging_obj, user_api_key_auth=typed_user_api_key_auth, request_data=logging_request_data, error=e, ) verbose_logger.exception("Error executing MCP tool call: %s", e) tool_results.append( { "tool_call_id": tool_call_id, "result": f"Error executing tool: {e}", "name": tool_name, } ) return tool_results @staticmethod def _create_follow_up_messages_for_chat( original_messages: list[object], response: ModelResponse, tool_results: Sequence[Mapping[str, object]], ) -> Sequence[Mapping[str, object]]: """Create follow-up chat messages that include tool execution results.""" from copy import deepcopy from litellm.utils import convert_list_message_to_dict follow_up_messages: list[dict[str, object]] = convert_list_message_to_dict(deepcopy(original_messages)) if not follow_up_messages: follow_up_messages = [] message_to_append: dict | None = None try: first_choice: Final = response.choices[0] if isinstance(first_choice, Choices) and getattr(first_choice, "message", None): message_to_append = first_choice.message.model_dump(exclude_none=True) # Ensure tool_calls have arguments field (required by OpenAI API) if message_to_append.get("tool_calls"): for tool_call in message_to_append["tool_calls"]: if isinstance(tool_call, dict) and "function" in tool_call: if "arguments" not in tool_call["function"]: tool_call["function"]["arguments"] = "{}" except Exception: verbose_logger.exception("Failed to convert assistant message for MCP flow") if message_to_append: follow_up_messages.append(message_to_append) for tool_result in tool_results: follow_up_messages.append( { "role": "tool", "tool_call_id": tool_result.get("tool_call_id"), "name": tool_result.get("name"), "content": tool_result.get("result", ""), } ) return follow_up_messages @staticmethod def _create_follow_up_input( response: ResponsesAPIResponse, tool_results: Sequence[Mapping[str, object]], original_input: str | ResponseInputParam | None = None, ) -> list[object]: """Create follow-up input with tool results in proper format.""" follow_up_input: Final[list[object]] = [] # Add original user input if available to maintain conversation context if original_input: if isinstance(original_input, str): follow_up_input.append({"type": "message", "role": "user", "content": original_input}) elif isinstance(original_input, list): follow_up_input.extend(original_input) else: follow_up_input.append(original_input) # Add the assistant message with function calls assistant_message_content: Final[list[object]] = [] function_calls: Final[list[dict[str, object]]] = [] for output_item in response.output: if not isinstance(output_item, dict) and hasattr(output_item, "model_dump"): output_item = output_item.model_dump() if isinstance(output_item, dict): if output_item.get("type") == "function_call": call_id = output_item.get("call_id") or output_item.get("id") name = output_item.get("name") arguments = output_item.get("arguments") # Only add if we have required fields if call_id and name: function_calls.append( { "type": "function_call", "call_id": call_id, "name": name, "arguments": arguments, } ) elif output_item.get("type") == "message": # Extract content from message content = output_item.get("content", []) if isinstance(content, list): assistant_message_content.extend(content) else: assistant_message_content.append(content) # Add assistant message only if there's actual content (not empty) # For example, gemini requires that function call turns come immediately after user turns, # so we should not add empty assistant messages if assistant_message_content: follow_up_input.append( { "type": "message", "role": "assistant", "content": assistant_message_content, } ) # Add function calls (these can come directly after user message for LLM) for function_call in function_calls: follow_up_input.append(function_call) # Add tool results (function call outputs) for tool_result in tool_results: follow_up_input.append( { "type": "function_call_output", "call_id": tool_result["tool_call_id"], "output": tool_result["result"], } ) return follow_up_input @staticmethod async def _make_follow_up_call( follow_up_input: list[Any], model: str, all_tools: Sequence[ResponsesToolParam] | None, response_id: str, **call_params: Any, ) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator: """Make follow-up response API call with tool results.""" return await aresponses( input=follow_up_input, model=model, tools=all_tools, # Keep tools for potential future calls previous_response_id=response_id, # Link to previous response **call_params, ) @staticmethod async def _log_mcp_tool_failure( *, proxy_logging_obj: Optional["ProxyLogging"], user_api_key_auth: "UserAPIKeyAuth | None", request_data: dict[str, object], error: Exception, ) -> None: """Log MCP tool failures via proxy logging hooks.""" if proxy_logging_obj is None or user_api_key_auth is None: return try: traceback_str: Final = traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG) await proxy_logging_obj.post_call_failure_hook( request_data=request_data, original_exception=error, user_api_key_dict=user_api_key_auth, route="/responses/mcp/call_tool", traceback_str=traceback_str, ) except Exception: verbose_logger.exception("Failed to log MCP tool call failure") @staticmethod def _create_mcp_streaming_response( input: str | ResponseInputParam, model: str, all_tools: Sequence[object] | None, mcp_tools_with_litellm_proxy: list[Mapping[str, object]], mcp_discovery_events: list[ResponsesAPIStreamingResponse], call_params: Mapping[str, object], previous_response_id: str | None, tool_server_map: dict[str, str], **kwargs, ) -> Any: """ Create MCP enhanced streaming response that handles the full MCP workflow. This creates a streaming iterator that: 1. Immediately emits MCP discovery events 2. Makes the LLM call and streams the response 3. Handles tool execution and follow-up calls """ from litellm.responses.mcp.mcp_streaming_iterator import ( MCPEnhancedStreamingIterator, ) # Build the complete request parameters by merging all sources request_params: Final = LiteLLM_Proxy_MCP_Handler._build_request_params( input=input, model=model, all_tools=all_tools, call_params=call_params, previous_response_id=previous_response_id, **kwargs, ) # Create the enhanced streaming iterator that will handle everything return MCPEnhancedStreamingIterator( base_iterator=None, # Will be created internally mcp_events=mcp_discovery_events, # Pre-generated MCP discovery events tool_server_map=tool_server_map, mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy, user_api_key_auth=kwargs.get("user_api_key_auth") or kwargs.get("litellm_metadata", {}).get("user_api_key_auth"), original_request_params=request_params, ) @staticmethod def _build_request_params( input: str | ResponseInputParam, model: str, all_tools: Sequence[object] | None, call_params: Mapping[str, object], previous_response_id: str | None, **kwargs: object, ) -> dict[str, object]: """ Build a clean request parameters dictionary for MCP streaming. Combines input, model, tools with call_params and additional kwargs in a clean, maintainable way. """ # Start with the core required parameters request_params: Final[dict[str, object]] = { "input": input, "model": model, "tools": all_tools, } # Add previous_response_id if provided if previous_response_id is not None: request_params["previous_response_id"] = previous_response_id # Merge in all call_params (which contains most of the API parameters) request_params.update(call_params) # Merge in any additional kwargs request_params.update(kwargs) return request_params @staticmethod def _create_tool_execution_events( tool_calls: Sequence[object], tool_results: Sequence[MCPToolResult] ) -> list[ResponsesAPIStreamingResponse]: """ Create MCP tool execution events for streaming. Args: tool_calls: List of tool calls from the LLM response tool_results: List of tool execution results Returns: List of MCP tool execution events for streaming """ from litellm._uuid import uuid from litellm.responses.mcp.mcp_streaming_iterator import create_mcp_call_events tool_execution_events: Final[list[ResponsesAPIStreamingResponse]] = [] # Create events for each tool execution for tool_result in tool_results: tool_call_id = tool_result.get("tool_call_id", "unknown") result_text = tool_result.get("result", "") # Extract tool name and arguments from tool calls tool_name = "unknown" tool_arguments = "{}" for tool_call in tool_calls: ( name, args, call_id, ) = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call) if call_id == tool_call_id: tool_name = name or "unknown" tool_arguments = args or "{}" break execution_events = create_mcp_call_events( tool_name=tool_name, tool_call_id=tool_call_id, arguments=tool_arguments, # Use actual arguments result=result_text, base_item_id=f"mcp_{uuid.uuid4().hex[:8]}", # Unique ID for each tool call sequence_start=len(tool_execution_events) + 1, ) tool_execution_events.extend(execution_events) return tool_execution_events @staticmethod def _prepare_initial_call_params(call_params: dict[str, Any], should_auto_execute: bool) -> dict[str, Any]: """ Prepare call parameters for the initial LLM call. For auto-execute scenarios, we need to disable streaming for the initial call so we can process the tool calls before streaming the final response. """ initial_params: Final = call_params.copy() if should_auto_execute: # Disable streaming for initial call when auto-executing tools initial_params["stream"] = False return initial_params @staticmethod def _prepare_follow_up_call_params(call_params: dict[str, Any], original_stream_setting: bool) -> dict[str, Any]: """ Prepare call parameters for the follow-up LLM call after tool execution. Restores the original streaming setting and removes tool_choice since we're now providing tool results, not requesting tool calls. """ follow_up_params: Final = call_params.copy() # Restore original streaming setting for follow-up call follow_up_params["stream"] = original_stream_setting # Remove tool_choice since we're providing results, not requesting tool calls follow_up_params.pop("tool_choice", None) return follow_up_params @staticmethod def _add_mcp_output_elements_to_response( response: ResponsesAPIResponse, mcp_tools_fetched: Sequence[object], tool_results: Sequence[Mapping[str, object]], ) -> ResponsesAPIResponse: """Add custom output elements to the final response for MCP tool execution.""" # Import the required classes for creating output items import json from litellm._uuid import uuid from litellm.types.responses.main import GenericResponseOutputItem, OutputText # Create output element for initial MCP tools mcp_tools_output: Final = GenericResponseOutputItem( type="mcp_tools_fetched", id=f"mcp_tools_{uuid.uuid4().hex[:8]}", status="completed", role="system", content=[ OutputText( type="output_text", text=json.dumps(mcp_tools_fetched, indent=2, default=str), annotations=[], ) ], ) # Create output element for tool execution results tool_results_output: Final = GenericResponseOutputItem( type="tool_execution_results", id=f"tool_results_{uuid.uuid4().hex[:8]}", status="completed", role="system", content=[ OutputText( type="output_text", text=json.dumps(tool_results, indent=2, default=str), annotations=[], ) ], ) # Add the new output elements to the response response.output.append(mcp_tools_output.model_dump()) response.output.append(tool_results_output.model_dump()) return response