feat(mcp_client): Implement ExternalMCPManager and StdioMCPServer with tests
This commit is contained in:
55
tests/test_external_mcp.py
Normal file
55
tests/test_external_mcp.py
Normal file
@@ -0,0 +1,55 @@
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
import pytest
|
||||
from src import mcp_client
|
||||
from src import models
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_external_mcp_real_process():
|
||||
manager = mcp_client.ExternalMCPManager()
|
||||
|
||||
# Use our mock script
|
||||
mock_script = "scripts/mock_mcp_server.py"
|
||||
config = models.MCPServerConfig(
|
||||
name="real-mock",
|
||||
command="python",
|
||||
args=[mock_script]
|
||||
)
|
||||
|
||||
await manager.add_server(config)
|
||||
|
||||
try:
|
||||
tools = manager.get_all_tools()
|
||||
assert "echo" in tools
|
||||
assert tools["echo"]["server"] == "real-mock"
|
||||
|
||||
result = await manager.async_dispatch("echo", {"hello": "world"})
|
||||
assert "ECHO: {'hello': 'world'}" in result
|
||||
finally:
|
||||
await manager.stop_all()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_tool_schemas_includes_external():
|
||||
manager = mcp_client.get_external_mcp_manager()
|
||||
# Reset manager
|
||||
await manager.stop_all()
|
||||
|
||||
mock_script = "scripts/mock_mcp_server.py"
|
||||
config = models.MCPServerConfig(
|
||||
name="test-server",
|
||||
command="python",
|
||||
args=[mock_script]
|
||||
)
|
||||
|
||||
await manager.add_server(config)
|
||||
|
||||
try:
|
||||
schemas = mcp_client.get_tool_schemas()
|
||||
echo_schema = next((s for s in schemas if s["name"] == "echo"), None)
|
||||
|
||||
assert echo_schema is not None
|
||||
assert echo_schema["description"] == "Echo input"
|
||||
assert echo_schema["parameters"] == {"type": "object"}
|
||||
finally:
|
||||
await manager.stop_all()
|
||||
Reference in New Issue
Block a user