"""
MCP tool generator for OpenAPI operations.
"""
import json
from typing import Any, Dict, List, Optional, Callable
from .models import APIOperation, Parameter, RequestBody
from .client import APIClient
class ToolGenerator:
"""Generates MCP tools from OpenAPI operations."""
def __init__(self, api_client: APIClient):
self.api_client = api_client
def generate_tool_function(self, operation: APIOperation) -> Callable:
"""Generate a tool function for an API operation."""
# Create function parameters dynamically
import inspect
from typing import Optional, Any
# Build parameter list for the function signature
params = []
param_annotations = {}
# Add path parameters
for param in operation.parameters:
param_name = param.name.replace('-', '_') # Replace hyphens with underscores for valid Python names
# Determine parameter type
param_type = str # Default to string
if param.schema and 'type' in param.schema:
schema_type = param.schema['type']
if schema_type == 'integer':
param_type = int
elif schema_type == 'number':
param_type = float
elif schema_type == 'boolean':
param_type = bool
# Make optional parameters have Optional type
if not param.required:
param_type = Optional[param_type]
# Add default value for optional parameters
params.append(inspect.Parameter(
param_name,
inspect.Parameter.KEYWORD_ONLY,
default=None,
annotation=param_type
))
else:
params.append(inspect.Parameter(
param_name,
inspect.Parameter.KEYWORD_ONLY,
annotation=param_type
))
param_annotations[param_name] = param_type
# Add request body parameter if exists
if operation.request_body:
params.append(inspect.Parameter(
'body',
inspect.Parameter.KEYWORD_ONLY,
default=None if not operation.request_body.required else inspect.Parameter.empty,
annotation=Optional[dict] if not operation.request_body.required else dict
))
param_annotations['body'] = Optional[dict] if not operation.request_body.required else dict
# Create the function signature
sig = inspect.Signature(params, return_annotation=Dict[str, Any])
# Create the actual function
async def tool_function(*args, **kwargs) -> Dict[str, Any]:
"""Generated tool function for API operation."""
# Map back parameter names (underscores to hyphens for API calls)
api_kwargs = {}
for key, value in kwargs.items():
# Map back underscore names to original parameter names
original_name = key
for param in operation.parameters:
if param.name.replace('-', '_') == key:
original_name = param.name
break
api_kwargs[original_name] = value
# Extract path parameters
path_params = {}
query_params = {}
headers = {}
request_body = None
for param in operation.parameters:
param_name = param.name
param_value = api_kwargs.get(param_name)
if param.required and param_value is None:
raise ValueError(f"Required parameter '{param_name}' is missing")
if param_value is not None:
if param.param_type == "path":
path_params[param_name] = param_value
elif param.param_type == "query":
query_params[param_name] = param_value
elif param.param_type == "header":
headers[param_name] = param_value
# Handle request body
if operation.request_body:
body_data = api_kwargs.get("body")
if operation.request_body.required and body_data is None:
raise ValueError("Request body is required")
if body_data is not None:
request_body = body_data
# Make the API call
result = await self.api_client.call_operation(
operation=operation,
path_params=path_params,
query_params=query_params,
headers=headers,
body=request_body
)
return result
# Set the signature and metadata
tool_function.__signature__ = sig
tool_function.__name__ = operation.operation_id
tool_function.__doc__ = self._generate_tool_docstring(operation)
tool_function.__annotations__ = param_annotations
return tool_function
def _generate_tool_docstring(self, operation: APIOperation) -> str:
"""Generate a docstring for the tool function."""
lines = []
# Add summary and description
if operation.summary:
lines.append(operation.summary)
if operation.description:
if operation.summary:
lines.append("")
lines.append(operation.description)
# Add parameter documentation
if operation.parameters or operation.request_body:
lines.append("")
lines.append("Parameters:")
for param in operation.parameters:
param_doc = f"- {param.name} ({param.param_type})"
if param.required:
param_doc += " [required]"
if param.description:
param_doc += f": {param.description}"
lines.append(param_doc)
if operation.request_body:
body_doc = "- body (request body)"
if operation.request_body.required:
body_doc += " [required]"
if operation.request_body.description:
body_doc += f": {operation.request_body.description}"
lines.append(body_doc)
# Add response information
if operation.responses:
lines.append("")
lines.append("Returns:")
success_responses = [
status for status in operation.responses.keys()
if status.startswith("2")
]
if success_responses:
status = success_responses[0]
response = operation.responses[status]
if response.description:
lines.append(f"- {response.description}")
else:
lines.append(f"- HTTP {status} response")
return "\n".join(lines)
def generate_tool_schema(self, operation: APIOperation) -> Dict[str, Any]:
"""Generate JSON schema for the tool parameters."""
properties = {}
required = []
# Add parameters
for param in operation.parameters:
param_schema = self._convert_openapi_schema_to_json_schema(param.schema)
if param.description:
param_schema["description"] = param.description
if param.example is not None:
param_schema["example"] = param.example
properties[param.name] = param_schema
if param.required:
required.append(param.name)
# Add request body
if operation.request_body:
body_schema = self._convert_openapi_schema_to_json_schema(
operation.request_body.schema
)
if operation.request_body.description:
body_schema["description"] = operation.request_body.description
properties["body"] = body_schema
if operation.request_body.required:
required.append("body")
schema = {
"type": "object",
"properties": properties,
}
if required:
schema["required"] = required
return schema
def _convert_openapi_schema_to_json_schema(self, openapi_schema: Dict[str, Any]) -> Dict[str, Any]:
"""Convert OpenAPI schema to JSON Schema format."""
# For now, we'll do a simple conversion
# In a full implementation, this would handle more complex cases
schema = openapi_schema.copy()
# Handle common OpenAPI to JSON Schema conversions
if "nullable" in schema:
if schema.pop("nullable"):
# Make the type nullable
current_type = schema.get("type")
if current_type:
schema["type"] = [current_type, "null"]
# Handle format conversions
if schema.get("type") == "string" and schema.get("format") == "binary":
# For file uploads, we'll treat as string for now
schema.pop("format", None)
return schema
