"""
MCP server generator for MCP-Ables tools
"""
from typing import Any, Dict, Annotated, get_type_hints, List
from fastmcp import FastMCP
from pydantic import Field
from .yaml_parser import ToolSpec, ArgumentSpec
from .executor import CommandExecutor, ExecutionError
class MCPGenerator:
"""Generates FastMCP servers from tool specifications"""
# Map YAML types to Python types
TYPE_MAPPING = {
'string': str,
'int': int,
'float': float,
'bool': bool
}
def __init__(self, tool_specs: List[ToolSpec]):
"""
Initialize generator with a list of tool specifications
Args:
tool_specs: List of tool specifications to register
"""
self.tool_specs = tool_specs
self.executors = {spec.name: CommandExecutor(spec) for spec in tool_specs}
def create_server(self) -> FastMCP:
"""
Create and configure a FastMCP server with all tools
Returns:
FastMCP server with all tools registered
"""
# Use fixed server name "mcpables"
mcp = FastMCP("mcpables")
# Register each tool
for tool_spec in self.tool_specs:
executor = self.executors[tool_spec.name]
tool_func = self._create_tool_function(tool_spec, executor)
mcp.tool(tool_func)
return mcp
def _create_tool_function(self, tool_spec: ToolSpec, executor: CommandExecutor):
"""Dynamically create a function with proper type annotations for FastMCP"""
# Build the function signature parts
params = []
annotations = {}
defaults = []
# Track which params have defaults
params_with_defaults = []
params_without_defaults = []
for arg_name, arg_spec in tool_spec.args.items():
python_type = self.TYPE_MAPPING[arg_spec.type]
# Create annotated type with Field for description
annotated_type = Annotated[
python_type,
Field(description=arg_spec.description)
]
if arg_spec.required:
params_without_defaults.append(arg_name)
else:
params_with_defaults.append((arg_name, arg_spec.default))
annotations[arg_name] = annotated_type
# Build parameter list (required first, then optional)
all_params = params_without_defaults + [p[0] for p in params_with_defaults]
# Create the async function
tool_name = tool_spec.name
tool_description = tool_spec.description
# Build function code dynamically
param_list = []
for arg_name in params_without_defaults:
param_list.append(arg_name)
for arg_name, default_val in params_with_defaults:
param_list.append(f"{arg_name}={repr(default_val)}")
params_str = ', '.join(param_list)
# Create function using exec with proper namespace
namespace = {
'executor': executor,
'ExecutionError': ExecutionError,
'Field': Field,
'Annotated': Annotated,
**self.TYPE_MAPPING
}
func_code = f'''
async def {tool_name}({params_str}) -> str:
"""{tool_description}"""
kwargs = {{{', '.join(f'"{p}": {p}' for p in all_params)}}}
try:
result = await executor.execute(kwargs)
output = []
if result['stdout']:
output.append(f"=== STDOUT ===\\n{{result['stdout']}}")
if result['stderr']:
output.append(f"=== STDERR ===\\n{{result['stderr']}}")
output.append(f"=== EXIT CODE ===\\n{{result['exit_code']}}")
output.append(f"=== COMMAND ===\\n{{result['command']}}")
return "\\n\\n".join(output)
except ExecutionError as e:
return f"ERROR: {{str(e)}}"
except Exception as e:
return f"UNEXPECTED ERROR: {{str(e)}}"
'''
exec(func_code, namespace)
tool_func = namespace[tool_name]
# Add type annotations
tool_func.__annotations__ = annotations
tool_func.__annotations__['return'] = str
return tool_func
