Merge pull request #27 from shiraayal-tadata/bugfix/mcp-dynamically-creates-http-tool-function-with-args

Bugfix/mcp dynamically creates http tool function with args

Author: shira-ayal

README.md

@@ -135,6 +135,8 @@ Find the path to mcp-proxy by running in Terminal: `which mcp-proxy`.
 
 ## Development and Contributing
 
+**Notice:** We are currently refactoring our MCP auto-generation system. To avoid potential conflicts, we kindly request that you delay submitting contributions until this notice is removed from the README. Thank you for your understanding and patience.
+
 Thank you for considering contributing to FastAPI-MCP open source projects! It’s people like you that make it a reality for users in our community.
 
 Before you get started, please see [CONTRIBUTING.md](CONTRIBUTING.md).

fastapi_mcp/http_tools.py

@@ -6,98 +6,23 @@
 
 import json
 import logging
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 import httpx
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
 from mcp.server.fastmcp import FastMCP
-from pydantic import Field
 
-logger = logging.getLogger("fastapi_mcp")
-
-
-def resolve_schema_references(schema: Dict[str, Any], openapi_schema: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Resolve schema references in OpenAPI schemas.
+from .openapi_utils import (
+    clean_schema_for_display,
+    generate_example_from_schema,
+    get_python_type_and_default,
+    get_single_param_type_from_schema,
+    resolve_schema_references,
+    PYTHON_TYPE_IMPORTS,
+)
 
-    Args:
-        schema: The schema that may contain references
-        openapi_schema: The full OpenAPI schema to resolve references from
-
-    Returns:
-        The schema with references resolved
-    """
-    # Make a copy to avoid modifying the input schema
-    schema = schema.copy()
-
-    # Handle $ref directly in the schema
-    if "$ref" in schema:
-        ref_path = schema["$ref"]
-        # Standard OpenAPI references are in the format "#/components/schemas/ModelName"
-        if ref_path.startswith("#/components/schemas/"):
-            model_name = ref_path.split("/")[-1]
-            if "components" in openapi_schema and "schemas" in openapi_schema["components"]:
-                if model_name in openapi_schema["components"]["schemas"]:
-                    # Replace with the resolved schema
-                    ref_schema = openapi_schema["components"]["schemas"][model_name].copy()
-                    # Remove the $ref key and merge with the original schema
-                    schema.pop("$ref")
-                    schema.update(ref_schema)
-
-    # Handle array items
-    if "type" in schema and schema["type"] == "array" and "items" in schema:
-        schema["items"] = resolve_schema_references(schema["items"], openapi_schema)
-
-    # Handle object properties
-    if "properties" in schema:
-        for prop_name, prop_schema in schema["properties"].items():
-            schema["properties"][prop_name] = resolve_schema_references(prop_schema, openapi_schema)
-
-    return schema
-
-
-def clean_schema_for_display(schema: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Clean up a schema for display by removing internal fields.
-
-    Args:
-        schema: The schema to clean
-
-    Returns:
-        The cleaned schema
-    """
-    # Make a copy to avoid modifying the input schema
-    schema = schema.copy()
-
-    # Remove common internal fields that are not helpful for LLMs
-    fields_to_remove = [
-        "allOf",
-        "anyOf",
-        "oneOf",
-        "nullable",
-        "discriminator",
-        "readOnly",
-        "writeOnly",
-        "xml",
-        "externalDocs",
-    ]
-    for field in fields_to_remove:
-        if field in schema:
-            schema.pop(field)
-
-    # Process nested properties
-    if "properties" in schema:
-        for prop_name, prop_schema in schema["properties"].items():
-            if isinstance(prop_schema, dict):
-                schema["properties"][prop_name] = clean_schema_for_display(prop_schema)
-
-    # Process array items
-    if "type" in schema and schema["type"] == "array" and "items" in schema:
-        if isinstance(schema["items"], dict):
-            schema["items"] = clean_schema_for_display(schema["items"])
-
-    return schema
+logger = logging.getLogger("fastapi_mcp")
 
 
 def create_mcp_tools_from_openapi(
@@ -126,6 +51,9 @@ def create_mcp_tools_from_openapi(
         routes=app.routes,
     )
 
+    # Resolve all references in the schema at once
+    resolved_openapi_schema = resolve_schema_references(openapi_schema, openapi_schema)
+
     if not base_url:
         # Try to determine the base URL from FastAPI config
         if hasattr(app, "root_path") and app.root_path:
@@ -144,7 +72,7 @@ def create_mcp_tools_from_openapi(
         base_url = base_url[:-1]
 
     # Process each path in the OpenAPI schema
-    for path, path_item in openapi_schema.get("paths", {}).items():
+    for path, path_item in resolved_openapi_schema.get("paths", {}).items():
         for method, operation in path_item.items():
             # Skip non-HTTP methods
             if method not in ["get", "post", "put", "delete", "patch"]:
@@ -167,11 +95,42 @@ def create_mcp_tools_from_openapi(
                 parameters=operation.get("parameters", []),
                 request_body=operation.get("requestBody", {}),
                 responses=operation.get("responses", {}),
-                openapi_schema=openapi_schema,
+                openapi_schema=resolved_openapi_schema,
                 describe_all_responses=describe_all_responses,
                 describe_full_response_schema=describe_full_response_schema,
             )
 
+def _create_http_tool_function(function_template: Callable, properties: Dict[str, Any], additional_variables: Dict[str, Any]) -> Callable:
+    # Build parameter string with type hints
+    parsed_parameters = {}
+    parsed_parameters_with_defaults = {}
+    for param_name, parsed_param_schema in properties.items():
+        type_hint, has_default_value = get_python_type_and_default(parsed_param_schema)
+        if has_default_value:
+            parsed_parameters_with_defaults[param_name] = f"{param_name}: {type_hint}"
+        else:
+            parsed_parameters[param_name] = f"{param_name}: {type_hint}"
+
+    parsed_parameters_keys = list(parsed_parameters.keys()) + list(parsed_parameters_with_defaults.keys())
+    parsed_parameters_values = list(parsed_parameters.values()) + list(parsed_parameters_with_defaults.values())
+    parameters_str = ", ".join(parsed_parameters_values)
+    kwargs_str = ', '.join([f"'{k}': {k}" for k in parsed_parameters_keys])
+
+    dynamic_function_body = f"""async def dynamic_http_tool_function({parameters_str}):
+        kwargs = {{{kwargs_str}}}
+        return await http_tool_function_template(**kwargs)
+    """
+
+    # Create function namespace with required imports
+    namespace = {
+        "http_tool_function_template": function_template,
+        **PYTHON_TYPE_IMPORTS,
+        **additional_variables
+    }
+    
+    # Execute the dynamic function definition
+    exec(dynamic_function_body, namespace)
+    return namespace["dynamic_http_tool_function"]
 
 def create_http_tool(
     mcp_server: FastMCP,
@@ -247,23 +206,13 @@ def create_http_tool(
                         schema = content_data["schema"]
                         response_info += f"\nContent-Type: {content_type}"
 
-                        # Resolve any schema references
-                        resolved_schema = resolve_schema_references(schema, openapi_schema)
-
                         # Clean the schema for display
-                        display_schema = clean_schema_for_display(resolved_schema)
+                        display_schema = clean_schema_for_display(schema)
 
                         # Get model name if it's a referenced model
                         model_name = None
                         model_examples = None
                         items_model_name = None
-                        if "$ref" in schema:
-                            ref_path = schema["$ref"]
-                            if ref_path.startswith("#/components/schemas/"):
-                                model_name = ref_path.split("/")[-1]
-                                response_info += f"\nModel: {model_name}"
-                                # Try to get examples from the model
-                                model_examples = extract_model_examples_from_components(model_name, openapi_schema)
 
                         # Check if this is an array of items
                         if schema.get("type") == "array" and "items" in schema and "$ref" in schema["items"]:
@@ -358,7 +307,6 @@ def create_http_tool(
     query_params = []
     header_params = []
     body_params = []
-
     for param in parameters:
         param_name = param.get("name")
         param_in = param.get("in")
@@ -416,10 +364,12 @@ def create_http_tool(
         param_required = param.get("required", False)
 
         properties[param_name] = {
-            "type": param_schema.get("type", "string"),
+            "type": get_single_param_type_from_schema(param_schema),
             "title": param_name,
             "description": param_desc,
         }
+        if "default" in param_schema:
+            properties[param_name]["default"] = param_schema["default"]
 
         if param_required:
             required_props.append(param_name)
@@ -429,14 +379,25 @@ def create_http_tool(
         param_schema = param.get("schema", {})
         param_required = param.get("required", False)
 
-        properties[param_name] = param_schema
-        properties[param_name]["title"] = param_name
+        # properties[param_name] = param_schema
+        properties[param_name] = {
+            "type": get_single_param_type_from_schema(param_schema),
+            "title": param_name,
+        }
+        if "default" in param_schema:
+            properties[param_name]["default"] = param_schema["default"]
 
         if param_required:
             required_props.append(param_name)
 
-    # Function to dynamically call the API endpoint
-    async def http_tool_function(kwargs: Dict[str, Any] = Field(default_factory=dict)):
+    # Create a proper input schema for the tool
+    input_schema = {"type": "object", "properties": properties, "title": f"{operation_id}Arguments"}
+
+    if required_props:
+        input_schema["required"] = required_props
+
+    # Dynamically create a function to  call the API endpoint
+    async def http_tool_function_template(**kwargs):
         # Prepare URL with path parameters
         url = f"{base_url}{path}"
         for param_name, _ in path_params:
@@ -480,13 +441,9 @@ async def http_tool_function(kwargs: Dict[str, Any] = Field(default_factory=dict
         except ValueError:
             return response.text
 
-    # Create a proper input schema for the tool
-    input_schema = {"type": "object", "properties": properties, "title": f"{operation_id}Arguments"}
-
-    if required_props:
-        input_schema["required"] = required_props
-
-    # Set the function name and docstring
+    # Create the http_tool_function (with name and docstring)
+    additional_variables = {"path_params": path_params, "query_params": query_params, "header_params": header_params}
+    http_tool_function = _create_http_tool_function(http_tool_function_template, properties, additional_variables) # type: ignore
     http_tool_function.__name__ = operation_id
     http_tool_function.__doc__ = tool_description
 
@@ -499,116 +456,3 @@ async def http_tool_function(kwargs: Dict[str, Any] = Field(default_factory=dict
 
     # Update the tool's parameters to use our custom schema instead of the auto-generated one
     tool.parameters = input_schema
-
-
-def extract_model_examples_from_components(
-    model_name: str, openapi_schema: Dict[str, Any]
-) -> Optional[List[Dict[str, Any]]]:
-    """
-    Extract examples from a model definition in the OpenAPI components.
-
-    Args:
-        model_name: The name of the model to extract examples from
-        openapi_schema: The full OpenAPI schema
-
-    Returns:
-        List of example dictionaries if found, None otherwise
-    """
-    if "components" not in openapi_schema or "schemas" not in openapi_schema["components"]:
-        return None
-
-    if model_name not in openapi_schema["components"]["schemas"]:
-        return None
-
-    schema = openapi_schema["components"]["schemas"][model_name]
-
-    # Look for examples in the schema
-    examples = None
-
-    # Check for examples field directly (OpenAPI 3.1.0+)
-    if "examples" in schema:
-        examples = schema["examples"]
-    # Check for example field (older OpenAPI versions)
-    elif "example" in schema:
-        examples = [schema["example"]]
-
-    return examples
-
-
-def generate_example_from_schema(schema: Dict[str, Any], model_name: Optional[str] = None) -> Any:
-    """
-    Generate a simple example response from a JSON schema.
-
-    Args:
-        schema: The JSON schema to generate an example from
-        model_name: Optional model name for special handling
-
-    Returns:
-        An example object based on the schema
-    """
-    if not schema or not isinstance(schema, dict):
-        return None
-
-    # Special handling for known model types
-    if model_name == "Item":
-        # Create a realistic Item example since this is commonly used
-        return {
-            "id": 1,
-            "name": "Hammer",
-            "description": "A tool for hammering nails",
-            "price": 9.99,
-            "tags": ["tool", "hardware"],
-        }
-    elif model_name == "HTTPValidationError":
-        # Create a realistic validation error example
-        return {"detail": [{"loc": ["body", "name"], "msg": "field required", "type": "value_error.missing"}]}
-
-    # Handle different types
-    schema_type = schema.get("type")
-
-    if schema_type == "object":
-        result = {}
-        if "properties" in schema:
-            for prop_name, prop_schema in schema["properties"].items():
-                # Generate an example for each property
-                prop_example = generate_example_from_schema(prop_schema)
-                if prop_example is not None:
-                    result[prop_name] = prop_example
-        return result
-
-    elif schema_type == "array":
-        if "items" in schema:
-            # Generate a single example item
-            item_example = generate_example_from_schema(schema["items"])
-            if item_example is not None:
-                return [item_example]
-        return []
-
-    elif schema_type == "string":
-        # Check if there's a format
-        format_type = schema.get("format")
-        if format_type == "date-time":
-            return "2023-01-01T00:00:00Z"
-        elif format_type == "date":
-            return "2023-01-01"
-        elif format_type == "email":
-            return "[email protected]"
-        elif format_type == "uri":
-            return "https://example.com"
-        # Use title or property name if available
-        return schema.get("title", "string")
-
-    elif schema_type == "integer":
-        return 1
-
-    elif schema_type == "number":
-        return 1.0
-
-    elif schema_type == "boolean":
-        return True
-
-    elif schema_type == "null":
-        return None
-
-    # Default case
-    return None