MCP Server

class AgentMCPServer:
    """Generic MCP server that wraps any MCPAgent subclass."""

    def __init__(
        self,
        agent_class: Type[MCPAgent],
        name: str = None,
        port: int = None,
        host: str = None,
        verbose: bool = False,
        agent_params: Dict[str, Any] = None,
        transport: Transport = "streamable-http",
        register_typed_tools: bool = False,
    ):
        """
        Initialize MCP server for an agent.

        Args:
            agent_class: MCPAgent subclass to wrap
            name: Display name (default: "GAIA {AgentName} MCP")
            port: Port to listen on (default: 8080). Ignored for stdio.
            host: Host to bind to (default: localhost). Ignored for stdio.
            verbose: Enable verbose logging
            agent_params: Parameters for agent __init__
            transport: Default transport for start() — "streamable-http"
                (binds a port) or "stdio" (JSON-RPC over stdin/stdout)
            register_typed_tools: When True, register each tool with an explicit
                typed signature derived from its inputSchema. When False (default),
                use the legacy **kwargs wrapper that tolerates VSCode/Copilot's
                {"kwargs": ...} envelope.

        Raises:
            TypeError: If agent_class doesn't inherit MCPAgent
        """
        pass

    def start(self, transport: Transport = None):
        """
        Start the MCP server.

        Uses the transport passed here, or the instance default from __init__.
        Prints the startup banner and blocks until Ctrl+C.
        """
        pass

    def stop(self):
        """Stop server (handled by KeyboardInterrupt with uvicorn)."""
        pass

    def _register_agent_tools(self):
        """Dynamically register agent tools with MCPServer."""
        pass

    def _print_startup_info(self):
        """Print startup banner with server info."""
        pass

# mcp 2.x takes host and port as run() arguments, not settings
self.mcp.run(transport="streamable-http", host="localhost", port=8080)

def _register_agent_tools(self):
    """Dynamically register agent tools with MCPServer."""

    tools = self.agent.get_mcp_tool_definitions()

    for tool_def in tools:
        tool_name = tool_def["name"]
        tool_description = tool_def.get("description", "")

        # Create wrapper function (captures tool_name in closure)
        def create_tool_wrapper(name: str, description: str, verbose: bool):
            async def tool_wrapper(**kwargs) -> Dict[str, Any]:
                """Dynamically generated tool wrapper."""

                if verbose:
                    logger.info(f"[MCP TOOL] Tool call: {name}")
                    logger.info(f"[MCP TOOL] Raw kwargs: {kwargs}")

                try:
                    # Handle VSCode kwargs wrapper
                    if "kwargs" in kwargs:
                        kwargs_value = kwargs["kwargs"]

                        if isinstance(kwargs_value, dict):
                            # Already dict, unwrap
                            kwargs = kwargs_value
                        elif isinstance(kwargs_value, str):
                            # Stringified JSON, parse
                            kwargs = json.loads(kwargs_value)

                    # Map parameter variations
                    if "app_dir" in kwargs and "appPath" not in kwargs:
                        kwargs["appPath"] = kwargs.pop("app_dir")

                    if "directory" in kwargs and "appPath" not in kwargs:
                        kwargs["appPath"] = kwargs.pop("directory")

                    if verbose:
                        logger.info(f"[MCP TOOL] Final args: {kwargs}")

                    # Execute tool
                    result = self.agent.execute_mcp_tool(name, kwargs)
                    return result

                except Exception as e:
                    logger.error(f"[MCP] Error executing tool {name}: {e}")
                    return {"error": str(e), "success": False}

            # Set metadata
            tool_wrapper.__name__ = name
            tool_wrapper.__doc__ = description
            return tool_wrapper

        # Create and register tool
        tool_func = create_tool_wrapper(tool_name, tool_description, self.verbose)
        self.mcp.tool()(tool_func)

        if self.verbose:
            logger.info(f"Registered tool: {tool_name}")

# VSCode sends: {"kwargs": {"appPath": "C:/path"}}
# or: {"kwargs": "{\"appPath\": \"C:/path\"}"}

if "kwargs" in kwargs:
    kwargs_value = kwargs["kwargs"]

    if isinstance(kwargs_value, dict):
        kwargs = kwargs_value  # Unwrap
    elif isinstance(kwargs_value, str):
        kwargs = json.loads(kwargs_value)  # Parse JSON string

# Map common variations
if "app_dir" in kwargs:
    kwargs["appPath"] = kwargs.pop("app_dir")

if "directory" in kwargs:
    kwargs["appPath"] = kwargs.pop("directory")

if "project_path" in kwargs:
    kwargs["appPath"] = kwargs.pop("project_path")

def _print_startup_info(self):
    """Print startup banner."""

    # Fix Windows Unicode
    if sys.platform == "win32":
        sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

    tools = self.agent.get_mcp_tool_definitions()

    print("=" * 60)
    print(f"🚀 {self.name}")
    print("=" * 60)
    print(f"Server: http://{self.host}:{self.port}")
    print(f"Agent: {self.agent_class.__name__}")
    print(f"Tools: {len(tools)}")
    for tool in tools:
        print(f"  - {tool['name']}: {tool.get('description', 'No description')}")
    if self.verbose:
        print(f"\n🔍 Verbose Mode: ENABLED")
    print("\n📍 MCP Endpoint:")
    print(f"  http://{self.host}:{self.port}/mcp")
    print("\n  Supports:")
    print("    - HTTP POST for requests")
    print("    - SSE streaming for real-time responses")
    print("=" * 60)
    print("\nPress Ctrl+C to stop\n")

# tests/mcp/test_agent_mcp_server.py

def test_server_initialization():
    """Test AgentMCPServer initializes."""
    from my_package.agent import GreeterAgent

    server = AgentMCPServer(
        agent_class=GreeterAgent,
        port=9999,
        agent_params={"silent_mode": True}
    )

    assert server.agent_class == GreeterAgent
    assert server.port == 9999

def test_server_rejects_non_mcp_agent():
    """Test server rejects non-MCPAgent classes."""
    from gaia.agents.base.agent import Agent

    with pytest.raises(TypeError, match="must inherit from MCPAgent"):
        AgentMCPServer(agent_class=Agent)

def test_tool_registration():
    """Test tools are registered from agent."""
    from my_package.agent import GreeterAgent

    server = AgentMCPServer(
        agent_class=GreeterAgent,
        agent_params={"silent_mode": True}
    )

    # Check tool was registered
    # MCPServer stores tools in internal registry
    tools = server.agent.get_mcp_tool_definitions()
    assert len(tools) > 0
    assert tools[0]["name"] == "greet"

@pytest.mark.asyncio
async def test_parameter_mapping():
    """Test VSCode parameter unwrapping."""
    from my_package.agent import GreeterAgent

    server = AgentMCPServer(
        agent_class=GreeterAgent,
        agent_params={"silent_mode": True}
    )

    # Simulate VSCode kwargs wrapper
    vscode_kwargs = {
        "kwargs": {"name": "Ada"}
    }

    # Tool wrapper should unwrap the nested kwargs before dispatch
    # (Testing internal logic, actual test would mock tool execution)

# Start server
python -m gaia.mcp.start_example_mcp --port 8081 --verbose

# Test with MCP client
curl -X POST http://localhost:8081/mcp/tools/greet \
  -H "Content-Type: application/json" \
  -d '{"name": "Ada"}'

# Verify response
{
  "message": "Hello, Ada!"
}

# MCP Python SDK (mcp 2.x)
from mcp.server import MCPServer

# GAIA
from gaia.agents.base.mcp_agent import MCPAgent
from gaia.logger import get_logger

from gaia.mcp.agent_mcp_server import AgentMCPServer
from my_package.agent import MyAgent

# Create server
server = AgentMCPServer(
    agent_class=MyAgent,
    name="My Agent MCP",
    port=8081,
    verbose=True,
    agent_params={
        "allowed_paths": ["/home/user/projects"],
        "silent_mode": False
    }
)

# Start (blocks until Ctrl+C)
server.start()

from typing import Any, Dict, List
from gaia.agents.base.mcp_agent import MCPAgent
from gaia.agents.base.console import SilentConsole
from gaia.mcp.agent_mcp_server import AgentMCPServer

class GreeterAgent(MCPAgent):
    def _get_system_prompt(self) -> str:
        return "You greet users."

    def _create_console(self):
        return SilentConsole()

    def _register_tools(self):
        pass

    def get_mcp_tool_definitions(self) -> List[Dict[str, Any]]:
        return [{
            "name": "greet",
            "description": "Greet a person by name",
            "inputSchema": {
                "type": "object",
                "properties": {"name": {"type": "string"}},
                "required": ["name"],
            },
        }]

    def execute_mcp_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        if tool_name == "greet":
            return {"message": f"Hello, {arguments['name']}!"}
        raise ValueError(f"Unknown tool: {tool_name}")

server = AgentMCPServer(
    agent_class=GreeterAgent,
    port=8081,
    verbose=False,
    agent_params={"silent_mode": True},
)

server.start()

# src/gaia/mcp/start_example_mcp.py
import argparse
from gaia.mcp.agent_mcp_server import AgentMCPServer
from my_package.agent import GreeterAgent

parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=8081)
parser.add_argument("--verbose", action="store_true")
args = parser.parse_args()

server = AgentMCPServer(
    agent_class=GreeterAgent,
    port=args.port,
    verbose=args.verbose
)

server.start()

python -m gaia.mcp.start_example_mcp --port 8081 --verbose