samples: dynamic mcp with input json schema

Author: cristipufu

samples/mcp-dynamic-server/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mcp-dynamic-server"
-version = "0.0.7"
+version = "0.0.16"
 description = "Dynamic tools MCP Server"
 authors = [{ name = "John Doe" }]
 dependencies = [

samples/mcp-dynamic-server/server.py

@@ -1,57 +1,86 @@
 from typing import Any, Callable, Dict, List, Optional
 
 from mcp.server.fastmcp import FastMCP
+from mcp.types import Tool as MCPTool
 
 # Initialize the MCP server
 mcp = FastMCP("Self-Extending MCP Server")
 
-built_in_tools = {
-    "get_tools": {
-        "description": "Get a list of all available tools in the MCP server.",
-        "parameters": {}
-    },
-    "add_tool": {
-        "description": "Add a new tool to the MCP server.",
-        "parameters": {
-            "name": "Name of the tool (required)",
-            "code": "Python code implementing the tool function (required)",
-            "description": "Description of what the tool does (required)",
-            "param_descriptions": "Dictionary of parameter names to descriptions (optional)"
-        }
-    },
-    "call_tool": {
-        "description": "Call a registered tool with the given arguments.",
-        "parameters": {
-            "name": "Name of the tool to call (required)",
-            "args": "Dictionary of arguments to pass to the tool"
-        }
-    }
-}
+built_in_tools: List[MCPTool] = [
+    MCPTool(
+        name="add_tool",
+        description="Add a new tool to the MCP server by providing its Python code.",
+        inputSchema={
+            "type": "object",
+            "properties": {
+                "name": {"type": "string", "description": "Name of the tool"},
+                "code": {
+                    "type": "string",
+                    "description": "Python code implementing the tool's function. Must define a function with the specified 'name'. Type hints in the function signature for the input schema.",
+                },
+                "description": {
+                    "type": "string",
+                    "description": "Description of what the tool does",
+                },
+                "inputSchema": {
+                    "type": "object",
+                    "description": "JSON schema object describing the parameters the new tool expects (optional). This schema will be returned by get_tools and used for documentation.",
+                },
+            },
+            "required": ["name", "code", "description"],
+        },
+    ),
+    MCPTool(
+        name="call_tool",
+        description="Call a registered dynamic tool with the given arguments.",
+        inputSchema={
+            "type": "object",
+            "properties": {
+                "name": {
+                    "type": "string",
+                    "description": "Name of the dynamic tool to call",
+                },
+                "args": {
+                    "type": "object",
+                    "description": "Dictionary of arguments to pass to the tool. Must conform to the dynamic tool's inferred JSON input schema.",
+                },
+            },
+            "required": ["name", "args"],
+        },
+    ),
+]
+
 
 # Tool registry to track dynamically added tools
 class ToolRegistry:
     def __init__(self):
         self.tools = {}  # name -> function
         self.metadata = {}  # name -> metadata
 
-    def register(self, name: str, func: Callable, description: str, param_descriptions: Dict[str, str] = None):
+    def register(
+        self,
+        name: str,
+        func: Callable,
+        description: str,
+        inputSchema: Dict[str, Any] = None,
+    ):
         """Register a new tool in the registry."""
         self.tools[name] = func
         self.metadata[name] = {
             "name": name,
             "description": description,
-            "parameters": param_descriptions or {}
+            "inputSchema": inputSchema or {},
         }
 
     def get_tool(self, name: str) -> Optional[Callable]:
         """Get a tool by name."""
         return self.tools.get(name)
 
-    def get_metadata(self, name: str) -> Optional[Dict]:
+    def get_metadata(self, name: str) -> Optional[Dict[str, Any]]:
         """Get tool metadata by name."""
         return self.metadata.get(name)
 
-    def list_tools(self) -> List[Dict]:
+    def list_tools(self) -> List[Dict[str, Any]]:
         """List all registered tools."""
         return [self.metadata[name] for name in sorted(self.tools.keys())]
 
@@ -60,54 +89,46 @@ def has_tool(self, name: str) -> bool:
         return name in self.tools
 
 
-# Create the registry
 registry = ToolRegistry()
 
 
-# Core functionality: List available tools
-@mcp.tool()
-def get_tools() -> Dict[str, Any]:
+@mcp._mcp_server.list_tools()
+async def get_tools() -> List[MCPTool]:
     """Get a list of all available tools in the MCP server.
 
     Returns:
-        Dictionary with list of available tools and their metadata
+        List of available tools and their metadata
     """
-    try:
-        tools = registry.list_tools()
-
-        # Combine built-in tools with dynamic tools
-        all_tools = []
-        for name, info in built_in_tools.items():
-            all_tools.append({
-                "name": name,
-                "description": info["description"],
-                "parameters": info["parameters"],
-                "built_in": True
-            })
-
-        for tool in tools:
-            all_tools.append({
-                "name": tool["name"],
-                "description": tool["description"],
-                "parameters": tool["parameters"],
-                "built_in": False
-            })
-
-        return {"status": "success", "tools": all_tools}
-    except Exception as e:
-        return {"status": "error", "message": str(e)}
+    all_tools = list(built_in_tools)
+
+    tools = registry.list_tools()
+
+    for tool in tools:
+        all_tools.append(
+            MCPTool(
+                name=tool["name"],
+                description=tool["description"],
+                inputSchema=tool["inputSchema"],
+            )
+        )
+
+    return all_tools
 
 
-# Core functionality: Add a new tool
 @mcp.tool()
-def add_tool(name: str = None, code: str = None, description: str = None, param_descriptions: Dict[str, str] = None) -> Dict[str, Any]:
-    """Add a new tool to the MCP server.
+def add_tool(
+    name: str = None,
+    code: str = None,
+    description: str = None,
+    inputSchema: Dict[str, Any] = None,
+) -> Dict[str, Any]:
+    """Add a new tool to the MCP server by providing its Python code.
 
     Args:
-        name: Name of the tool
-        code: Python code implementing the tool function
-        description: Description of what the tool does
-        param_descriptions: Dictionary of parameter names to descriptions
+        name: Name of the tool (required)
+        code: Python code implementing the tool's function. Must define a function with the specified 'name'. Type hints in the function signature will be used to infer the input schema. (required)
+        description: Description of what the tool does (required)
+        inputSchema: JSON schema object describing the parameters the new tool expects (optional). This schema will be returned by get_tools and used for documentation.
 
     Returns:
         Dictionary with operation status
@@ -126,7 +147,7 @@ def add_tool(name: str = None, code: str = None, description: str = None, param_
             return {
                 "status": "error",
                 "message": f"Missing required parameters: {', '.join(missing_params)}",
-                "usage_example": "add_tool(name='tool_name', code='def tool_name(param1, param2):\\n    # code here\\n    return {\"status\": \"success\"}', description='Tool description', param_descriptions={'param1': 'Description of param1', 'param2': 'Description of param2'})"
+                "example": "add_tool(name='tool_name', code='def tool_name(param1: str, param2: str):\\n    # code here\\n    return {\"status\": \"success\"}', description='Tool description', inputSchema={'param1': 'Description of param1', 'param2': 'Description of param2'})",
             }
 
         # Check if tool already exists
@@ -156,15 +177,15 @@ def add_tool(name: str = None, code: str = None, description: str = None, param_
                 }
 
             # Register the tool with our registry
-            registry.register(name, func, description, param_descriptions)
+            registry.register(name, func, description, inputSchema)
 
             # Get the parameter information to return
-            params = registry.get_metadata(name)["parameters"]
+            params = registry.get_metadata(name)["inputSchema"]
 
             return {
                 "status": "success",
                 "message": f"Tool '{name}' added successfully",
-                "parameters": params
+                "inputSchema": params,
             }
 
         except SyntaxError as e:
@@ -179,14 +200,13 @@ def add_tool(name: str = None, code: str = None, description: str = None, param_
         return {"status": "error", "message": str(e)}
 
 
-# Core functionality: Call a tool
 @mcp.tool()
 def call_tool(name: str, args: Dict[str, Any] = None) -> Dict[str, Any]:
-    """Call a registered tool with the given arguments.
+    """Call a registered dynamic tool with the given arguments.
 
     Args:
-        name: Name of the tool to call
-        args: Dictionary of arguments to pass to the tool
+        name: Name of the dynamic tool to call (required)
+        args: Dictionary of arguments to pass to the tool. You should consult the tool's schema from get_tools to know the expected structure. (required)
 
     Returns:
         Dictionary with the tool's response
@@ -196,12 +216,13 @@ def call_tool(name: str, args: Dict[str, Any] = None) -> Dict[str, Any]:
 
     try:
         # Check if it's a built-in tool
-        if name in built_in_tools:
+        matching_tool = next((tool for tool in built_in_tools if tool.name == name), None)
+        if matching_tool:
             return {
                 "status": "error",
                 "message": f"Cannot call built-in tool '{name}' using call_tool",
                 "note": f"Use the {name} function directly instead of call_tool",
-                "parameters": built_in_tools[name]["parameters"]
+                "inputSchema": matching_tool.inputSchema,
             }
 
         # Get the tool
@@ -211,7 +232,7 @@ def call_tool(name: str, args: Dict[str, Any] = None) -> Dict[str, Any]:
             return {
                 "status": "error",
                 "message": f"Tool '{name}' not found",
-                "available_tools": [t["name"] for t in registry.list_tools()]
+                "available_tools": [t["name"] for t in registry.list_tools()],
             }
 
         # Call the tool with the provided arguments
@@ -220,12 +241,34 @@ def call_tool(name: str, args: Dict[str, Any] = None) -> Dict[str, Any]:
             return result
         except TypeError as e:
             # Likely an argument mismatch
-            params = registry.get_metadata(name)["parameters"]
+            params = registry.get_metadata(name)["inputSchema"]
+
+            # Build a usage example with actual parameter names
+            param_examples = {}
+
+            # Handle different possible inputSchema structures
+            if isinstance(params, dict):
+                if "properties" in params:
+                    # Standard JSON Schema format
+                    for param_name in params["properties"]:
+                        param_examples[param_name] = f"<{param_name}_value>"
+                else:
+                    # Simple dict of param_name -> description
+                    for param_name in params:
+                        param_examples[param_name] = f"<{param_name}_value>"
+
+            # If no parameters found or empty schema, provide generic example
+            if not param_examples:
+                param_examples = {"param1": "<value1>", "param2": "<value2>"}
+
+            # Format the dictionary for better readability
+            usage_str = str(param_examples).replace("'<", "<").replace(">'", ">")
+
             return {
                 "status": "error",
-                "message": f"Argument error calling tool '{name}': {str(e)}",
-                "expected_parameters": params,
-                "usage_example": f"call_tool(name='{name}', args={{'param1': value1, 'param2': value2, ...}})"
+                "message": f"Argument error calling tool '{name}': {str(e)}. Please fix your mistakes, add proper 'args' values!",
+                "inputSchema": params,
+                "example": f"call_tool(name='{name}', args={usage_str})",
             }
         except Exception as e:
             return {
@@ -236,6 +279,7 @@ def call_tool(name: str, args: Dict[str, Any] = None) -> Dict[str, Any]:
     except Exception as e:
         return {"status": "error", "message": str(e)}
 
+
 # Run the server when the script is executed
 if __name__ == "__main__":
     mcp.run()