chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[router][grpc] Add more mcp test cases to responses api (#12749)

Browse Source
This commit is contained in:
Chang Su
2025-11-06 10:20:00 -08:00
committed by GitHub
parent 0c006b8809
commit 2cb42dc184
3 changed files with 266 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sgl-router/py_test/e2e_response_api/backends/test_grpc_backend.py
+15 -171
View File
@@ -23,7 +23,7 @@ from util import kill_process_tree
class TestGrpcBackend(StateManagementTests, MCPTests):
"""End to end tests for gRPC backend."""
"""End to end tests for gRPC backend (Regular backend with Llama)."""
@classmethod
def setUpClass(cls):
@@ -37,7 +37,7 @@ class TestGrpcBackend(StateManagementTests, MCPTests):
num_workers=1,
tp_size=2,
policy="round_robin",
router_args=["--history-backend", "memory"],
router_args=["--history-backend", "memory", "--tool-call-parser", "llama"],
)
cls.base_url = cls.cluster["base_url"]
@@ -48,9 +48,6 @@ class TestGrpcBackend(StateManagementTests, MCPTests):
for worker in cls.cluster.get("workers", []):
kill_process_tree(worker.pid)
@unittest.skip(
"TODO: transport error, details: [], metadata: MetadataMap { headers: {} }"
)
def test_previous_response_id_chaining(self):
super().test_previous_response_id_chaining()
@@ -62,18 +59,17 @@ class TestGrpcBackend(StateManagementTests, MCPTests):
def test_mutually_exclusive_parameters(self):
super().test_mutually_exclusive_parameters()
@unittest.skip(
"TODO: Pipeline execution failed: Pipeline stage WorkerSelection failed"
)
def test_mcp_basic_tool_call(self):
super().test_mcp_basic_tool_call()
@unittest.skip("TODO: no event fields")
def test_mcp_basic_tool_call_streaming(self):
return super().test_mcp_basic_tool_call_streaming()
# Inherited from MCPTests:
# - test_mcp_basic_tool_call
# - test_mcp_basic_tool_call_streaming
# - test_mixed_mcp_and_function_tools (requires external MCP server)
# - test_mixed_mcp_and_function_tools_streaming (requires external MCP server)
class TestHarmonyBackend(StateManagementTests, MCPTests, FunctionCallingBaseTest):
class TestGrpcHarmonyBackend(StateManagementTests, MCPTests, FunctionCallingBaseTest):
"""End to end tests for Harmony backend."""
@classmethod
@@ -108,168 +104,16 @@ class TestHarmonyBackend(StateManagementTests, MCPTests, FunctionCallingBaseTest
def test_mutually_exclusive_parameters(self):
super().test_mutually_exclusive_parameters()
def test_mcp_basic_tool_call(self):
"""Test basic MCP tool call (non-streaming)."""
tools = [
{
"type": "mcp",
"server_label": "deepwiki",
"server_url": "https://mcp.deepwiki.com/mcp",
"require_approval": "never",
}
]
resp = self.create_response(
"What transport protocols does the 2025-03-26 version of the MCP spec (modelcontextprotocol/modelcontextprotocol) support?",
tools=tools,
stream=False,
)
# Should successfully make the request
self.assertEqual(resp.status_code, 200)
data = resp.json()
# Basic response structure
self.assertIn("id", data)
self.assertIn("status", data)
self.assertEqual(data["status"], "completed")
self.assertIn("output", data)
self.assertIn("model", data)
# Verify output array is not empty
output = data["output"]
self.assertIsInstance(output, list)
self.assertGreater(len(output), 0)
# Check for MCP-specific output types
output_types = [item.get("type") for item in output]
# Should have mcp_list_tools - tools are listed before calling
self.assertIn(
"mcp_list_tools", output_types, "Response should contain mcp_list_tools"
)
# Should have at least one mcp_call
mcp_calls = [item for item in output if item.get("type") == "mcp_call"]
self.assertGreater(
len(mcp_calls), 0, "Response should contain at least one mcp_call"
)
# Verify mcp_call structure
for mcp_call in mcp_calls:
self.assertIn("id", mcp_call)
self.assertIn("status", mcp_call)
self.assertEqual(mcp_call["status"], "completed")
self.assertIn("server_label", mcp_call)
self.assertEqual(mcp_call["server_label"], "deepwiki")
self.assertIn("name", mcp_call)
self.assertIn("arguments", mcp_call)
self.assertIn("output", mcp_call)
def test_mcp_basic_tool_call_streaming(self):
"""Test basic MCP tool call (streaming)."""
tools = [
{
"type": "mcp",
"server_label": "deepwiki",
"server_url": "https://mcp.deepwiki.com/mcp",
"require_approval": "never",
}
]
resp = self.create_response(
"What transport protocols does the 2025-03-26 version of the MCP spec (modelcontextprotocol/modelcontextprotocol) support?",
tools=tools,
stream=True,
)
# Should successfully make the request
self.assertEqual(resp.status_code, 200)
events = self.parse_sse_events(resp)
self.assertGreater(len(events), 0)
event_types = [e.get("event") for e in events]
# Check for lifecycle events
self.assertIn(
"response.created", event_types, "Should have response.created event"
)
self.assertIn(
"response.completed", event_types, "Should have response.completed event"
)
# Check for MCP list tools events
self.assertIn(
"response.output_item.added",
event_types,
"Should have output_item.added events",
)
self.assertIn(
"response.mcp_list_tools.in_progress",
event_types,
"Should have mcp_list_tools.in_progress event",
)
self.assertIn(
"response.mcp_list_tools.completed",
event_types,
"Should have mcp_list_tools.completed event",
)
# Check for MCP call events
self.assertIn(
"response.mcp_call.in_progress",
event_types,
"Should have mcp_call.in_progress event",
)
self.assertIn(
"response.mcp_call_arguments.delta",
event_types,
"Should have mcp_call_arguments.delta event",
)
self.assertIn(
"response.mcp_call_arguments.done",
event_types,
"Should have mcp_call_arguments.done event",
)
self.assertIn(
"response.mcp_call.completed",
event_types,
"Should have mcp_call.completed event",
)
# Verify final completed event has full response
completed_events = [e for e in events if e.get("event") == "response.completed"]
self.assertEqual(len(completed_events), 1)
final_response = completed_events[0].get("data", {}).get("response", {})
self.assertIn("id", final_response)
self.assertEqual(final_response.get("status"), "completed")
self.assertIn("output", final_response)
# Verify final output contains expected items
final_output = final_response.get("output", [])
final_output_types = [item.get("type") for item in final_output]
self.assertIn("mcp_list_tools", final_output_types)
self.assertIn("mcp_call", final_output_types)
# Verify mcp_call items in final output
mcp_calls = [item for item in final_output if item.get("type") == "mcp_call"]
self.assertGreater(len(mcp_calls), 0)
for mcp_call in mcp_calls:
self.assertEqual(mcp_call.get("status"), "completed")
self.assertEqual(mcp_call.get("server_label"), "deepwiki")
self.assertIn("name", mcp_call)
self.assertIn("arguments", mcp_call)
self.assertIn("output", mcp_call)
@unittest.skip("TODO: 501 Not Implemented")
def test_conversation_with_multiple_turns(self):
super().test_conversation_with_multiple_turns()
# Inherited from MCPTests:
# - test_mcp_basic_tool_call
# - test_mcp_basic_tool_call_streaming
# - test_mixed_mcp_and_function_tools (requires external MCP server)
# - test_mixed_mcp_and_function_tools_streaming (requires external MCP server)
if __name__ == "__main__":
unittest.main()
sgl-router/py_test/e2e_response_api/backends/test_http_backend.py
+19
View File
@@ -36,6 +36,7 @@ class TestOpenaiBackend(
"""End to end tests for OpenAI backend."""
api_key = os.environ.get("OPENAI_API_KEY")
mcp_validation_mode = "strict" # Enable strict validation for HTTP backend
@classmethod
def setUpClass(cls):
@@ -54,6 +55,24 @@ class TestOpenaiBackend(
def tearDownClass(cls):
kill_process_tree(cls.cluster["router"].pid)
# Inherited from MCPTests:
# - test_mcp_basic_tool_call (with strict validation)
# - test_mcp_basic_tool_call_streaming (with strict validation)
# - test_mixed_mcp_and_function_tools (requires external MCP server)
# - test_mixed_mcp_and_function_tools_streaming (requires external MCP server)
@unittest.skip(
"Requires external MCP server (deepwiki) - may not be accessible in CI"
)
def test_mixed_mcp_and_function_tools(self):
super().test_mixed_mcp_and_function_tools()
@unittest.skip(
"Requires external MCP server (deepwiki) - may not be accessible in CI"
)
def test_mixed_mcp_and_function_tools_streaming(self):
super().test_mixed_mcp_and_function_tools_streaming()
class TestXaiBackend(StateManagementTests):
"""End to end tests for XAI backend."""