Added MCP

agent/__init__.py

@@ -1,9 +1,16 @@
 from .agent import Agent
-from .tools import Tool, fetch_webpage, FETCH_WEBPAGE_TOOL
+from .tools import Tool, tool, fetch_webpage, FETCH_WEBPAGE_TOOL
+from .mcp import MCPClient, mcp_tool, load_mcp_tools, load_all_mcp_tools, MCP_SERVERS
 
 __all__ = [
     "Agent",
     "Tool",
+    "tool",
     "fetch_webpage",
     "FETCH_WEBPAGE_TOOL",
+    "MCPClient",
+    "mcp_tool",
+    "load_mcp_tools",
+    "load_all_mcp_tools",
+    "MCP_SERVERS",
 ]

agent/agent.py

@@ -4,6 +4,7 @@ from typing import Any, Generator
 from openai import OpenAI
 
 from .tools import Tool
+from .mcp import MCPClient, load_mcp_tools, load_all_mcp_tools
 
 # ---------------------------------------------------------------------------
 # Streaming event contract
@@ -49,6 +50,25 @@ class Agent:
     def register_tool(self, tool: Tool) -> None:
         self._tools.append(tool)
 
+    def register_mcp(self, url: str, headers: dict[str, str] | None = None) -> list[Tool]:
+        """Connect to an MCP server and register all its tools.
+
+        Returns the list of registered Tool instances.
+        """
+        tools = load_mcp_tools(url=url, headers=headers)
+        self._tools.extend(tools)
+        return tools
+
+    def register_all_mcp(self) -> dict[str, list[Tool]]:
+        """Load tools from all pre-configured MCP servers.
+
+        Returns ``{server_name: [Tool, ...]}`` and registers them.
+        """
+        all_tools = load_all_mcp_tools()
+        for server_tools in all_tools.values():
+            self._tools.extend(server_tools)
+        return all_tools
+
     def register_final_message_tool(self) -> None:
         """Register a no-input ``final_message`` tool the model **must** call
         to signal that it is done.

agent/mcp.py

@@ -0,0 +1,261 @@
+"""MCP (Model Context Protocol) client for remote tool servers.
+
+Supports Streamable HTTP transport (JSON-RPC 2.0 over HTTP).
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from typing import Any
+
+import httpx
+
+from .tools import Tool
+
+
+# ---------------------------------------------------------------------------
+# MCP Client
+# ---------------------------------------------------------------------------
+
+
+class MCPClient:
+    """Connect to a remote MCP server via Streamable HTTP transport."""
+
+    def __init__(
+        self,
+        url: str,
+        headers: dict[str, str] | None = None,
+        timeout: float = 30.0,
+    ) -> None:
+        self.url = url
+        self.headers = headers or {}
+        self.timeout = timeout
+        self._session_id: str | None = None
+        self._initialized = False
+
+    # ------------------------------------------------------------------
+    # Low-level JSON-RPC
+    # ------------------------------------------------------------------
+
+    def _request(self, method: str, params: dict[str, Any] | None = None) -> Any:
+        """Send a JSON-RPC request and return the result."""
+        payload: dict[str, Any] = {
+            "jsonrpc": "2.0",
+            "id": str(uuid.uuid4()),
+            "method": method,
+        }
+        if params is not None:
+            payload["params"] = params
+
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json, text/event-stream",
+            **self.headers,
+        }
+        if self._session_id:
+            headers["Mcp-Session-Id"] = self._session_id
+
+        with httpx.Client(timeout=self.timeout) as client:
+            resp = client.post(self.url, json=payload, headers=headers)
+
+            # Capture session ID from response
+            sid = resp.headers.get("mcp-session-id")
+            if sid:
+                self._session_id = sid
+
+            resp.raise_for_status()
+            content_type = resp.headers.get("content-type", "")
+
+            # Handle SSE response
+            if "text/event-stream" in content_type:
+                return self._parse_sse_response(resp.text)
+
+            # Handle JSON response
+            data = resp.json()
+            if "error" in data:
+                raise RuntimeError(f"MCP error: {data['error']}")
+            return data.get("result")
+
+    def _parse_sse_response(self, text: str) -> Any:
+        """Parse SSE response text, extracting JSON-RPC results."""
+        result = None
+        for line in text.split("\n"):
+            line = line.strip()
+            if line.startswith("data:"):
+                data_str = line[len("data:") :].strip()
+                if data_str:
+                    try:
+                        data = json.loads(data_str)
+                        if "result" in data:
+                            result = data["result"]
+                        elif "error" in data:
+                            raise RuntimeError(f"MCP error: {data['error']}")
+                    except json.JSONDecodeError:
+                        continue
+        return result
+
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+
+    def initialize(self) -> dict[str, Any]:
+        """Initialize the MCP session."""
+        result = self._request(
+            "initialize",
+            {
+                "protocolVersion": "2025-03-26",
+                "capabilities": {},
+                "clientInfo": {"name": "mythos-agent", "version": "0.1.0"},
+            },
+        )
+        self._initialized = True
+        # Send initialized notification (no response expected)
+        self._notify("notifications/initialized", {})
+        return result or {}
+
+    def _notify(self, method: str, params: dict[str, Any]) -> None:
+        """Send a JSON-RPC notification (no id, no response expected)."""
+        payload = {"jsonrpc": "2.0", "method": method, "params": params}
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json, text/event-stream",
+            **self.headers,
+        }
+        if self._session_id:
+            headers["Mcp-Session-Id"] = self._session_id
+
+        with httpx.Client(timeout=self.timeout) as client:
+            client.post(self.url, json=payload, headers=headers)
+
+    # ------------------------------------------------------------------
+    # Tool discovery
+    # ------------------------------------------------------------------
+
+    def list_tools(self) -> list[dict[str, Any]]:
+        """List available tools from the server."""
+        if not self._initialized:
+            self.initialize()
+        result = self._request("tools/list", {})
+        return (result or {}).get("tools", [])
+
+    # ------------------------------------------------------------------
+    # Tool execution
+    # ------------------------------------------------------------------
+
+    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
+        """Call a tool on the server."""
+        if not self._initialized:
+            self.initialize()
+        return self._request("tools/call", {"name": name, "arguments": arguments})
+
+
+# ---------------------------------------------------------------------------
+# MCP Tool wrapper
+# ---------------------------------------------------------------------------
+
+
+def _json_schema_type_to_python(type_str: str) -> type:
+    """Map JSON Schema type to Python type."""
+    mapping = {"string": str, "integer": int, "number": float, "boolean": bool, "array": list, "object": dict}
+    return mapping.get(type_str, str)
+
+
+def mcp_tool(client: MCPClient, mcp_tool_def: dict[str, Any]) -> Tool:
+    """Wrap an MCP tool definition as a local Tool instance.
+
+    Args:
+        client: Connected MCPClient.
+        mcp_tool_def: Tool dict from ``tools/list`` response.
+    """
+    name = mcp_tool_def["name"]
+    description = mcp_tool_def.get("description", "")
+    input_schema = mcp_tool_def.get("inputSchema", {})
+
+    # Convert inputSchema to OpenAI-style parameters
+    properties: dict[str, dict] = {}
+    for param_name, param_def in input_schema.get("properties", {}).items():
+        prop: dict[str, Any] = {"type": param_def.get("type", "string")}
+        if "description" in param_def:
+            prop["description"] = param_def["description"]
+        if "enum" in param_def:
+            prop["enum"] = param_def["enum"]
+        properties[param_name] = prop
+
+    required = input_schema.get("required", [])
+
+    parameters = {
+        "type": "object",
+        "properties": properties,
+        "required": required,
+    }
+
+    def handler(**kwargs: Any) -> str:
+        result = client.call_tool(name, kwargs)
+        # Extract text content from MCP result
+        if isinstance(result, dict) and "content" in result:
+            parts = []
+            for block in result["content"]:
+                if isinstance(block, dict) and block.get("type") == "text":
+                    parts.append(block.get("text", ""))
+            return "\n".join(parts) if parts else json.dumps(result)
+        return json.dumps(result) if not isinstance(result, str) else result
+
+    return Tool(name=name, description=description, parameters=parameters, handler=handler)
+
+
+# ---------------------------------------------------------------------------
+# Convenience: connect and load all tools
+# ---------------------------------------------------------------------------
+
+
+def load_mcp_tools(url: str, headers: dict[str, str] | None = None) -> list[Tool]:
+    """Connect to an MCP server and return all its tools as Tool instances."""
+    client = MCPClient(url=url, headers=headers)
+    client.initialize()
+    tools_defs = client.list_tools()
+    return [mcp_tool(client, td) for td in tools_defs]
+
+
+# ---------------------------------------------------------------------------
+# Pre-configured MCP servers
+# ---------------------------------------------------------------------------
+
+MCP_SERVERS: dict[str, dict[str, Any]] = {
+    "exa": {
+        "url": "https://mcp.exa.ai/mcp",
+        "description": "Web search via Exa",
+    },
+    "context7": {
+        "url": "https://mcp.context7.com/mcp",
+        "description": "Library documentation lookup",
+    },
+    "grep_app": {
+        "url": "https://mcp.grep.app",
+        "description": "GitHub code search via grep.app",
+    },
+}
+
+
+def load_all_mcp_tools(
+    servers: dict[str, dict[str, Any]] | None = None,
+) -> dict[str, list[Tool]]:
+    """Load tools from all configured MCP servers.
+
+    Returns:
+        ``{server_name: [Tool, ...]}``
+    """
+    if servers is None:
+        servers = MCP_SERVERS
+
+    result: dict[str, list[Tool]] = {}
+    for name, cfg in servers.items():
+        try:
+            result[name] = load_mcp_tools(
+                url=cfg["url"],
+                headers=cfg.get("headers"),
+            )
+        except Exception as e:
+            print(f"[MCP] Failed to load {name}: {e}")
+            result[name] = []
+    return result

agent/tools.py

@@ -1,7 +1,21 @@
 from markitdown import MarkItDown
-from typing import Any, Callable
+from typing import Any, Callable, get_type_hints
+import inspect
 import requests
 
+
+def python_type_to_json_schema(tp: type) -> str:
+    """Map a Python type to a JSON Schema type string."""
+    mapping = {
+        str: "string",
+        int: "integer",
+        float: "number",
+        bool: "boolean",
+        list: "array",
+        dict: "object",
+    }
+    return mapping.get(tp, "string")
+
 class Tool:
     """A callable tool the agent can invoke."""
 
@@ -27,8 +41,102 @@ class Tool:
         return self.handler(**kwargs)
 
 
+def _parse_docstring(docstring: str) -> tuple[str, dict[str, tuple[bool, str]]]:
+    """Parse a tool docstring into description and param metadata.
+
+    Returns:
+        (description, {param_name: (required, description)})
+
+    Expected format:
+        First line: tool description.
+        Subsequent lines: ``param_name (required): description`` or
+        ``param_name: description``.
+    """
+    lines = (docstring or "").strip().split("\n")
+    description = lines[0].strip()
+    param_info: dict[str, tuple[bool, str]] = {}
+
+    for line in lines[1:]:
+        line = line.strip()
+        if not line:
+            continue
+        # Match: param_name (required): description
+        #    or: param_name: description
+        if ":" not in line:
+            continue
+        key, desc = line.split(":", 1)
+        key = key.strip()
+        desc = desc.strip()
+        required = False
+        if key.endswith("(required)"):
+            required = True
+            key = key[: -len("(required)")].strip()
+        if key:
+            param_info[key] = (required, desc)
+
+    return description, param_info
+
+
+def tool(fn: Callable[..., str]) -> Tool:
+    """Decorator that converts a function into a Tool instance.
+
+    Extracts name, description (first line of docstring), and parameters
+    from the function's type hints and signature.
+
+    Docstring format:
+        First line: tool description.
+        Subsequent lines: ``param_name (required): description`` or
+        ``param_name: description``.
+    """
+    name = fn.__name__
+    docstring = fn.__doc__ or ""
+    description, param_info = _parse_docstring(docstring)
+
+    hints = get_type_hints(fn)
+    sig = inspect.signature(fn)
+
+    properties: dict[str, dict] = {}
+    required: list[str] = []
+
+    for param_name, param in sig.parameters.items():
+        if param_name in hints:
+            param_schema: dict[str, Any] = {
+                "type": python_type_to_json_schema(hints[param_name])
+            }
+            # Enrich with docstring info if present
+            if param_name in param_info:
+                doc_required, doc_desc = param_info[param_name]
+                if doc_desc:
+                    param_schema["description"] = doc_desc
+                # Docstring (required) overrides signature default check
+                if doc_required:
+                    required.append(param_name)
+                elif param.default is inspect.Parameter.empty:
+                    required.append(param_name)
+            elif param.default is inspect.Parameter.empty:
+                required.append(param_name)
+            properties[param_name] = param_schema
+
+    parameters = {
+        "type": "object",
+        "properties": properties,
+        "required": required,
+    }
+
+    return Tool(
+        name=name,
+        description=description,
+        parameters=parameters,
+        handler=fn,
+    )
+
+
+@tool
 def fetch_webpage(url: str) -> str:
-    """Fetch a webpage and return its text content."""
+    """Fetch a webpage and return its text content.
+
+    url (required): The URL to fetch
+    """
     try:
         jina_ai_url = "https://r.jina.ai/"
         response = requests.get(jina_ai_url + url)
@@ -38,18 +146,4 @@ def fetch_webpage(url: str) -> str:
         md = MarkItDown()
         return md.convert(url).text_content
 
-FETCH_WEBPAGE_TOOL = Tool(
-    name="fetch_webpage",
-    description="Fetch the content of a webpage and return its text",
-    parameters={
-        "type": "object",
-        "properties": {
-            "url": {
-                "type": "string",
-                "description": "The URL to fetch",
-            }
-        },
-        "required": ["url"],
-    },
-    handler=fetch_webpage,
-)
+FETCH_WEBPAGE_TOOL = fetch_webpage  # @tool already makes it a Tool instance

app.py

@@ -31,6 +31,7 @@ agent = Agent(
     system_prompt=_SYSTEM_PROMPT,
 )
 agent.register_tool(FETCH_WEBPAGE_TOOL)
+agent.register_all_mcp()
 agent.register_final_message_tool()
 
 # Load JS from external files