mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
Merge pull request #32579 from thibault-linktree/litellm_fix_mcp_gateway_failure_handling
fix(responses): fail loudly on MCP gateway failures (initial call, mid-stream, zero resolved tools)
This commit is contained in:
commit
9e7ba65759
6 changed files with 542 additions and 11 deletions
|
|
@ -316,6 +316,12 @@ disable_add_transform_inline_image_block: bool = False
|
|||
disable_add_user_agent_to_request_tags: bool = False
|
||||
disable_anthropic_gemini_context_caching_transform: bool = False
|
||||
disable_vertex_batch_output_transformation: bool = False
|
||||
# Raise a 400 when a Responses API request asks for MCP gateway tools
|
||||
# (server_url litellm_proxy/...) but zero tools resolve (key/team lacks server
|
||||
# access, unknown server name, or allowed_tools matches nothing) and the
|
||||
# request carries no other tools. Without this the model is silently called
|
||||
# with no tools and hallucinates. Set to False to restore the old behaviour.
|
||||
reject_empty_mcp_resolved_tools: bool = True
|
||||
extra_spend_tag_headers: Optional[List[str]] = None
|
||||
in_memory_llm_clients_cache: "LLMClientCache"
|
||||
safe_memory_mode: bool = False
|
||||
|
|
|
|||
|
|
@ -216,6 +216,35 @@ async def aresponses_api_with_mcp(
|
|||
)
|
||||
openai_tools = LiteLLM_Proxy_MCP_Handler._transform_mcp_tools_to_openai(original_mcp_tools)
|
||||
|
||||
if (
|
||||
litellm.reject_empty_mcp_resolved_tools
|
||||
and mcp_tools_with_litellm_proxy
|
||||
and not original_mcp_tools
|
||||
and not other_tools
|
||||
):
|
||||
# The request explicitly asked for MCP tools but none resolved, and
|
||||
# there are no other tools to fall back on. This is almost always a
|
||||
# misconfiguration: the API key/team has no access to the MCP server
|
||||
# (allow_all_keys=false and no object-permission grant), the server
|
||||
# name does not exist, or allowed_tools matches no tool on the server.
|
||||
# Silently calling the model with no tools makes it hallucinate, and
|
||||
# the only trace is a list_mcp_tools spend log with an empty response —
|
||||
# so fail loudly instead.
|
||||
requested_mcp_urls = [tool.get("server_url") for tool in mcp_tools_with_litellm_proxy if isinstance(tool, dict)]
|
||||
raise litellm.BadRequestError(
|
||||
message=(
|
||||
"MCP gateway resolved 0 tools for the requested MCP tool(s) "
|
||||
f"(server_url(s): {requested_mcp_urls}). Likely causes: the API "
|
||||
"key/team does not have access to the MCP server (server has "
|
||||
"allow_all_keys=false and no key/team object-permission grant), "
|
||||
"the server name does not exist, or allowed_tools matches no "
|
||||
"tool on the server. Set litellm.reject_empty_mcp_resolved_tools "
|
||||
"= False to restore the previous silent behaviour."
|
||||
),
|
||||
model=model,
|
||||
llm_provider=custom_llm_provider or "openai",
|
||||
)
|
||||
|
||||
# Combine with other tools
|
||||
all_tools = openai_tools + other_tools if (openai_tools or other_tools) else None
|
||||
|
||||
|
|
@ -261,7 +290,7 @@ async def aresponses_api_with_mcp(
|
|||
pre_processed_mcp_tools=original_mcp_tools,
|
||||
)
|
||||
|
||||
return LiteLLM_Proxy_MCP_Handler._create_mcp_streaming_response(
|
||||
mcp_streaming_response = LiteLLM_Proxy_MCP_Handler._create_mcp_streaming_response(
|
||||
input=input,
|
||||
model=model,
|
||||
all_tools=all_tools,
|
||||
|
|
@ -272,6 +301,16 @@ async def aresponses_api_with_mcp(
|
|||
tool_server_map=tool_server_map,
|
||||
**kwargs,
|
||||
)
|
||||
# Make the initial LLM call eagerly, before any SSE bytes are written,
|
||||
# so a pre-stream failure (e.g. an invalid previous_response_id ->
|
||||
# provider 400 "No tool output found for function call ...") surfaces
|
||||
# as a normal HTTP error instead of an HTTP 200 whose stream emits
|
||||
# mcp_list_tools events with no response.created (which crashes SDK
|
||||
# stream accumulators).
|
||||
await mcp_streaming_response._create_initial_response_iterator()
|
||||
if mcp_streaming_response._initial_creation_error is not None:
|
||||
raise mcp_streaming_response._initial_creation_error
|
||||
return mcp_streaming_response
|
||||
|
||||
# Determine if we should auto-execute tools
|
||||
should_auto_execute = bool(mcp_tools_with_litellm_proxy) and LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools(
|
||||
|
|
|
|||
|
|
@ -5,6 +5,8 @@ from litellm._uuid import uuid
|
|||
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
|
||||
from litellm.types.llms.openai import (
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
ErrorEvent,
|
||||
ErrorEventError,
|
||||
MCPCallArgumentsDeltaEvent,
|
||||
MCPCallArgumentsDoneEvent,
|
||||
MCPCallCompletedEvent,
|
||||
|
|
@ -304,6 +306,18 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
# Cache the response ID to ensure consistency across all events
|
||||
self._cached_response_id: Optional[str] = None
|
||||
|
||||
# Internal failures (initial LLM call, tool execution, follow-up call)
|
||||
# are stashed here so they can be surfaced to the client as an `error`
|
||||
# stream event, or re-raised before any SSE bytes are written (eager
|
||||
# path in aresponses_api_with_mcp for the initial call).
|
||||
self._initial_creation_error: Optional[Exception] = None
|
||||
self._stream_error: Optional[Exception] = None
|
||||
self._error_event_emitted = False
|
||||
# Highest sequence_number emitted so far; the terminal `error` event
|
||||
# must be numbered after it to keep the stream monotonic for strict
|
||||
# clients.
|
||||
self._last_sequence_number = 0
|
||||
|
||||
def _extract_mcp_headers_from_params(self) -> None:
|
||||
"""Extract MCP headers from original request params to pass to tool calls"""
|
||||
from typing import Dict, Optional
|
||||
|
|
@ -368,10 +382,34 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
|
||||
return LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools(self.mcp_tools_with_litellm_proxy)
|
||||
|
||||
def _make_stream_error_event(self) -> ResponsesAPIStreamingResponse:
|
||||
"""Build an OpenAI-style `error` stream event from the stashed internal
|
||||
failure, so clients receive a real terminal error instead of a stream
|
||||
that silently ends mid-flow."""
|
||||
err = self._stream_error
|
||||
status_code = getattr(err, "status_code", None)
|
||||
return ErrorEvent(
|
||||
type=ResponsesAPIStreamEvents.ERROR,
|
||||
sequence_number=self._last_sequence_number + 1,
|
||||
error=ErrorEventError(
|
||||
type="mcp_gateway_error",
|
||||
code=str(status_code) if status_code is not None else "internal_error",
|
||||
message=str(err) if err is not None else "MCP gateway stream failed",
|
||||
param=None,
|
||||
),
|
||||
)
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self) -> ResponsesAPIStreamingResponse:
|
||||
chunk = await self._anext_impl()
|
||||
sequence_number = getattr(chunk, "sequence_number", None)
|
||||
if isinstance(sequence_number, int) and sequence_number > self._last_sequence_number:
|
||||
self._last_sequence_number = sequence_number
|
||||
return chunk
|
||||
|
||||
async def _anext_impl(self) -> ResponsesAPIStreamingResponse:
|
||||
"""
|
||||
Phase-based streaming:
|
||||
1. initial_response - Stream the first LLM response (includes response.created, response.in_progress, response.output_item.added)
|
||||
|
|
@ -429,6 +467,12 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
raise
|
||||
else:
|
||||
self.phase = "finished"
|
||||
# Tool execution or the follow-up call failed: emit a terminal
|
||||
# `error` event so the client can distinguish a failed stream
|
||||
# from a completed one.
|
||||
if self._stream_error is not None and not self._error_event_emitted:
|
||||
self._error_event_emitted = True
|
||||
return self._make_stream_error_event()
|
||||
raise StopAsyncIteration
|
||||
|
||||
# Phase 6: Finished
|
||||
|
|
@ -451,13 +495,17 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
await self._create_initial_response_iterator()
|
||||
|
||||
if self.base_iterator is None:
|
||||
# LLM call failed — still emit MCP discovery events before finishing
|
||||
if self.mcp_discovery_events:
|
||||
self.phase = "mcp_discovery"
|
||||
else:
|
||||
self.phase = "finished"
|
||||
raise StopAsyncIteration
|
||||
return None
|
||||
# The initial LLM call failed. Do NOT emit MCP discovery events: a
|
||||
# stream that starts with mcp_list_tools events and no
|
||||
# response.created violates the Responses API streaming contract
|
||||
# and crashes SDK stream accumulators (openai-node: "expected
|
||||
# 'response.created' event, got response.mcp_list_tools.in_progress").
|
||||
# Surface the failure as an `error` event instead.
|
||||
self.phase = "finished"
|
||||
if self._stream_error is not None:
|
||||
self._error_event_emitted = True
|
||||
return self._make_stream_error_event()
|
||||
raise StopAsyncIteration
|
||||
|
||||
if self.base_iterator:
|
||||
if hasattr(self.base_iterator, "__anext__"):
|
||||
|
|
@ -580,8 +628,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
|
||||
traceback.print_exc()
|
||||
self.base_iterator = None
|
||||
# Don't set phase to "finished" here — let __anext__ emit any
|
||||
# pre-generated MCP discovery events before ending the iteration.
|
||||
# Stash the failure so aresponses_api_with_mcp can re-raise it
|
||||
# before any SSE bytes are written (eager creation), or so
|
||||
# __anext__ can emit an `error` event instead of ending silently.
|
||||
self._initial_creation_error = e
|
||||
self._stream_error = e
|
||||
|
||||
async def _generate_tool_execution_events(self) -> None:
|
||||
"""Generate tool execution events and execute tools"""
|
||||
|
|
@ -693,11 +744,26 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
|
||||
traceback.print_exc()
|
||||
self.tool_results = []
|
||||
# Drop the queued per-tool events: emitting mcp_call.in_progress
|
||||
# items that never receive a completed/failed terminal event is a
|
||||
# protocol deviation. The terminal `error` event carries the
|
||||
# failure instead.
|
||||
self.tool_execution_events = []
|
||||
# Remember the failure. Without this, the follow-up call is made
|
||||
# with function_call items but no function_call_output items and
|
||||
# the provider rejects it with "No tool output found for function
|
||||
# call ...".
|
||||
self._stream_error = e
|
||||
|
||||
async def _create_follow_up_iterator(self) -> None:
|
||||
"""Create the follow-up response iterator with tool results"""
|
||||
if not self.collected_response or not hasattr(self, "tool_results"):
|
||||
return
|
||||
# Tool execution already failed; skip the doomed follow-up call (it
|
||||
# would be rejected with "No tool output found for function call ...")
|
||||
# and let __anext__ emit the terminal error event.
|
||||
if self._stream_error is not None:
|
||||
return
|
||||
|
||||
from litellm.responses.main import aresponses
|
||||
from litellm.responses.mcp.litellm_proxy_mcp_handler import (
|
||||
|
|
@ -738,6 +804,9 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
|
||||
traceback.print_exc()
|
||||
self.follow_up_iterator = None
|
||||
# Surface via a terminal `error` event in __anext__ instead of
|
||||
# silently ending the stream with no terminal event.
|
||||
self._stream_error = e
|
||||
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
|
|
|||
|
|
@ -278,7 +278,12 @@ async def test_aresponses_api_with_mcp_passes_mcp_server_auth_headers_to_process
|
|||
|
||||
async def mock_process(**kwargs):
|
||||
captured_process_kwargs.update(kwargs)
|
||||
return ([], {})
|
||||
from mcp.types import Tool as MCPTool
|
||||
|
||||
dummy_tool = MCPTool(
|
||||
name="dummy_tool", description="dummy", inputSchema={"type": "object"}
|
||||
)
|
||||
return ([dummy_tool], {"dummy_tool": "dummy_server"})
|
||||
|
||||
mock_response = ResponsesAPIResponse(
|
||||
**{
|
||||
|
|
|
|||
|
|
@ -0,0 +1,145 @@
|
|||
"""
|
||||
Guard tests: a request that explicitly asks for MCP tools via the litellm_proxy
|
||||
gateway but resolves zero of them (and has no other tools to fall back on) must
|
||||
fail loudly with a 400 instead of silently calling the model with no tools —
|
||||
which makes it hallucinate, with the only trace being a "success"
|
||||
list_mcp_tools spend log with an empty response.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.responses.mcp.litellm_proxy_mcp_handler import LiteLLM_Proxy_MCP_Handler
|
||||
from litellm.types.llms.openai import ResponsesAPIResponse
|
||||
|
||||
# See test_mcp_streaming_iterator.py: look the real submodule up in sys.modules
|
||||
# to sidestep litellm.responses being shadowed by the re-exported function.
|
||||
responses_main_module = sys.modules["litellm.responses.main"]
|
||||
|
||||
MCP_TOOL = {
|
||||
"type": "mcp",
|
||||
"server_url": "litellm_proxy/mcp/nonexistent_server",
|
||||
"require_approval": "never",
|
||||
"allowed_tools": ["get_links"],
|
||||
}
|
||||
|
||||
|
||||
def _patch_resolved_tools(monkeypatch: pytest.MonkeyPatch, resolved_tools: list) -> None:
|
||||
monkeypatch.setattr(
|
||||
LiteLLM_Proxy_MCP_Handler,
|
||||
"_process_mcp_tools_without_openai_transform",
|
||||
AsyncMock(return_value=(resolved_tools, {})),
|
||||
)
|
||||
|
||||
|
||||
def _model_response() -> ResponsesAPIResponse:
|
||||
return ResponsesAPIResponse(id="resp-1", created_at=0, output=[])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zero_resolved_mcp_tools_raises_before_model_call(monkeypatch):
|
||||
# The guard runs before the stream/non-stream branch in
|
||||
# aresponses_api_with_mcp, so one case covers both.
|
||||
_patch_resolved_tools(monkeypatch, [])
|
||||
aresponses_mock = AsyncMock()
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
with pytest.raises(litellm.BadRequestError) as excinfo:
|
||||
await responses_main_module.aresponses_api_with_mcp(
|
||||
input="how many links do i have?",
|
||||
model="gpt-4",
|
||||
stream=False,
|
||||
tools=[MCP_TOOL],
|
||||
)
|
||||
|
||||
message = str(excinfo.value)
|
||||
assert "resolved 0 tools" in message
|
||||
assert "litellm_proxy/mcp/nonexistent_server" in message
|
||||
assert "allow_all_keys" in message
|
||||
# The model was never called without its tools.
|
||||
aresponses_mock.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zero_resolved_mcp_tools_with_function_tools_falls_back(monkeypatch):
|
||||
"""Mixed requests keep working: with other (function) tools present, the
|
||||
request proceeds using those tools instead of hard-failing."""
|
||||
_patch_resolved_tools(monkeypatch, [])
|
||||
response = _model_response()
|
||||
aresponses_mock = AsyncMock(return_value=response)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
function_tool = {"type": "function", "name": "my_fn", "parameters": {}}
|
||||
result = await responses_main_module.aresponses_api_with_mcp(
|
||||
input="hello",
|
||||
model="gpt-4",
|
||||
stream=False,
|
||||
tools=[MCP_TOOL, function_tool],
|
||||
)
|
||||
|
||||
assert result is response
|
||||
aresponses_mock.assert_called_once()
|
||||
assert aresponses_mock.call_args.kwargs["tools"] == [function_tool]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zero_resolved_mcp_tools_flag_off_restores_old_behaviour(monkeypatch):
|
||||
_patch_resolved_tools(monkeypatch, [])
|
||||
response = _model_response()
|
||||
aresponses_mock = AsyncMock(return_value=response)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
monkeypatch.setattr(litellm, "reject_empty_mcp_resolved_tools", False)
|
||||
|
||||
result = await responses_main_module.aresponses_api_with_mcp(
|
||||
input="how many links do i have?",
|
||||
model="gpt-4",
|
||||
stream=False,
|
||||
tools=[MCP_TOOL],
|
||||
)
|
||||
|
||||
assert result is response
|
||||
aresponses_mock.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolved_mcp_tools_proceed_to_model_call(monkeypatch):
|
||||
from mcp.types import Tool as MCPTool
|
||||
|
||||
resolved = [MCPTool(name="get_links", description="List links", inputSchema={"type": "object"})]
|
||||
_patch_resolved_tools(monkeypatch, resolved)
|
||||
|
||||
response = _model_response()
|
||||
aresponses_mock = AsyncMock(return_value=response)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
result = await responses_main_module.aresponses_api_with_mcp(
|
||||
input="how many links do i have?",
|
||||
model="gpt-4",
|
||||
stream=False,
|
||||
tools=[MCP_TOOL],
|
||||
)
|
||||
|
||||
assert result is response
|
||||
aresponses_mock.assert_called_once()
|
||||
assert aresponses_mock.call_args.kwargs["tools"], "model call must carry the resolved tools"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_without_mcp_tools_is_unaffected(monkeypatch):
|
||||
"""Plain function-tool requests never hit the guard."""
|
||||
response = _model_response()
|
||||
aresponses_mock = AsyncMock(return_value=response)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
result = await responses_main_module.aresponses_api_with_mcp(
|
||||
input="hello",
|
||||
model="gpt-4",
|
||||
stream=False,
|
||||
tools=[{"type": "function", "name": "my_fn", "parameters": {}}],
|
||||
)
|
||||
|
||||
assert result is response
|
||||
aresponses_mock.assert_called_once()
|
||||
267
tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py
Normal file
267
tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py
Normal file
|
|
@ -0,0 +1,267 @@
|
|||
import sys
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.responses.mcp.mcp_streaming_iterator import MCPEnhancedStreamingIterator
|
||||
from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStreamEvents
|
||||
|
||||
# `litellm.__init__` re-exports a function named `responses`, which shadows the
|
||||
# `litellm.responses` subpackage as an attribute — `import litellm.responses.main`
|
||||
# can resolve to the unrelated third-party `responses` package instead. Look the
|
||||
# real submodule up in sys.modules directly to sidestep the shadowing.
|
||||
responses_main_module = sys.modules["litellm.responses.main"]
|
||||
|
||||
|
||||
class _FakeAsyncStream:
|
||||
"""Minimal async iterator yielding pre-built chunks, one per __anext__ call."""
|
||||
|
||||
def __init__(self, chunks):
|
||||
self._chunks = list(chunks)
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
if not self._chunks:
|
||||
raise StopAsyncIteration
|
||||
return self._chunks.pop(0)
|
||||
|
||||
|
||||
def _completed_chunk(output):
|
||||
response = ResponsesAPIResponse(id="resp-1", created_at=0, output=output)
|
||||
return SimpleNamespace(type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=response)
|
||||
|
||||
|
||||
def _text_message(text: str):
|
||||
return {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]}
|
||||
|
||||
|
||||
def _text_only_stream(text: str) -> _FakeAsyncStream:
|
||||
return _FakeAsyncStream([_completed_chunk([_text_message(text)])])
|
||||
|
||||
|
||||
def _make_lazy_iterator(mcp_events=None) -> MCPEnhancedStreamingIterator:
|
||||
"""Iterator with no base_iterator: the initial LLM call happens lazily on iteration."""
|
||||
return MCPEnhancedStreamingIterator(
|
||||
base_iterator=None,
|
||||
mcp_events=list(mcp_events or []),
|
||||
tool_server_map={"read_wiki_contents": "deepwiki"},
|
||||
mcp_tools_with_litellm_proxy=[{"require_approval": "never"}],
|
||||
user_api_key_auth=None,
|
||||
original_request_params={
|
||||
"model": "gpt-4",
|
||||
"input": "what is berriai/litellm?",
|
||||
"tools": [{"type": "mcp"}],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initial_call_failure_emits_error_event_not_discovery_events(monkeypatch):
|
||||
"""
|
||||
Regression test: when the initial LLM call fails (e.g. an invalid
|
||||
previous_response_id -> provider 400 "No tool output found for function
|
||||
call ..."), the stream used to emit the pre-generated mcp_list_tools
|
||||
discovery events with no response.created before them — which violates
|
||||
the Responses API streaming contract and crashes SDK stream accumulators
|
||||
(openai-node: "expected 'response.created' event, got
|
||||
response.mcp_list_tools.in_progress"). The stream must instead surface a
|
||||
single terminal `error` event and end.
|
||||
"""
|
||||
aresponses_mock = AsyncMock(
|
||||
side_effect=litellm.BadRequestError(
|
||||
message="No tool output found for function call call_x.",
|
||||
model="gpt-4",
|
||||
llm_provider="openai",
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
discovery_event = SimpleNamespace(type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS)
|
||||
iterator = _make_lazy_iterator(mcp_events=[discovery_event])
|
||||
|
||||
chunks = [chunk async for chunk in iterator]
|
||||
|
||||
assert len(chunks) == 1
|
||||
error_event = chunks[0]
|
||||
assert error_event.type == ResponsesAPIStreamEvents.ERROR
|
||||
assert error_event.error.code == "400"
|
||||
assert "No tool output found" in error_event.error.message
|
||||
# No discovery events leaked before/after the error.
|
||||
assert discovery_event not in chunks
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_eager_creation_reraises_pre_stream_failure_as_http_error(monkeypatch):
|
||||
"""
|
||||
aresponses_api_with_mcp creates the initial response eagerly and re-raises
|
||||
the stashed creation failure, so the proxy returns a real 4xx/5xx before
|
||||
any SSE bytes are written instead of an HTTP 200 with a broken stream.
|
||||
"""
|
||||
from mcp.types import Tool as MCPTool
|
||||
|
||||
from litellm.responses.mcp.litellm_proxy_mcp_handler import LiteLLM_Proxy_MCP_Handler
|
||||
|
||||
resolved_tool = MCPTool(name="read_wiki_contents", description="read", inputSchema={"type": "object"})
|
||||
monkeypatch.setattr(
|
||||
LiteLLM_Proxy_MCP_Handler,
|
||||
"_process_mcp_tools_without_openai_transform",
|
||||
AsyncMock(return_value=([resolved_tool], {"read_wiki_contents": "deepwiki"})),
|
||||
)
|
||||
boom = litellm.BadRequestError(
|
||||
message="Previous response with id 'resp_bogus' not found.",
|
||||
model="gpt-4",
|
||||
llm_provider="openai",
|
||||
)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", AsyncMock(side_effect=boom))
|
||||
|
||||
with pytest.raises(litellm.BadRequestError) as excinfo:
|
||||
await responses_main_module.aresponses_api_with_mcp(
|
||||
input="hi",
|
||||
model="gpt-4",
|
||||
stream=True,
|
||||
previous_response_id="resp_bogus",
|
||||
tools=[{"type": "mcp", "server_url": "litellm_proxy/mcp/deepwiki", "require_approval": "never"}],
|
||||
)
|
||||
|
||||
assert "resp_bogus" in str(excinfo.value)
|
||||
|
||||
|
||||
import types
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from mcp.types import CallToolResult, TextContent
|
||||
|
||||
|
||||
def _output_item_added_chunk():
|
||||
return SimpleNamespace(type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED)
|
||||
|
||||
|
||||
def _function_call(call_id: str, name: str, arguments: str = "{}"):
|
||||
return {"type": "function_call", "call_id": call_id, "name": name, "arguments": arguments}
|
||||
|
||||
|
||||
def _mock_mcp_environment(monkeypatch: pytest.MonkeyPatch) -> AsyncMock:
|
||||
"""Patch the MCP tool-call plumbing so _execute_tool_calls can run in tests."""
|
||||
call_tool = AsyncMock(return_value=CallToolResult(content=[TextContent(type="text", text="ok")], isError=False))
|
||||
fake_manager = types.SimpleNamespace(
|
||||
call_tool=call_tool,
|
||||
_get_mcp_server_from_tool_name=MagicMock(return_value=None),
|
||||
get_mcp_server_by_name=MagicMock(return_value=None),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager",
|
||||
fake_manager,
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"litellm.proxy.proxy_server",
|
||||
types.SimpleNamespace(proxy_logging_obj=MagicMock()),
|
||||
)
|
||||
return call_tool
|
||||
|
||||
|
||||
def _make_tool_call_iterator() -> MCPEnhancedStreamingIterator:
|
||||
return MCPEnhancedStreamingIterator(
|
||||
base_iterator=_FakeAsyncStream(
|
||||
[
|
||||
_output_item_added_chunk(),
|
||||
_completed_chunk([_function_call("call_1", "read_wiki_contents")]),
|
||||
]
|
||||
),
|
||||
mcp_events=[],
|
||||
tool_server_map={"read_wiki_contents": "deepwiki"},
|
||||
mcp_tools_with_litellm_proxy=[{"require_approval": "never"}],
|
||||
user_api_key_auth=None,
|
||||
original_request_params={
|
||||
"model": "gpt-4",
|
||||
"input": "what is berriai/litellm?",
|
||||
"tools": [{"type": "mcp"}],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_execution_failure_emits_error_event_and_skips_follow_up(monkeypatch):
|
||||
"""
|
||||
When tool execution blows up as a batch (not a per-tool error string),
|
||||
the stream used to proceed to a follow-up call carrying function_call
|
||||
items with no outputs — rejected by the provider with "No tool output
|
||||
found for function call ..." — and then end silently. It must instead
|
||||
skip the doomed follow-up and emit a terminal `error` event.
|
||||
"""
|
||||
from litellm.responses.mcp.litellm_proxy_mcp_handler import LiteLLM_Proxy_MCP_Handler
|
||||
|
||||
monkeypatch.setattr(
|
||||
LiteLLM_Proxy_MCP_Handler,
|
||||
"_execute_tool_calls",
|
||||
AsyncMock(side_effect=RuntimeError("mcp server exploded")),
|
||||
)
|
||||
aresponses_mock = AsyncMock()
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
iterator = _make_tool_call_iterator()
|
||||
chunks = [chunk async for chunk in iterator]
|
||||
|
||||
error_events = [c for c in chunks if getattr(c, "type", None) == ResponsesAPIStreamEvents.ERROR]
|
||||
assert len(error_events) == 1
|
||||
assert "mcp server exploded" in error_events[0].error.message
|
||||
# The doomed follow-up call was never made.
|
||||
aresponses_mock.assert_not_called()
|
||||
# No orphaned per-tool events: a batch failure must not emit
|
||||
# mcp_call.in_progress items that never receive a terminal event.
|
||||
assert all(getattr(c, "type", None) != ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS for c in chunks)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_follow_up_failure_emits_error_event(monkeypatch):
|
||||
"""
|
||||
When the follow-up LLM call after successful tool execution fails, the
|
||||
stream used to end with no terminal event (the client saw tool events
|
||||
and then... nothing). It must emit a terminal `error` event carrying the
|
||||
mapped provider failure.
|
||||
"""
|
||||
_mock_mcp_environment(monkeypatch)
|
||||
|
||||
boom = litellm.BadRequestError(
|
||||
message="No tool output found for function call call_1.",
|
||||
model="gpt-4",
|
||||
llm_provider="openai",
|
||||
)
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", AsyncMock(side_effect=boom))
|
||||
|
||||
iterator = _make_tool_call_iterator()
|
||||
chunks = [chunk async for chunk in iterator]
|
||||
|
||||
error_events = [c for c in chunks if getattr(c, "type", None) == ResponsesAPIStreamEvents.ERROR]
|
||||
assert len(error_events) == 1
|
||||
assert error_events[0].error.code == "400"
|
||||
assert "No tool output found" in error_events[0].error.message
|
||||
# Tool-execution events were still streamed before the error surfaced.
|
||||
assert any(getattr(c, "type", None) == ResponsesAPIStreamEvents.MCP_CALL_COMPLETED for c in chunks)
|
||||
# The terminal error event keeps sequence numbers monotonic for strict clients.
|
||||
prior_sequence_numbers = [
|
||||
c.sequence_number
|
||||
for c in chunks
|
||||
if isinstance(getattr(c, "sequence_number", None), int) and c is not error_events[0]
|
||||
]
|
||||
assert error_events[0].sequence_number > max(prior_sequence_numbers)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_call_happy_path_emits_no_error_event(monkeypatch):
|
||||
"""Regression guard: the tool-call success path must stay error-free."""
|
||||
_mock_mcp_environment(monkeypatch)
|
||||
|
||||
aresponses_mock = AsyncMock(return_value=_text_only_stream("final answer"))
|
||||
monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
|
||||
|
||||
iterator = _make_tool_call_iterator()
|
||||
chunks = [chunk async for chunk in iterator]
|
||||
|
||||
assert all(getattr(c, "type", None) != ResponsesAPIStreamEvents.ERROR for c in chunks)
|
||||
completed = [c for c in chunks if getattr(c, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED]
|
||||
assert completed[-1].response.output[0]["content"][0]["text"] == "final answer"
|
||||
Loading…
Add table
Reference in a new issue