Skip to content

samples: dynamic mcp with input json schema #89

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 4, 2025
Merged

samples: dynamic mcp with input json schema #89

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion samples/mcp-dynamic-server/pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 = [
Expand Down
198 changes: 121 additions & 77 deletions samples/mcp-dynamic-server/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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())]

Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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 {
Expand All @@ -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()