preliminary mcp

agent/MCP_INTEGRATION.md

@@ -0,0 +1,205 @@
+# MCP Integration for HF Agent
+
+This agent now supports the Model Context Protocol (MCP), allowing it to connect to and use tools from MCP servers.
+
+## Overview
+
+The MCP integration allows the agent to:
+- Connect to multiple MCP servers simultaneously
+- Automatically discover and use tools from connected servers
+- Execute tool calls through the MCP protocol
+- Seamlessly integrate MCP tools with the agent's existing tool system
+
+## Architecture
+
+The integration consists of several components:
+
+1. **MCPClient** (`agent/core/mcp_client.py`): Manages connections to MCP servers
+2. **ToolExecutor** (`agent/core/executor.py`): Executes both MCP and local tools
+3. **Config** (`agent/config.py`): Stores MCP server configurations
+4. **Session** (`agent/core/session.py`): Initializes MCP connections and manages lifecycle
+
+## Configuration
+
+To use MCP servers with your agent, add them to your configuration file:
+
+```json
+{
+  "model_name": "anthropic/claude-sonnet-4-5-20250929",
+  "tools": [],
+  "system_prompt_path": "",
+  "mcp_servers": [
+    {
+      "name": "weather",
+      "command": "python",
+      "args": ["path/to/weather_server.py"],
+      "env": null
+    },
+    {
+      "name": "filesystem",
+      "command": "node",
+      "args": ["path/to/filesystem_server.js"],
+      "env": {
+        "ALLOWED_PATHS": "/home/user/documents"
+      }
+    }
+  ]
+}
+```
+
+### Configuration Fields
+
+- `name`: Unique identifier for the MCP server
+- `command`: Command to execute the server (`python`, `node`, etc.)
+- `args`: Arguments to pass to the command (path to server script)
+- `env`: (Optional) Environment variables for the server process
+
+## Usage
+
+### Basic Usage
+
+```python
+import asyncio
+from agent.config import Config, load_config
+from agent.core.agent_loop import submission_loop
+
+async def main():
+    # Load config with MCP servers
+    config = load_config("config.json")
+
+    # Create queues
+    submission_queue = asyncio.Queue()
+    event_queue = asyncio.Queue()
+
+    # Start agent loop (MCP connections initialized automatically)
+    await submission_loop(submission_queue, event_queue, config)
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+### Programmatic Configuration
+
+```python
+from agent.config import Config, MCPServerConfig
+
+config = Config(
+    model_name="anthropic/claude-sonnet-4-5-20250929",
+    tools=[],
+    system_prompt_path="",
+    mcp_servers=[
+        MCPServerConfig(
+            name="weather",
+            command="python",
+            args=["weather_server.py"],
+            env=None
+        )
+    ]
+)
+```
+
+## How It Works
+
+1. **Initialization**: When the agent loop starts, it calls `session.initialize_mcp()`
+2. **Connection**: The session connects to all configured MCP servers
+3. **Tool Discovery**: Tools from all servers are discovered and added to the agent's tool list
+4. **Tool Naming**: MCP tools are prefixed with their server name (e.g., `weather__get_forecast`)
+5. **Execution**: When the LLM calls a tool, the ToolExecutor routes it to the appropriate MCP server
+6. **Cleanup**: When the agent shuts down, all MCP connections are cleaned up properly
+
+## Tool Naming Convention
+
+MCP tools are automatically prefixed with their server name to avoid conflicts:
+
+- Original tool: `get_forecast`
+- MCP tool name: `weather__get_forecast`
+
+This ensures that tools from different servers don't conflict, even if they have the same name.
+
+## Example: Creating a Simple MCP Server
+
+Here's a minimal example of an MCP server (save as `calculator_server.py`):
+
+```python
+import asyncio
+from mcp.server import Server, stdio_server
+from mcp.types import Tool, TextContent
+
+app = Server("calculator")
+
+@app.list_tools()
+async def list_tools() -> list[Tool]:
+    return [
+        Tool(
+            name="add",
+            description="Add two numbers",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "a": {"type": "number"},
+                    "b": {"type": "number"}
+                },
+                "required": ["a", "b"]
+            }
+        )
+    ]
+
+@app.call_tool()
+async def call_tool(name: str, arguments: dict) -> list[TextContent]:
+    if name == "add":
+        result = arguments["a"] + arguments["b"]
+        return [TextContent(type="text", text=str(result))]
+
+    raise ValueError(f"Unknown tool: {name}")
+
+async def main():
+    async with stdio_server() as (read_stream, write_stream):
+        await app.run(read_stream, write_stream, app.create_initialization_options())
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+## Troubleshooting
+
+### Server Connection Issues
+
+If you see errors connecting to an MCP server:
+
+1. Check that the server script path is correct
+2. Ensure the command (`python`, `node`) is in your PATH
+3. Verify the server script is executable
+4. Check server logs for initialization errors
+
+### Tool Not Found
+
+If the agent can't find an MCP tool:
+
+1. Verify the server is connected (check startup logs)
+2. Check tool naming (should be `servername__toolname`)
+3. Ensure the server properly implements `list_tools()`
+
+### Performance Considerations
+
+- MCP server initialization happens once at startup
+- Tool calls are asynchronous and don't block the agent
+- Multiple servers can be used simultaneously
+- Consider using local tools for high-frequency operations
+
+## Best Practices
+
+1. **Unique Server Names**: Give each MCP server a unique, descriptive name
+2. **Error Handling**: MCP connection failures are logged but don't crash the agent
+3. **Resource Cleanup**: Always let the agent shut down gracefully to cleanup connections
+4. **Testing**: Test MCP servers independently before integrating them
+5. **Security**: Be cautious with file system and network access in MCP servers
+
+## Future Enhancements
+
+Potential improvements to consider:
+
+- Dynamic server addition/removal during runtime
+- Server health monitoring and auto-reconnection
+- Tool caching and performance optimization
+- Support for MCP resources and prompts
+- Rate limiting and timeout configuration

agent/config.py

@@ -2,12 +2,22 @@ from litellm import Tool
 from pydantic import BaseModel
 
 
+class MCPServerConfig(BaseModel):
+    """Configuration for an MCP server"""
+
+    name: str
+    command: str
+    args: list[str]
+    env: dict[str, str] | None = None
+
+
 class Config(BaseModel):
     """Configuration manager"""
 
     model_name: str
     tools: list[Tool]
     system_prompt_path: str
+    mcp_servers: list[MCPServerConfig] = []
 
 
 def load_config(config_path: str = "config.json") -> Config:

agent/config_mcp_example.json

@@ -0,0 +1,13 @@
+{
+  "model_name": "anthropic/claude-sonnet-4-5-20250929",
+  "tools": [],
+  "system_prompt_path": "",
+  "mcp_servers": [
+    {
+      "name": "weather",
+      "command": "python",
+      "args": ["path/to/your/weather_server.py"],
+      "env": null
+    }
+  ]
+}

agent/core/__init__.py

@@ -4,5 +4,6 @@ Contains the main agent logic, decision-making, and orchestration
 """
 
 from agent.core.executor import ToolExecutor
+from agent.core.mcp_client import MCPClient, MCPServerConfig
 
-__all__ = ["ToolExecutor"]
+__all__ = ["ToolExecutor", "MCPClient", "MCPServerConfig"]

agent/core/agent_loop.py

@@ -141,6 +141,40 @@ class Handlers:
         return True
 
 
+async def process_submission(session: Session, submission) -> bool:
+    """
+    Process a single submission and return whether to continue running.
+
+    Returns:
+        bool: True to continue, False to shutdown
+    """
+    op = submission.operation
+    print(f"📨 Received: {op.op_type.value}")
+
+    if op.op_type == OpType.USER_INPUT:
+        text = op.data.get("text", "") if op.data else ""
+        await Handlers.run_agent(session, text, max_iterations=10)
+        return True
+
+    if op.op_type == OpType.INTERRUPT:
+        await Handlers.interrupt(session)
+        return True
+
+    if op.op_type == OpType.COMPACT:
+        await Handlers.compact(session)
+        return True
+
+    if op.op_type == OpType.UNDO:
+        await Handlers.undo(session)
+        return True
+
+    if op.op_type == OpType.SHUTDOWN:
+        return not await Handlers.shutdown(session)
+
+    print(f"⚠️  Unknown operation: {op.op_type}")
+    return True
+
+
 async def submission_loop(
     submission_queue: asyncio.Queue,
     event_queue: asyncio.Queue,
@@ -151,45 +185,31 @@ async def submission_loop(
     This is the core of the agent (like submission_loop in codex.rs:1259-1340)
     """
     session = Session(event_queue, config=config)
-
     print("🤖 Agent loop started")
 
-    # Main processing loop
-    while session.is_running:
+    # Initialize MCP connections
+    if session.config.mcp_servers:
         try:
-            # Wait for next submission (like rx_sub.recv() in codex.rs:1262)
-            submission = await submission_queue.get()
-
-            print(f"📨 Received: {submission.operation.op_type.value}")
-
-            # Dispatch to handler based on operation type
-            op = submission.operation
-
-            if op.op_type == OpType.USER_INPUT:
-                text = op.data.get("text", "") if op.data else ""
-                await Handlers.run_agent(session, text, max_iterations=10)
-
-            elif op.op_type == OpType.INTERRUPT:
-                # im not currently sure what this does lol
-                await Handlers.interrupt(session)
-
-            elif op.op_type == OpType.COMPACT:
-                await Handlers.compact(session)
+            print(f"Initializing MCP connections for {session.config.mcp_servers}")
+            await session.initialize_mcp()
+        except Exception as e:
+            print(f"⚠️  Error initializing MCP: {e}")
 
-            elif op.op_type == OpType.UNDO:
-                await Handlers.undo(session)
+    try:
+        # Main processing loop
+        while session.is_running:
+            submission = await submission_queue.get()
 
-            elif op.op_type == OpType.SHUTDOWN:
-                if await Handlers.shutdown(session):
+            try:
+                should_continue = await process_submission(session, submission)
+                if not should_continue:
                     break
-
-            else:
-                print(f"⚠️  Unknown operation: {op.op_type}")
-
-        except asyncio.CancelledError:
-            break
-        except Exception as e:
-            print(f"❌ Error in agent loop: {e}")
-            await session.send_event(Event(event_type="error", data={"error": str(e)}))
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                print(f"❌ Error in agent loop: {e}")
+                await session.send_event(Event(event_type="error", data={"error": str(e)}))
+    finally:
+        await session.cleanup()
 
     print("🛑 Agent loop exited")

agent/core/executor.py

@@ -2,6 +2,7 @@
 Task execution engine
 """
 
+import json
 from typing import Any, List
 
 from litellm import ChatCompletionMessageToolCall
@@ -18,10 +19,30 @@ class ToolResult(BaseModel):
 class ToolExecutor:
     """Executes planned tasks using available tools"""
 
-    def __init__(self, tools: List[Any] = None):
+    def __init__(self, tools: List[Any] = None, mcp_client=None):
         self.tools = tools or []
+        self.mcp_client = mcp_client
 
     async def execute_tool(self, tool_call: ToolCall) -> ToolResult:
         """Execute a single step in the plan"""
-        # TODO: Implement step execution
-        return ToolResult(output="", success=True)
+        tool_name = tool_call.function.name
+
+        # Parse arguments
+        try:
+            if isinstance(tool_call.function.arguments, str):
+                tool_args = json.loads(tool_call.function.arguments)
+            else:
+                tool_args = tool_call.function.arguments
+        except json.JSONDecodeError as e:
+            return ToolResult(
+                output=f"Error parsing tool arguments: {str(e)}", success=False
+            )
+
+        # Check if this is an MCP tool (prefixed with server name)
+        if self.mcp_client and "__" in tool_name:
+            success, result = await self.mcp_client.call_tool(tool_name, tool_args)
+            return ToolResult(output=result, success=success)
+
+        # If not an MCP tool, try local tools
+        # TODO: Implement local tool execution
+        return ToolResult(output=f"Tool {tool_name} not found", success=False)

agent/core/mcp_client.py

@@ -0,0 +1,149 @@
+"""
+MCP (Model Context Protocol) client integration for the agent
+"""
+
+from contextlib import AsyncExitStack
+from typing import Optional
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+
+class MCPServerConfig:
+    """Configuration for an MCP server"""
+
+    def __init__(
+        self,
+        name: str,
+        command: str,
+        args: list[str],
+        env: Optional[dict[str, str]] = None,
+    ):
+        self.name = name
+        self.command = command
+        self.args = args
+        self.env = env
+
+
+class MCPClient:
+    """
+    Manages connections to MCP servers and provides tool access
+    """
+
+    def __init__(self):
+        self.sessions: dict[str, ClientSession] = {}
+        self.exit_stack = AsyncExitStack()
+        self._tools_cache: Optional[list[dict]] = None
+
+    async def connect_to_server(self, server_config: MCPServerConfig) -> None:
+        """
+        Connect to an MCP server
+
+        Args:
+            server_config: Configuration for the MCP server
+        """
+        server_params = StdioServerParameters(
+            command=server_config.command,
+            args=server_config.args,
+            env=server_config.env,
+        )
+
+        stdio_transport = await self.exit_stack.enter_async_context(
+            stdio_client(server_params)
+        )
+        stdio, write = stdio_transport
+        session = await self.exit_stack.enter_async_context(ClientSession(stdio, write))
+
+        await session.initialize()
+
+        # Store the session
+        self.sessions[server_config.name] = session
+
+        # Invalidate tools cache
+        self._tools_cache = None
+
+        print(f"✅ Connected to MCP server: {server_config.name}")
+
+    async def list_tools(self) -> list[dict]:
+        """
+        Get all available tools from all connected servers
+
+        Returns:
+            List of tool definitions compatible with LiteLLM format
+        """
+        if self._tools_cache is not None:
+            return self._tools_cache
+
+        all_tools = []
+
+        for server_name, session in self.sessions.items():
+            try:
+                response = await session.list_tools()
+                for tool in response.tools:
+                    # Convert MCP tool format to LiteLLM tool format
+                    tool_def = {
+                        "type": "function",
+                        "function": {
+                            "name": f"{server_name}__{tool.name}",  # Prefix with server name
+                            "description": tool.description or "",
+                            "parameters": tool.inputSchema,
+                        },
+                    }
+                    all_tools.append(tool_def)
+            except Exception as e:
+                print(f"⚠️  Error listing tools from {server_name}: {e}")
+
+        self._tools_cache = all_tools
+        return all_tools
+
+    async def call_tool(self, tool_name: str, tool_args: dict) -> tuple[bool, str]:
+        """
+        Call a tool on the appropriate MCP server
+
+        Args:
+            tool_name: Name of the tool (format: "server_name__tool_name")
+            tool_args: Arguments to pass to the tool
+
+        Returns:
+            Tuple of (success, result_content)
+        """
+        # Parse server name from tool name
+        if "__" not in tool_name:
+            return False, f"Invalid tool name format: {tool_name}"
+
+        server_name, actual_tool_name = tool_name.split("__", 1)
+
+        if server_name not in self.sessions:
+            return False, f"Server not found: {server_name}"
+
+        session = self.sessions[server_name]
+
+        try:
+            result = await session.call_tool(actual_tool_name, tool_args)
+
+            # Extract content from result
+            if hasattr(result, "content"):
+                if isinstance(result.content, list):
+                    # Handle list of content items
+                    content_parts = []
+                    for item in result.content:
+                        if hasattr(item, "text"):
+                            content_parts.append(item.text)
+                        else:
+                            content_parts.append(str(item))
+                    content = "\n".join(content_parts)
+                else:
+                    content = str(result.content)
+            else:
+                content = str(result)
+
+            return True, content
+
+        except Exception as e:
+            return False, f"Error calling tool {tool_name}: {str(e)}"
+
+    async def cleanup(self) -> None:
+        """Clean up all MCP connections"""
+        await self.exit_stack.aclose()
+        self.sessions.clear()
+        self._tools_cache = None

agent/core/session.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel
 from agent.config import Config
 from agent.context_manager.manager import ContextManager
 from agent.core import ToolExecutor
+from agent.core.mcp_client import MCPClient, MCPServerConfig as MCPServerConfigClass
 
 
 class OpType(Enum):
@@ -41,15 +42,48 @@ class Session:
 
     def __init__(self, event_queue: asyncio.Queue, config: Config | None = None):
         self.context_manager = ContextManager()
-        self.tool_executor = ToolExecutor()
         self.event_queue = event_queue
         self.config = config or Config(
             model_name="anthropic/claude-sonnet-4-5-20250929",
             tools=[],
             system_prompt_path="",
         )
+
+        # Initialize MCP client
+        self.mcp_client = MCPClient()
+        self.tool_executor = ToolExecutor(mcp_client=self.mcp_client)
+
         self.is_running = True
         self.current_task: asyncio.Task | None = None
+        self._mcp_initialized = False
+
+    async def initialize_mcp(self) -> None:
+        """Initialize MCP server connections"""
+        if self._mcp_initialized:
+            return
+
+        for server_config in self.config.mcp_servers:
+            try:
+                mcp_server_config = MCPServerConfigClass(
+                    name=server_config.name,
+                    command=server_config.command,
+                    args=server_config.args,
+                    env=server_config.env,
+                )
+                await self.mcp_client.connect_to_server(mcp_server_config)
+            except Exception as e:
+                print(f"⚠️  Failed to connect to MCP server {server_config.name}: {e}")
+
+        # Get MCP tools and merge with config tools
+        try:
+            mcp_tools = await self.mcp_client.list_tools()
+            # Merge with existing tools
+            self.config.tools = list(self.config.tools) + mcp_tools
+            print(f"📦 Loaded {len(mcp_tools)} tools from MCP servers")
+        except Exception as e:
+            print(f"⚠️  Error loading MCP tools: {e}")
+
+        self._mcp_initialized = True
 
     async def send_event(self, event: Event) -> None:
         """Send event back to client"""
@@ -59,3 +93,7 @@ class Session:
         """Interrupt current running task"""
         if self.current_task and not self.current_task.done():
             self.current_task.cancel()
+
+    async def cleanup(self) -> None:
+        """Cleanup session resources"""
+        await self.mcp_client.cleanup()