Merge remote-tracking branch 'upstream/main'

nanobot/agent/loop.py

@@ -1,7 +1,9 @@
 """Agent loop: the core processing engine."""
 
 import asyncio
+from contextlib import AsyncExitStack
 import json
+import json_repair
 from pathlib import Path
 from typing import Any
 
@@ -50,6 +52,7 @@ class AgentLoop:
         cron_service: "CronService | None" = None,
         restrict_to_workspace: bool = False,
         session_manager: SessionManager | None = None,
+        mcp_servers: dict | None = None,
     ):
         from nanobot.config.schema import ExecToolConfig
         from nanobot.cron.service import CronService
@@ -82,6 +85,9 @@ class AgentLoop:
         )
         
         self._running = False
+        self._mcp_servers = mcp_servers or {}
+        self._mcp_stack: AsyncExitStack | None = None
+        self._mcp_connected = False
         self._register_default_tools()
     
     def _register_default_tools(self) -> None:
@@ -116,6 +122,16 @@ class AgentLoop:
         if self.cron_service:
             self.tools.register(CronTool(self.cron_service))
     
+    async def _connect_mcp(self) -> None:
+        """Connect to configured MCP servers (one-time, lazy)."""
+        if self._mcp_connected or not self._mcp_servers:
+            return
+        self._mcp_connected = True
+        from nanobot.agent.tools.mcp import connect_mcp_servers
+        self._mcp_stack = AsyncExitStack()
+        await self._mcp_stack.__aenter__()
+        await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack)
+
     def _set_tool_context(self, channel: str, chat_id: str) -> None:
         """Update context for all tools that need routing info."""
         if message_tool := self.tools.get("message"):
@@ -191,6 +207,7 @@ class AgentLoop:
     async def run(self) -> None:
         """Run the agent loop, processing messages from the bus."""
         self._running = True
+        await self._connect_mcp()
         logger.info("Agent loop started")
 
         while self._running:
@@ -213,6 +230,15 @@ class AgentLoop:
             except asyncio.TimeoutError:
                 continue
     
+    async def close_mcp(self) -> None:
+        """Close MCP connections."""
+        if self._mcp_stack:
+            try:
+                await self._mcp_stack.aclose()
+            except (RuntimeError, BaseExceptionGroup):
+                pass  # MCP SDK cancel scope cleanup is noisy but harmless
+            self._mcp_stack = None
+
     def stop(self) -> None:
         """Stop the agent loop."""
         self._running = False
@@ -403,9 +429,15 @@ Respond with ONLY valid JSON, no markdown fences."""
                 model=self.model,
             )
             text = (response.content or "").strip()
+            if not text:
+                logger.warning("Memory consolidation: LLM returned empty response, skipping")
+                return
             if text.startswith("```"):
                 text = text.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
-            result = json.loads(text)
+            result = json_repair.loads(text)
+            if not isinstance(result, dict):
+                logger.warning(f"Memory consolidation: unexpected response type, skipping. Response: {text[:200]}")
+                return
 
             if entry := result.get("history_entry"):
                 # Defensive: ensure entry is a string (LLM may return dict)
@@ -446,6 +478,7 @@ Respond with ONLY valid JSON, no markdown fences."""
         Returns:
             The agent's response.
         """
+        await self._connect_mcp()
         msg = InboundMessage(
             channel=channel,
             sender_id="user",

nanobot/agent/tools/mcp.py

@@ -0,0 +1,80 @@
+"""MCP client: connects to MCP servers and wraps their tools as native nanobot tools."""
+
+from contextlib import AsyncExitStack
+from typing import Any
+
+from loguru import logger
+
+from nanobot.agent.tools.base import Tool
+from nanobot.agent.tools.registry import ToolRegistry
+
+
+class MCPToolWrapper(Tool):
+    """Wraps a single MCP server tool as a nanobot Tool."""
+
+    def __init__(self, session, server_name: str, tool_def):
+        self._session = session
+        self._original_name = tool_def.name
+        self._name = f"mcp_{server_name}_{tool_def.name}"
+        self._description = tool_def.description or tool_def.name
+        self._parameters = tool_def.inputSchema or {"type": "object", "properties": {}}
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def description(self) -> str:
+        return self._description
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return self._parameters
+
+    async def execute(self, **kwargs: Any) -> str:
+        from mcp import types
+        result = await self._session.call_tool(self._original_name, arguments=kwargs)
+        parts = []
+        for block in result.content:
+            if isinstance(block, types.TextContent):
+                parts.append(block.text)
+            else:
+                parts.append(str(block))
+        return "\n".join(parts) or "(no output)"
+
+
+async def connect_mcp_servers(
+    mcp_servers: dict, registry: ToolRegistry, stack: AsyncExitStack
+) -> None:
+    """Connect to configured MCP servers and register their tools."""
+    from mcp import ClientSession, StdioServerParameters
+    from mcp.client.stdio import stdio_client
+
+    for name, cfg in mcp_servers.items():
+        try:
+            if cfg.command:
+                params = StdioServerParameters(
+                    command=cfg.command, args=cfg.args, env=cfg.env or None
+                )
+                read, write = await stack.enter_async_context(stdio_client(params))
+            elif cfg.url:
+                from mcp.client.streamable_http import streamable_http_client
+                read, write, _ = await stack.enter_async_context(
+                    streamable_http_client(cfg.url)
+                )
+            else:
+                logger.warning(f"MCP server '{name}': no command or url configured, skipping")
+                continue
+
+            session = await stack.enter_async_context(ClientSession(read, write))
+            await session.initialize()
+
+            tools = await session.list_tools()
+            for tool_def in tools.tools:
+                wrapper = MCPToolWrapper(session, name, tool_def)
+                registry.register(wrapper)
+                logger.debug(f"MCP: registered tool '{wrapper.name}' from server '{name}'")
+
+            logger.info(f"MCP server '{name}': connected, {len(tools.tools)} tools registered")
+        except Exception as e:
+            logger.error(f"MCP server '{name}': failed to connect: {e}")