cleveragents-core/features/mocks/mcp_stub_server.py

"""MCP stub server mock for M2 actor + tool source smoke tests.

Simulates MCP tool discovery and invocation in-process without requiring
a real MCP adapter.  Returns deterministic tool descriptors and execution
results for Behave-level testing.

This mock lives in ``features/mocks/`` per ADR-022 (no mocks in src/).
"""

from __future__ import annotations

from typing import Any, ClassVar


class McpStubTool:
    """Descriptor for a single tool exposed by the MCP stub server."""

    def __init__(
        self,
        name: str,
        description: str,
        input_schema: dict[str, Any] | None = None,
        output_schema: dict[str, Any] | None = None,
    ) -> None:
        self.name = name
        self.description = description
        self.input_schema = input_schema or {
            "type": "object",
            "properties": {"query": {"type": "string"}},
        }
        self.output_schema = output_schema or {
            "type": "object",
            "properties": {"result": {"type": "string"}},
        }


class McpStubServer:
    """In-process MCP stub server for deterministic testing.

    Provides ``discover()`` and ``invoke()`` methods that mirror the
    future MCP adapter interface (#159) without requiring network I/O.

    Usage::

        server = McpStubServer()
        server.start()
        tools = server.discover()
        result = server.invoke("mcp/search", {"query": "hello"})
        server.stop()
    """

    #: Default tools returned by ``discover()``.
    DEFAULT_TOOLS: ClassVar[list[McpStubTool]] = [
        McpStubTool(
            name="mcp/search",
            description="Stub MCP search tool",
            input_schema={
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
            output_schema={
                "type": "object",
                "properties": {
                    "results": {"type": "array", "items": {"type": "string"}}
                },
            },
        ),
        McpStubTool(
            name="mcp/fetch",
            description="Stub MCP fetch tool",
            input_schema={
                "type": "object",
                "properties": {"url": {"type": "string"}},
                "required": ["url"],
            },
            output_schema={
                "type": "object",
                "properties": {
                    "content": {"type": "string"},
                    "status": {"type": "integer"},
                },
            },
        ),
        McpStubTool(
            name="mcp/transform",
            description="Stub MCP transform tool",
            input_schema={
                "type": "object",
                "properties": {
                    "data": {"type": "string"},
                    "format": {"type": "string"},
                },
                "required": ["data"],
            },
            output_schema={
                "type": "object",
                "properties": {"transformed": {"type": "string"}},
            },
        ),
    ]

    def __init__(self, tools: list[McpStubTool] | None = None) -> None:
        self._tools = tools or list(self.DEFAULT_TOOLS)
        self._running = False
        self._invocation_log: list[dict[str, Any]] = []

    @property
    def is_running(self) -> bool:
        """Whether the stub server is currently started."""
        return self._running

    @property
    def invocation_log(self) -> list[dict[str, Any]]:
        """Log of all invocations since the last ``start()``."""
        return list(self._invocation_log)

    def start(self) -> None:
        """Start the stub server (in-process, no network)."""
        self._running = True
        self._invocation_log = []

    def stop(self) -> None:
        """Stop the stub server."""
        self._running = False

    def discover(self) -> list[McpStubTool]:
        """Return the list of available tools.

        Raises:
            RuntimeError: If the server is not running.
        """
        if not self._running:
            msg = "McpStubServer is not running. Call start() first."
            raise RuntimeError(msg)
        return list(self._tools)

    def invoke(self, tool_name: str, params: dict[str, Any]) -> dict[str, Any]:
        """Invoke a stub tool by name with deterministic results.

        Raises:
            RuntimeError: If the server is not running.
            KeyError: If the tool name is not found.
        """
        if not self._running:
            msg = "McpStubServer is not running. Call start() first."
            raise RuntimeError(msg)

        tool = self._find_tool(tool_name)
        if tool is None:
            msg = f"Tool '{tool_name}' not found in MCP stub server."
            raise KeyError(msg)

        # Generate deterministic result based on tool name
        result = self._generate_result(tool, params)
        self._invocation_log.append(
            {
                "tool_name": tool_name,
                "params": params,
                "result": result,
            }
        )
        return result

    def _find_tool(self, name: str) -> McpStubTool | None:
        """Find a tool by name."""
        for tool in self._tools:
            if tool.name == name:
                return tool
        return None

    def _generate_result(
        self, tool: McpStubTool, params: dict[str, Any]
    ) -> dict[str, Any]:
        """Generate a deterministic result for a tool invocation."""
        if tool.name == "mcp/search":
            query = params.get("query", "")
            return {
                "results": [f"result_1_for_{query}", f"result_2_for_{query}"],
            }
        if tool.name == "mcp/fetch":
            url = params.get("url", "")
            return {
                "content": f"<html>stub content for {url}</html>",
                "status": 200,
            }
        if tool.name == "mcp/transform":
            data = params.get("data", "")
            fmt = params.get("format", "upper")
            transformed = data.upper() if fmt == "upper" else data.lower()
            return {"transformed": transformed}

        # Generic fallback
        return {"result": f"stub result from {tool.name}"}