local MCP support

README.md

@@ -23,7 +23,7 @@ Features:
    * example: download an ICS calendar file the model has created for you
 * streaming chat
 * image generation (via DALL-E 3)
-* remote MCP server support via configurable registry
+* MCP server support (both remote and local) via configurable registry
 * optional UnrestrictedPython execution when `CODE_EXEC_UNRESTRICTED_PYTHON=1`
 
 The MCP registry is looked up in the following order:
@@ -34,5 +34,19 @@ The MCP registry is looked up in the following order:
 See `mcp_registry.sample.json` for an example configuration.
 Headers and query parameters may reference environment variables using the `env:` prefix.
 Use `"allowed_tools": ["*"]` to permit all tools from a server.
+
+For local MCP servers, use the `command` and `args` fields to specify how to launch the server. Environment variables can be passed via the `env` field. For example:
+```json
+{
+  "name": "exa_local",
+  "command": "npx",
+  "args": ["-y", "exa-mcp-server"],
+  "env": {
+    "EXA_API_KEY": "env:EXA_API_KEY"
+  },
+  "allowed_tools": ["*"]
+}
+```
+
 When an MCP tool requires approval, the assistant will notify you in chat.
 Reply with `y` to approve or `n` to deny the request, optionally adding a comment after the `y` or `n`.

app.py

@@ -6,9 +6,10 @@ from openai import OpenAI
 import json
 from PIL import Image
 import io
+import asyncio
 from settings_mgr import generate_download_settings_js, generate_upload_settings_js
 from chat_export import import_history, get_export_js
-from mcp_registry import load_registry, to_openai_tool
+from mcp_registry import load_registry, get_tools_for_server, call_local_mcp_tool, function_to_mcp_map, shutdown_local_mcp_clients
 from gradio.components.base import Component
 from types import SimpleNamespace
 
@@ -185,7 +186,7 @@ def process_values_js():
     }
     """
 
-def bot(message, history, oai_key, system_prompt, temperature, max_tokens, model, python_use, web_search, *mcp_selected):
+async def bot(message, history, oai_key, system_prompt, temperature, max_tokens, model, python_use, web_search, *mcp_selected):
     global pending_mcp_request
     try:
         client = OpenAI(
@@ -296,9 +297,10 @@ def bot(message, history, oai_key, system_prompt, temperature, max_tokens, model
                     "type": "web_search",
                     "search_context_size": "high"
                     })
+            # Add selected MCP servers to tools
             for sel, entry in zip(mcp_selected, mcp_servers):
                 if sel:
-                    tools.append(to_openai_tool(entry))
+                    tools.extend(await get_tools_for_server(entry))
             if not tools:
                 tools = None
 
@@ -480,7 +482,113 @@ def bot(message, history, oai_key, system_prompt, temperature, max_tokens, model
                                                 yield assistant_msgs
 
                             elif output.type == "function_call":
-                                if output.name == "eval_python":
+                                # Check if this is a local MCP tool call
+                                function_name = output.name
+                                if function_name in function_to_mcp_map:
+                                    try:
+                                        mcp_info = function_to_mcp_map[function_name]
+                                        server_name = mcp_info["server_name"]
+                                        tool_name = mcp_info["tool_name"]
+                                        
+                                        # Find the server entry
+                                        server_entry = None
+                                        for entry in mcp_servers:
+                                            if entry["name"] == server_name:
+                                                server_entry = entry
+                                                break
+                                        
+                                        if server_entry:
+                                            history_openai_format.append({
+                                                "type": "function_call",
+                                                "name": function_name,
+                                                "arguments": output.arguments,
+                                                "call_id": output.call_id
+                                            })
+                                            
+                                            # Parse arguments
+                                            arguments = json.loads(output.arguments)
+                                            call_id = output.call_id
+                                            
+                                            # Show the function call to the user
+                                            parent_msg = gr.ChatMessage(
+                                                role="assistant",
+                                                content="",
+                                                metadata={"title": f"MCP: {server_name} - {tool_name}", "id": call_id, "status": "pending"},
+                                            )
+                                            assistant_msgs.append(parent_msg)
+                                            assistant_msgs.append(
+                                                gr.ChatMessage(
+                                                    role="assistant",
+                                                    content=f"``` arguments\n{output.arguments}\n```",
+                                                    metadata={"title": "request", "parent_id": call_id},
+                                                )
+                                            )
+                                            yield assistant_msgs
+                                            
+                                            # Call the MCP tool (async)
+                                            try:
+                                                tool_result = await call_local_mcp_tool(server_entry, tool_name, arguments)
+                                                # Extract text from result
+                                                if isinstance(tool_result, list) and tool_result and hasattr(tool_result[0], 'text'):
+                                                    result_text = "\n".join([item.text for item in tool_result])
+                                                elif hasattr(tool_result, 'text'):
+                                                    result_text = tool_result.text
+                                                else:
+                                                    result_text = str(tool_result)
+                                                # Show result to the user
+                                                assistant_msgs.append(
+                                                    gr.ChatMessage(
+                                                        role="assistant",
+                                                        content=f"``` result\n{result_text}\n```",
+                                                        metadata={"title": "response", "parent_id": call_id, "status": "done"},
+                                                    )
+                                                )
+                                                parent_msg.metadata["status"] = "done"
+                                                yield assistant_msgs
+                                                # Add result to history
+                                                history_openai_format.append(
+                                                    {
+                                                        "type": "function_call_output",
+                                                        "call_id": output.call_id,
+                                                        "output": result_text,
+                                                    }
+                                                )
+                                            except Exception as e:
+                                                error_message = str(e)
+                                                history_openai_format.append({
+                                                    "type": "function_call_output",
+                                                    "call_id": output.call_id,
+                                                    "output": json.dumps({"error": error_message})
+                                                })
+                                                assistant_msgs.append(
+                                                    gr.ChatMessage(
+                                                        role="assistant",
+                                                        content=f"``` error\n{error_message}\n```",
+                                                        metadata={"title": "response", "parent_id": call_id, "status": "done"},
+                                                    )
+                                                )
+                                                parent_msg.metadata["status"] = "done"
+                                                yield assistant_msgs
+                                            
+                                            # Need to continue the loop to process the function output
+                                            loop_tool_calling = True
+                                        else:
+                                            # Server entry not found
+                                            error_message = f"Server {server_name} not found"
+                                            history_openai_format.append({
+                                                "type": "function_call_output",
+                                                "call_id": output.call_id,
+                                                "output": json.dumps({"error": error_message})
+                                            })
+                                    except Exception as e:
+                                        # Some error occurred during processing
+                                        error_message = f"Error processing local MCP tool call: {str(e)}"
+                                        history_openai_format.append({
+                                            "type": "function_call_output",
+                                            "call_id": output.call_id,
+                                            "output": json.dumps({"error": error_message})
+                                        })
+                                elif output.name == "eval_python":
                                     try:
                                         history_openai_format.append({
                                             "type": "function_call",

mcp_registry.py

@@ -1,6 +1,21 @@
 import os
 import json
+import asyncio
+import logging
 from urllib.parse import urlencode
+from typing import Dict, List, Any, Optional, Union
+
+try:
+    from fastmcp import Client
+    from fastmcp.client.transports import StdioTransport, PythonStdioTransport
+except ImportError:
+    logging.warning("FastMCP library not installed. Local MCP servers will not be available.")
+    Client = None
+    StdioTransport = None
+
+# Global dictionary to store local MCP clients
+local_mcp_clients = {}
+local_mcp_tools_cache = {}
 
 
 _SEARCH_PATHS = [
@@ -9,18 +24,17 @@ _SEARCH_PATHS = [
     os.path.expanduser("~/.oai_chat/mcp_registry.json"),
 ]
 
+async def log(msg):
+    print("[MCP SERVER]", msg.data, flush=True)
+
 
 def _merge_defaults(reg: dict) -> list[dict]:
     defaults = reg.get("defaults", {})
     servers = []
     for entry in reg.get("servers", []):
-        if entry.get("url"):
-            merged = dict(defaults)
-            merged.update(entry)
-            servers.append(merged)
-        else:
-            # Local MCPs not yet supported
-            pass
+        merged = dict(defaults)
+        merged.update(entry)
+        servers.append(merged)
     return servers
 
 
@@ -45,20 +59,123 @@ def env_subst(values: dict, kind: str) -> dict:
     return out
 
 
-def to_openai_tool(entry: dict) -> dict:
-    server_url = entry["url"]
-    if "query_params" in entry:
-        qp = urlencode(env_subst(entry["query_params"], "query parameter"))
-        if "?" in server_url:
-            server_url += "&" + qp
-        else:
-            server_url += "?" + qp
-    tool = {
-        "type": "mcp",
-        "server_label": entry.get("server_label", entry["name"]),
-        "server_url": server_url,
-        "headers": env_subst(entry.get("headers", {}), "header"),
-    }
+def is_local_mcp(entry: dict) -> bool:
+    """Check if an MCP entry is a local MCP server"""
+    return "command" in entry and "args" in entry
+
+async def start_local_mcp_client(entry: dict) -> Optional[Client]:
+    """Start a local MCP client for a given entry"""
+    if Client is None or StdioTransport is None:
+        logging.error("FastMCP library not installed. Cannot start local MCP client.")
+        return None
+        
+    try:
+        name = entry["name"]
+        command = entry["command"]
+        args = entry["args"]
+        
+        # Prepare environment variables
+        env_vars = {}
+        if "env" in entry:
+            env_vars = env_subst(entry["env"], "environment variable")
+        
+        # Create transport with environment variables
+        transport = StdioTransport(
+            command=command,
+            args=args,
+            env=env_vars if env_vars else None
+        )
+        
+        # Create client with the transport
+        client = Client(transport, log_handler=log)
+        
+        # Store the client in the global dictionary
+        local_mcp_clients[name] = client
+        
+        return client
+    except Exception as e:
+        logging.error(f"Failed to start local MCP client: {str(e)}")
+        return None
+
+async def get_local_mcp_tools(entry: dict) -> List[Dict[str, Any]]:
+    """Get available tools from a local MCP server"""
+    name = entry["name"]
+    
+    # Check if we have cached tools for this server
+    if name in local_mcp_tools_cache:
+        return local_mcp_tools_cache[name]
+        
+    # Check if client exists or create a new one
+    client = local_mcp_clients.get(name)
+    if client is None:
+        client = await start_local_mcp_client(entry)
+        if client is None:
+            return []
+    
+    try:
+        # Use client in async context manager
+        async with client:
+            # List available tools
+            tools = await client.list_tools()
+            # Cache the tools
+            local_mcp_tools_cache[name] = tools
+            return tools
+    except Exception as e:
+        logging.error(f"Failed to list tools from local MCP server: {str(e)}")
+        return []
+
+async def call_local_mcp_tool(entry: dict, tool_name: str, arguments: Dict[str, Any]) -> Any:
+    """Call a tool on a local MCP server"""
+    name = entry["name"]
+    
+    # Check if client exists or create a new one
+    client = local_mcp_clients.get(name)
+    if client is None:
+        client = await start_local_mcp_client(entry)
+        if client is None:
+            return {"error": "Failed to connect to local MCP server"}
+    
+    try:
+        # Use client in async context manager
+        async with client:
+            if not client.is_connected():
+                logging.warning("MCP server not connected")
+
+            # Call the tool
+            result = await client.call_tool(tool_name, arguments)
+            return result
+    except Exception as e:
+        logging.error(f"Failed to call tool on local MCP server: {str(e)}")
+        return {"error": str(e)}
+
+async def shutdown_local_mcp_clients():
+    """Shutdown all local MCP clients"""
+    for name, client in local_mcp_clients.items():
+        try:
+            await client.close()
+        except Exception as e:
+            logging.error(f"Failed to close local MCP client {name}: {str(e)}")
+    local_mcp_clients.clear()
+    local_mcp_tools_cache.clear()
+
+def to_openai_tool(entry: dict) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
+    """Convert an MCP entry to an OpenAI tool definition(s)"""
+    # For remote MCP servers, use the standard "mcp" type
+    if "url" in entry:
+        tool = {
+            "type": "mcp",
+            "server_label": entry.get("server_label", entry["name"]),
+        }
+        
+        server_url = entry["url"]
+        if "query_params" in entry:
+            qp = urlencode(env_subst(entry["query_params"], "query parameter"))
+            if "?" in server_url:
+                server_url += "&" + qp
+            else:
+                server_url += "?" + qp
+        tool["server_url"] = server_url
+        tool["headers"] = env_subst(entry.get("headers", {}), "header")
     if "allowed_tools" in entry:
         allowed = entry["allowed_tools"]
         if not (len(allowed) == 1 and allowed[0] == "*"):
@@ -66,3 +183,62 @@ def to_openai_tool(entry: dict) -> dict:
     if "require_approval" in entry:
         tool["require_approval"] = entry["require_approval"]
     return tool
+
+# Global mapping to track function names back to their MCP servers and tool names
+function_to_mcp_map = {}
+
+# Cache for local MCP tools
+local_mcp_tool_cache = {}
+
+# Helper function to create a function tool definition for a local MCP tool
+def create_function_tool_for_local_mcp_tool(server_name: str, tool_name: str, tool_obj) -> Dict[str, Any]:
+    """Create an OpenAI function tool definition for a local MCP tool"""
+    function_name = f"{server_name}_{tool_name}"
+    
+    # Save the mapping for later lookup during function call
+    function_to_mcp_map[function_name] = {
+        "server_name": server_name,
+        "tool_name": tool_name
+    }
+    
+    # Handle FastMCP Tool object format (based on observed structure)
+    description = getattr(tool_obj, 'description', f"Tool {tool_name} from {server_name} MCP server")
+    parameters = getattr(tool_obj, 'inputSchema', {"type": "object", "properties": {}})
+    
+    return {
+        "type": "function",
+        "name": function_name,
+        "description": description,
+        "parameters": parameters
+    }
+
+async def get_tools_for_server(entry: dict) -> List[Dict[str, Any]]:
+    """Get all tools for a given server entry (local or remote)
+    For remote servers, it returns a single MCP tool.
+    For local servers, it returns multiple function tools (one for each MCP tool).
+    """
+    if is_local_mcp(entry):
+        server_name = entry["name"]
+        # Try to get tools from cache first
+        if server_name in local_mcp_tool_cache:
+            mcp_tools = local_mcp_tool_cache[server_name]
+        else:
+            try:
+                mcp_tools = await get_local_mcp_tools(entry)
+                local_mcp_tool_cache[server_name] = mcp_tools
+            except Exception as e:
+                logging.error(f"Error getting tools from local MCP server {server_name}: {str(e)}")
+                mcp_tools = []
+        result = []
+        for tool_obj in mcp_tools:
+            tool_name = getattr(tool_obj, 'name', None)
+            if tool_name:
+                function_tool = create_function_tool_for_local_mcp_tool(server_name, tool_name, tool_obj)
+                result.append(function_tool)
+        return result
+    else:
+        tool = to_openai_tool(entry)
+        if isinstance(tool, list):
+            return tool
+        else:
+            return [tool]

mcp_registry.sample.json

@@ -16,6 +16,12 @@
       "require_approval": {
         "never": { "tool_names": ["search"] }
       }
+    },
+    {
+      "name": "greeting",
+      "command": "python3",
+      "args": ["-u", "tests/greet_mcp.py"],
+      "allowed_tools": ["*"]
     }
   ]
 }

requirements.txt

@@ -1,5 +1,6 @@
 pydantic == 2.10.6
-gradio == 5.36.2
+gradio == 5.38.0
 openai == 1.76.0
 lxml
 RestrictedPython
+fastmcp

tests/greet_mcp.py

@@ -0,0 +1,16 @@
+# taken from FastMCP documentation at https://gofastmcp.com/servers/server: my_server.py
+from fastmcp import FastMCP
+
+mcp = FastMCP(name="MyServer")
+
+@mcp.tool
+def greet(name: str) -> str:
+    """Greet a user by name."""
+    return f"Hello, {name}! .:This message is powered by MCP.:"
+
+if __name__ == "__main__":
+    # This runs the server, defaulting to STDIO transport
+    mcp.run()
+    
+    # To use a different transport, e.g., Streamable HTTP:
+    # mcp.run(transport="http", host="127.0.0.1", port=9000)