test(mcp): add server-side execution test case for 100% codecov coverage

This commit is contained in:
Arjun Pakhan 2026-08-19 19:13:04 +00:00
parent 9a4ef077d3
commit 5746f4b0f5
2 changed files with 78 additions and 4 deletions

View file

@ -64,17 +64,17 @@ def _has_client_side_tool(
tool_use_blocks: Sequence[Mapping[str, object]],
other_tools: Sequence[Mapping[str, object]] | None,
tool_server_map: Mapping[str, str],
) -> bool: # kwargs-ok: helper function inspecting tool blocks for client passthrough
) -> bool: # kwargs-ok: helper inspecting tool blocks for client passthrough
client_tool_names = {
t.get("name") # kwargs-ok: extract client tool name
t.get("name") # kwargs-ok: extract client tool name from dictionary
for t in (other_tools or ())
if isinstance(t, dict) and t.get("name") # kwargs-ok: extract client tool name
}
for block in tool_use_blocks:
name = block.get("name") # kwargs-ok: extract block tool name
name = block.get("name") # kwargs-ok: extract block tool name from dictionary
if name in client_tool_names or (
bool(tool_server_map) and name not in tool_server_map
): # kwargs-ok: check map membership
): # kwargs-ok: map membership test
return True
return False

View file

@ -77,3 +77,77 @@ async def test_mcp_auto_execute_bypasses_client_side_tools():
mock_execute.assert_not_called()
assert response == mock_anthropic_response
@pytest.mark.asyncio
async def test_mcp_auto_execute_runs_server_side_mcp_tools():
"""
Ensure that if a response contains ONLY server-side MCP tools,
auto-execution proceeds as expected and returns False for client-side tool check.
"""
mock_mcp_references = [{"type": "mcp", "server_url": "http://localhost/mcp", "require_approval": "never"}]
mock_mcp_tools = [SimpleNamespace(name="mcp_tool_1", description="MCP Tool", inputSchema={"type": "object"})]
mock_tool_server_map = {"mcp_tool_1": "http://localhost/mcp"}
mock_anthropic_response = {
"id": "msg_123",
"type": "message",
"role": "assistant",
"content": [
{"type": "tool_use", "id": "call_mcp", "name": "mcp_tool_1", "input": {}},
],
"stop_reason": "tool_use",
}
mock_context = MagicMock()
mock_context.user_api_key_auth = None
mock_context.litellm_trace_id = "trace_123"
mock_context.mcp_auth_header = None
mock_context.mcp_server_auth_headers = None
mock_context.request_tags = None
mock_context.oauth2_headers = None
mock_context.raw_headers = None
mock_context.litellm_call_id = "call_123"
path_resolve = "litellm.responses.mcp.request_context.MCPRequestContext.resolve"
path_parse = "litellm.responses.mcp.litellm_proxy_mcp_handler.LiteLLM_Proxy_MCP_Handler._parse_mcp_tools"
path_process = (
"litellm.responses.mcp.litellm_proxy_mcp_handler."
"LiteLLM_Proxy_MCP_Handler._process_mcp_tools_without_openai_transform"
)
path_auto = "litellm.responses.mcp.litellm_proxy_mcp_handler.LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools"
path_exec = "litellm.responses.mcp.litellm_proxy_mcp_handler.LiteLLM_Proxy_MCP_Handler._execute_tool_calls"
path_call = "litellm.llms.anthropic.experimental_pass_through.messages.mcp_handler._AnthropicMessagesCall"
with (
patch(path_resolve, return_value=mock_context),
patch(path_parse, return_value=(mock_mcp_references, [])),
patch(path_process, new_callable=AsyncMock) as mock_process,
patch(path_auto, return_value=True),
patch(path_exec, new_callable=AsyncMock) as mock_execute,
patch(path_call) as mock_call,
):
mock_process.return_value = (mock_mcp_tools, mock_tool_server_map)
mock_execute.return_value = [{"tool_call_id": "call_mcp", "result": "ok"}]
mock_final_response = {
"id": "msg_124",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "done"}],
"stop_reason": "end_turn",
}
mock_fn = AsyncMock(side_effect=[mock_anthropic_response, mock_final_response])
mock_call.return_value.fn = mock_fn
response = await anthropic_messages_with_mcp(
max_tokens=100,
messages=[{"role": "user", "content": "run mcp_tool_1"}],
model="claude-3-5-sonnet-20241022",
tools=mock_mcp_references,
)
mock_execute.assert_called_once()
assert response == mock_final_response