From eda671fc5efe1d7e8f59ae898b8cba035d98ea12 Mon Sep 17 00:00:00 2001 From: Cristian Pufu Date: Sun, 4 May 2025 16:38:46 +0300 Subject: [PATCH] samples: dynamic mcp with input json schema --- samples/mcp-dynamic-server/pyproject.toml | 2 +- samples/mcp-dynamic-server/server.py | 198 +++++++++++++--------- 2 files changed, 122 insertions(+), 78 deletions(-) diff --git a/samples/mcp-dynamic-server/pyproject.toml b/samples/mcp-dynamic-server/pyproject.toml index 5d96491..53966ff 100644 --- a/samples/mcp-dynamic-server/pyproject.toml +++ b/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 = [ diff --git a/samples/mcp-dynamic-server/server.py b/samples/mcp-dynamic-server/server.py index 4bf20fe..e362068 100644 --- a/samples/mcp-dynamic-server/server.py +++ b/samples/mcp-dynamic-server/server.py @@ -1,32 +1,55 @@ 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: @@ -34,24 +57,30 @@ 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": "", "param2": ""} + + # 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()