cleveragents-core/robot/helper_mcp_adapter.py

"""Robot Framework helper for MCP Tool Adapter smoke tests.

Usage:
    python robot/helper_mcp_adapter.py create-adapter
    python robot/helper_mcp_adapter.py discover-tools
    python robot/helper_mcp_adapter.py invoke-tool
    python robot/helper_mcp_adapter.py register-tools
    python robot/helper_mcp_adapter.py reject-no-command
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

_ROOT = str(Path(__file__).resolve().parents[1])
_SRC = str(Path(__file__).resolve().parents[1] / "src")
for _p in (_SRC, _ROOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from features.mocks.mock_mcp_transport import MockMCPTransport  # noqa: E402

from cleveragents.mcp.adapter import MCPServerConfig, MCPToolAdapter  # noqa: E402
from cleveragents.tool.registry import ToolRegistry  # noqa: E402


def _mock_tool(name: str) -> dict[str, Any]:
    return {"name": name, "description": f"Mock {name}", "inputSchema": {}}


def main() -> int:
    if len(sys.argv) < 2:
        print("Usage: helper_mcp_adapter.py <command>")
        return 1

    dispatch = {
        "create-adapter": _create_adapter,
        "discover-tools": _discover_tools,
        "invoke-tool": _invoke_tool,
        "register-tools": _register_tools,
        "reject-no-command": _reject_no_command,
    }
    handler = dispatch.get(sys.argv[1])
    if handler is None:
        print(f"Unknown command: {sys.argv[1]}")
        return 1
    return handler()


def _create_adapter() -> int:
    config = MCPServerConfig(name="github", transport="stdio", command="echo")
    adapter = MCPToolAdapter(config=config)
    print(f"server-name: {adapter.server_name}")
    print(f"transport: {adapter.transport_type}")
    print(f"connected: {adapter.is_connected}")
    return 0


def _discover_tools() -> int:
    tools = [
        _mock_tool("create_issue"),
        _mock_tool("list_repos"),
        _mock_tool("search"),
    ]
    config = MCPServerConfig(name="test", transport="stdio", command="echo")
    transport = MockMCPTransport(tools=tools)
    adapter = MCPToolAdapter(config=config, transport=transport)
    adapter.connect()
    discovered = adapter.discover_tools()
    print(f"discovered: {len(discovered)}")
    for i, t in enumerate(discovered):
        print(f"tool-{i}: {t.name}")
    return 0


def _invoke_tool() -> int:
    tools = [_mock_tool("create_issue")]
    config = MCPServerConfig(name="test", transport="stdio", command="echo")
    transport = MockMCPTransport(
        tools=tools, invoke_results={"create_issue": {"id": 42}}
    )
    adapter = MCPToolAdapter(config=config, transport=transport)
    adapter.connect()
    adapter.discover_tools()
    result = adapter.invoke("create_issue", {"title": "Bug"})
    print(f"success: {result.success}")
    print(f"has-data: {bool(result.data)}")
    return 0


def _register_tools() -> int:
    tools = [_mock_tool("tool_a"), _mock_tool("tool_b")]
    config = MCPServerConfig(name="test", transport="stdio", command="echo")
    transport = MockMCPTransport(tools=tools)
    adapter = MCPToolAdapter(config=config, transport=transport)
    adapter.connect()
    registry = ToolRegistry()
    names = adapter.register_tools(registry, namespace="mcp-test")
    print(f"registered: {len(names)}")
    if names and "/" in names[0]:
        print(f"prefix: {names[0].split('/')[0]}/")
    return 0


def _reject_no_command() -> int:
    config = MCPServerConfig(name="bad", transport="stdio")
    try:
        MCPToolAdapter.from_server_config(config)
        print("unexpected-success")
        return 1
    except ValueError as exc:
        err = str(exc).lower()
        if "command" in err:
            print("validation-error: command")
        else:
            print(f"validation-error-other: {exc}")
        return 0


if __name__ == "__main__":
    sys.exit(main())