aditya
ee82d6101e
Implement MCPToolAdapter to connect to external MCP servers, enumerate tools, and register them in ToolRegistry with source="mcp". Includes connect/reconnect/disconnect lifecycle with timeout enforcement, input validation on invoke, capability inference, Behave/Robot/ASV tests, and docs/reference/mcp_adapter.md. ISSUES CLOSED: #159
154 lines
4.6 KiB
Python
154 lines
4.6 KiB
Python
"""ASV benchmarks for MCP Tool Adapter performance.
|
|
|
|
Measures the performance of:
|
|
- MCPServerConfig creation and validation
|
|
- Tool discovery from a mock transport
|
|
- Tool invocation with schema validation
|
|
- Tool registration into ToolRegistry
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
import cleveragents # noqa: E402
|
|
|
|
importlib.reload(cleveragents)
|
|
|
|
from cleveragents.mcp.adapter import ( # noqa: E402
|
|
MCPServerConfig,
|
|
MCPToolAdapter,
|
|
MCPTransport,
|
|
)
|
|
from cleveragents.tool.registry import ToolRegistry # noqa: E402
|
|
|
|
|
|
class _BenchTransport(MCPTransport):
|
|
"""Minimal transport for benchmarks — returns canned data."""
|
|
|
|
def __init__(self, tools: list[dict[str, Any]]) -> None:
|
|
self._tools = tools
|
|
|
|
def connect(self, config: MCPServerConfig) -> dict[str, Any]:
|
|
return {"capabilities": {"tools": True}}
|
|
|
|
def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
|
if method == "tools/list":
|
|
return {"tools": self._tools}
|
|
if method == "tools/call":
|
|
return {"content": {"status": "ok"}}
|
|
return {}
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
def _mock_tool(name: str, schema: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return {
|
|
"name": name,
|
|
"description": f"Bench tool {name}",
|
|
"inputSchema": schema or {},
|
|
}
|
|
|
|
|
|
_TOOLS_3 = [_mock_tool(f"tool_{i}") for i in range(3)]
|
|
_TOOLS_50 = [_mock_tool(f"tool_{i}") for i in range(50)]
|
|
_SCHEMA_TOOL = _mock_tool(
|
|
"validated",
|
|
{
|
|
"type": "object",
|
|
"properties": {"x": {"type": "integer"}, "y": {"type": "string"}},
|
|
"required": ["x"],
|
|
},
|
|
)
|
|
|
|
|
|
class TimeConfigCreation:
|
|
"""Benchmark MCPServerConfig instantiation."""
|
|
|
|
def time_create_stdio_config(self) -> None:
|
|
MCPServerConfig(name="bench", transport="stdio", command="echo")
|
|
|
|
def time_create_sse_config(self) -> None:
|
|
MCPServerConfig(
|
|
name="bench",
|
|
transport="sse",
|
|
url="https://example.com/mcp",
|
|
headers={"Authorization": "Bearer tok"},
|
|
)
|
|
|
|
|
|
class TimeToolDiscovery:
|
|
"""Benchmark tool discovery with varying tool counts."""
|
|
|
|
def setup(self) -> None:
|
|
self._config = MCPServerConfig(name="bench", transport="stdio", command="echo")
|
|
|
|
def time_discover_3_tools(self) -> None:
|
|
t = _BenchTransport(_TOOLS_3)
|
|
adapter = MCPToolAdapter(config=self._config, transport=t)
|
|
adapter.connect()
|
|
adapter.discover_tools()
|
|
|
|
def time_discover_50_tools(self) -> None:
|
|
t = _BenchTransport(_TOOLS_50)
|
|
adapter = MCPToolAdapter(config=self._config, transport=t)
|
|
adapter.connect()
|
|
adapter.discover_tools()
|
|
|
|
|
|
class TimeToolInvocation:
|
|
"""Benchmark tool invocation with and without schema validation."""
|
|
|
|
def setup(self) -> None:
|
|
config = MCPServerConfig(name="bench", transport="stdio", command="echo")
|
|
self._transport = _BenchTransport([_SCHEMA_TOOL])
|
|
self._adapter = MCPToolAdapter(config=config, transport=self._transport)
|
|
self._adapter.connect()
|
|
self._adapter.discover_tools()
|
|
|
|
def time_invoke_with_validation(self) -> None:
|
|
self._adapter.invoke("validated", {"x": 42, "y": "hello"})
|
|
|
|
def time_invoke_no_schema(self) -> None:
|
|
no_schema = _BenchTransport([_mock_tool("plain")])
|
|
config = MCPServerConfig(name="bench", transport="stdio", command="echo")
|
|
adapter = MCPToolAdapter(config=config, transport=no_schema)
|
|
adapter.connect()
|
|
adapter.discover_tools()
|
|
adapter.invoke("plain", {"any": "data"})
|
|
|
|
|
|
class TimeToolRegistration:
|
|
"""Benchmark registering discovered tools into ToolRegistry."""
|
|
|
|
def setup(self) -> None:
|
|
self._config = MCPServerConfig(name="bench", transport="stdio", command="echo")
|
|
|
|
def time_register_3_tools(self) -> None:
|
|
t = _BenchTransport(_TOOLS_3)
|
|
adapter = MCPToolAdapter(config=self._config, transport=t)
|
|
adapter.connect()
|
|
registry = ToolRegistry()
|
|
adapter.register_tools(registry, namespace="bench")
|
|
|
|
def time_register_50_tools(self) -> None:
|
|
t = _BenchTransport(_TOOLS_50)
|
|
adapter = MCPToolAdapter(config=self._config, transport=t)
|
|
adapter.connect()
|
|
registry = ToolRegistry()
|
|
adapter.register_tools(registry, namespace="bench")
|
|
|
|
|
|
time_config = TimeConfigCreation()
|
|
time_disc = TimeToolDiscovery()
|
|
time_invoke = TimeToolInvocation()
|
|
time_reg = TimeToolRegistration()
|