windmill/backend/generate_mcp_endpoints_tools/generate_mcp_tools.py

#!/usr/bin/env python3
"""
Script to parse the OpenAPI YAML file and generate Rust and TypeScript code with MCP tools.
Searches for endpoints tagged with 'x-mcp-tool: true' and creates:
- Rust code: A const array in backend/windmill-api/src/mcp/tools/auto_generated_endpoints.rs
- TypeScript code: An exported array in frontend/src/lib/mcpEndpointTools.ts
"""

import json
import sys
from pathlib import Path
from typing import Dict, List, Any, Optional

IMPORTS = """
use std::borrow::Cow;
use windmill_mcp::server::EndpointTool;
"""

def load_openapi_spec(file_path: str) -> Dict[str, Any]:
    """Load and parse the OpenAPI YAML specification."""
    try:
        import yaml
        with open(file_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    except ImportError:
        print("PyYAML not found. Please install it with: pip install PyYAML", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"Error loading OpenAPI spec: {e}", file=sys.stderr)
        sys.exit(1)

def flatten_allof_schema(schema: Dict[str, Any]) -> Dict[str, Any]:
    """Flatten an allOf schema into a single object schema by merging all properties."""
    if 'allOf' not in schema:
        return schema

    merged = {"type": "object", "properties": {}, "required": []}

    def collect_from(s: Dict[str, Any]):
        if 'allOf' in s:
            for item in s['allOf']:
                if isinstance(item, dict):
                    collect_from(item)
        if 'properties' in s:
            merged['properties'].update(s['properties'])
        if 'required' in s and isinstance(s['required'], list):
            merged['required'].extend(s['required'])
        if 'description' in s and 'description' not in merged:
            merged['description'] = s['description']

    collect_from(schema)

    # Preserve additional top-level keys from the original schema
    preserved_keys = {'additionalProperties', 'title', 'nullable', 'default', 'example'}
    for key in preserved_keys:
        if key in schema and key not in merged:
            merged[key] = schema[key]

    if not merged['required']:
        del merged['required']

    return merged

def extract_separate_schemas(parameters: List[Dict[str, Any]], request_body: Optional[Dict[str, Any]], spec: Dict[str, Any], required_fields: Optional[List[str]] = None, base_path: str = "", include_fields: Optional[List[str]] = None, opaque_fields: Optional[List[str]] = None, include_query_params: Optional[List[str]] = None) -> tuple:
    """Extract separate schemas for path parameters, query parameters, and request body."""
    path_params_schema = {
        "type": "object",
        "properties": {},
        "required": []
    }
    
    query_params_schema = {
        "type": "object", 
        "properties": {},
        "required": []
    }
    
    body_schema = None
    
    # Process parameters
    for param in parameters:
        # Resolve $ref if present
        if '$ref' in param:
            param = resolve_schema_refs(param, spec, base_path)

        param_name = param.get('name', '')
        param_schema = param.get('schema', {'type': 'string'})
        param_required = param.get('required', False)
        param_description = param.get('description', '')
        param_in = param.get('in', 'query')

        # Resolve any refs in the parameter schema
        param_schema = resolve_schema_refs(param_schema, spec, base_path)
        
        # Add description if available
        if param_description:
            param_schema = dict(param_schema)
            param_schema['description'] = param_description
        
        # Route to appropriate schema based on parameter location
        if param_in == 'path':
            # Skip 'workspace' path parameter as it's automatically provided by the MCP context
            if param_name != 'workspace':
                path_params_schema['properties'][param_name] = param_schema
                if param_required:
                    path_params_schema['required'].append(param_name)
        elif param_in == 'query' and (include_query_params is None or param_name in include_query_params):
            query_params_schema['properties'][param_name] = param_schema
            if param_required:
                query_params_schema['required'].append(param_name)
    
    # Process request body if present
    if request_body:
        body_schema = extract_request_body_schema(request_body, spec, base_path)

        # Flatten allOf schemas into a single object schema for filtering
        if body_schema and (include_fields is not None or opaque_fields):
            body_schema = flatten_allof_schema(body_schema)

        # Apply include_fields filter: only keep listed top-level properties
        if body_schema and include_fields is not None and 'properties' in body_schema:
            body_schema['properties'] = {
                k: v for k, v in body_schema['properties'].items()
                if k in include_fields
            }
            if 'required' in body_schema:
                body_schema['required'] = [
                    r for r in body_schema['required'] if r in include_fields
                ]

        # Apply opaque_fields: simplify listed properties to {"type": "object"}
        if body_schema and opaque_fields and 'properties' in body_schema:
            for field in opaque_fields:
                if field in body_schema['properties']:
                    body_schema['properties'][field] = {"type": "object"}

        # If we have required fields specified and a body schema, update the required array
        if body_schema and required_fields:
            if 'required' not in body_schema:
                body_schema['required'] = []
            
            # Add each required field if it exists in the schema properties
            for field in required_fields:
                if 'properties' in body_schema and field in body_schema['properties']:
                    if field not in body_schema['required']:
                        body_schema['required'].append(field)
                else:
                    # Log warning when a required field is missing from schema properties
                    print(f"Warning: Required field '{field}' not found in body schema properties", file=sys.stderr)

        # Marks a body the MCP layer must refuse to leave empty (see non_empty_body_fields
        # in windmill-mcp). A pass-through body (no declared properties, e.g.
        # runScriptByPath) is excluded: `{}` is a valid body there, for a runnable that
        # takes no arguments.
        if body_schema and request_body.get('required') and body_schema.get('properties'):
            body_schema['minProperties'] = 1

    # Sanitize empty schemas for JSON Schema draft 2020-12 compliance
    path_params_schema = sanitize_empty_schemas(path_params_schema)
    query_params_schema = sanitize_empty_schemas(query_params_schema)
    body_schema = sanitize_empty_schemas(body_schema)

    # Convert enums to descriptions for client compatibility
    path_params_schema = convert_enums_to_descriptions(path_params_schema)
    query_params_schema = convert_enums_to_descriptions(query_params_schema)
    body_schema = convert_enums_to_descriptions(body_schema)

    # Detect overlapping property names across schemas and rename with suffixes
    path_keys = set(path_params_schema['properties'].keys()) if path_params_schema and path_params_schema.get('properties') else set()
    query_keys = set(query_params_schema['properties'].keys()) if query_params_schema and query_params_schema.get('properties') else set()
    body_keys = set(body_schema['properties'].keys()) if body_schema and body_schema.get('properties') else set()

    conflicts = (path_keys & query_keys) | (path_keys & body_keys) | (query_keys & body_keys)

    query_field_renames = {}
    body_field_renames = {}

    for field in conflicts:
        # The path parameter keeps the plain name: it identifies the item, is what a
        # caller reaches for first, and matches the non-colliding endpoints (`getFlowByPath`
        # takes `path`). Only the other locations carry a suffix.
        schemas_and_renames = [
            (query_params_schema, query_keys, '__query', query_field_renames),
            (body_schema, body_keys, '__body', body_field_renames),
        ]
        for schema, keys, suffix, renames_map in schemas_and_renames:
            if field in keys and schema and 'properties' in schema:
                new_name = field + suffix
                # Rename in properties
                schema['properties'][new_name] = schema['properties'].pop(field)
                # Update description to clarify the renamed field
                if 'description' not in schema['properties'][new_name]:
                    schema['properties'][new_name] = dict(schema['properties'][new_name])
                prop = schema['properties'][new_name]
                if isinstance(prop, dict):
                    existing_desc = prop.get('description', '')
                    location = suffix.lstrip('_')
                    if not existing_desc:
                        prop['description'] = f"({location} parameter)"
                    else:
                        prop['description'] = f"{existing_desc} ({location} parameter)"
                # Rename in required array
                if 'required' in schema and field in schema['required']:
                    schema['required'] = [new_name if r == field else r for r in schema['required']]
                # Store the reverse mapping: renamed -> original
                renames_map[new_name] = field

        # A body field colliding with a same-named path parameter (`path` on the update
        # endpoints) holds the new value and differs only when moving the item, so it must
        # stay optional; the server defaults it from the URL path when the caller omits it.
        if field in path_keys and field in body_keys and body_schema:
            body_name = field + '__body'
            if 'required' in body_schema:
                body_schema['required'] = [r for r in body_schema['required'] if r != body_name]
            prop = body_schema['properties'].get(body_name)
            if isinstance(prop, dict):
                existing_desc = prop.get('description', '').rstrip('. ')
                prop['description'] = (
                    f"{existing_desc}. Defaults to `{field}` when omitted; "
                    f"set it only to change the {field}."
                ).lstrip('. ')

    # Return None for empty schemas
    path_params_schema = path_params_schema if path_params_schema and path_params_schema.get('properties') else None
    query_params_schema = query_params_schema if query_params_schema and query_params_schema.get('properties') else None

    return (path_params_schema, query_params_schema, body_schema, query_field_renames, body_field_renames)

# Cache for loaded external files
_external_file_cache: Dict[str, Dict[str, Any]] = {}

def load_external_file(file_path: str, base_path: str) -> Optional[Dict[str, Any]]:
    """Load an external YAML file relative to the base path."""
    if file_path in _external_file_cache:
        return _external_file_cache[file_path]

    try:
        import yaml
        from pathlib import Path

        # Resolve the path relative to the base file
        base_dir = Path(base_path).parent
        full_path = (base_dir / file_path).resolve()

        with open(full_path, 'r', encoding='utf-8') as f:
            content = yaml.safe_load(f)
            _external_file_cache[file_path] = content
            return content
    except Exception as e:
        print(f"Warning: Could not load external file {file_path}: {e}", file=sys.stderr)
        return None

def resolve_ref(ref_path: str, spec: Dict[str, Any], base_path: str = "") -> tuple:
    """Resolve a $ref path to the actual schema definition.

    Handles both internal refs (#/...) and external file refs (file.yaml#/...).

    Returns a tuple of (resolved_schema, resolved_spec) where resolved_spec is the spec
    that should be used for resolving any nested refs within the resolved schema.
    """
    # Check if this is an external file reference
    if '#' in ref_path and not ref_path.startswith('#'):
        # External file reference: "../../openflow.openapi.yaml#/components/schemas/Retry"
        file_part, fragment = ref_path.split('#', 1)
        external_spec = load_external_file(file_part, base_path)
        if external_spec is None:
            return None, spec
        # Resolve the fragment within the external file, and return external_spec for nested refs
        resolved, _ = resolve_ref('#' + fragment, external_spec, base_path)
        return resolved, external_spec

    if not ref_path.startswith('#/'):
        return None, spec

    # Remove the '#/' prefix and split by '/'
    path_parts = ref_path[2:].split('/')

    # Navigate through the spec following the path
    current = spec
    for part in path_parts:
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return None, spec

    return (current if isinstance(current, dict) else None), spec

def resolve_schema_refs(schema: Dict[str, Any], spec: Dict[str, Any], base_path: str = "", _visited_refs: Optional[set] = None) -> Dict[str, Any]:
    """Recursively resolve all $ref references in a schema."""
    if _visited_refs is None:
        _visited_refs = set()

    if not isinstance(schema, dict):
        return schema

    # If this is a $ref, resolve it
    if '$ref' in schema:
        ref_path = schema['$ref']
        if ref_path in _visited_refs:
            # Circular reference detected - return empty object to break the cycle
            return {"type": "object"}
        _visited_refs = _visited_refs | {ref_path}
        resolved, resolved_spec = resolve_ref(ref_path, spec, base_path)
        if resolved:
            # Recursively resolve any refs in the resolved schema using the appropriate spec
            return resolve_schema_refs(resolved, resolved_spec, base_path, _visited_refs)
        else:
            print(f"Warning: Could not resolve $ref: {ref_path}")
            return schema

    # Recursively process all values in the schema
    resolved_schema = {}
    for key, value in schema.items():
        if isinstance(value, dict):
            resolved_schema[key] = resolve_schema_refs(value, spec, base_path, _visited_refs)
        elif isinstance(value, list):
            resolved_schema[key] = [
                resolve_schema_refs(item, spec, base_path, _visited_refs) if isinstance(item, dict) else item
                for item in value
            ]
        else:
            resolved_schema[key] = value

    return resolved_schema

def convert_enums_to_descriptions(schema: Any) -> Any:
    """Recursively convert enum arrays into description text to avoid client compatibility issues."""
    if isinstance(schema, list):
        return [convert_enums_to_descriptions(item) for item in schema]
    if not isinstance(schema, dict):
        return schema

    result = {}
    enum_value = None
    # First pass: copy all non-enum keys so 'description' is available before enum processing
    for key, value in schema.items():
        if key == 'enum':
            enum_value = value
        else:
            result[key] = convert_enums_to_descriptions(value)

    # Second pass: process enum using the already-copied description
    if enum_value is not None:
        values_str = ', '.join(str(v) for v in enum_value)
        existing = result.get('description', '')
        enum_desc = f"Possible values: {values_str}"
        result['description'] = f"{existing}. {enum_desc}" if existing else enum_desc

    return result

def sanitize_empty_schemas(schema: Any) -> Any:
    """Replace empty {} schemas with valid JSON Schema draft 2020-12 equivalents.

    In OpenAPI, {} means 'any value' but strict JSON Schema validators (e.g. Claude's API)
    reject empty objects. This converts them to proper schemas.
    """
    if isinstance(schema, list):
        return [sanitize_empty_schemas(item) for item in schema]
    if not isinstance(schema, dict):
        return schema

    result = {}
    for key, value in schema.items():
        if key == 'additionalProperties' and isinstance(value, dict) and len(value) == 0:
            result[key] = True
        elif key == 'properties' and isinstance(value, dict):
            # properties is a map of name -> schema; sanitize each property schema
            result[key] = {
                k: {"type": "object"} if isinstance(v, dict) and len(v) == 0 else sanitize_empty_schemas(v)
                for k, v in value.items()
            }
        else:
            result[key] = sanitize_empty_schemas(value)
    return result

def extract_request_body_schema(request_body: Dict[str, Any], spec: Dict[str, Any], base_path: str = "") -> Optional[Dict[str, Any]]:
    """Extract request body schema from OpenAPI requestBody definition and resolve refs."""
    if not request_body:
        return None

    content = request_body.get('content', {})
    json_content = content.get('application/json', {})
    schema = json_content.get('schema', {})

    if schema:
        # Resolve any $ref references in the schema
        return resolve_schema_refs(schema, spec, base_path)

    return None

def http_method_to_rust(method: str) -> str:
    """Convert HTTP method string to Rust http::Method enum."""
    method_map = {
        'get': 'http::Method::GET',
        'post': 'http::Method::POST',
        'put': 'http::Method::PUT',
        'delete': 'http::Method::DELETE',
        'patch': 'http::Method::PATCH',
        'head': 'http::Method::HEAD',
        'options': 'http::Method::OPTIONS'
    }
    return method_map.get(method.lower(), f'http::Method::{method.upper()}')

def schema_to_rust_value(schema: Optional[Dict[str, Any]]) -> str:
    """Convert a schema dict to a Rust serde_json::json! expression."""
    if schema is None:
        return "None"
    # ensure_ascii=False: json.dumps would escape a non-ASCII character in a
    # description as \uXXXX, which Rust rejects — its string literals spell it
    # \u{XXXX}. Emitting the character itself is valid in both.
    return f"Some(serde_json::json!({json.dumps(schema, indent=8, ensure_ascii=False)}))"

def build_tool_description(operation: Dict[str, Any], method: str, path: str) -> str:
    """Build the MCP tool description from OpenAPI summary and description.

    `x-mcp-tool-description` stands in for the operation's own description, for a route
    whose two audiences need different text: the description documents the endpoint,
    including fields like `parent_hash` that `x-mcp-tool-include-fields` deliberately
    keeps out of the tool, and naming one there tells an agent to send what its schema
    does not offer.
    """
    summary = operation.get('summary', '').strip()
    description = (
        operation.get('x-mcp-tool-description') or operation.get('description', '')
    ).strip()

    if summary and description:
        return f"{summary}: {description}".rstrip('.!? ')
    if summary:
        return summary
    if description:
        return description.rstrip('.!? ')
    return f'{method.upper()} {path}'

def find_mcp_tools(spec: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Find all endpoints marked with x-mcp-tool: true."""
    tools = []
    paths = spec.get('paths', {})
    
    for path, path_item in paths.items():
        for method, operation in path_item.items():
            if isinstance(operation, dict) and operation.get('x-mcp-tool') is True:
                # Extract tool information
                tool = {
                    # The name an agent sees. Defaults to the operationId, which
                    # every generated client is keyed on, so an endpoint whose
                    # tool should read differently overrides it here rather than
                    # by renaming the operation.
                    'name': operation.get('x-mcp-tool-name') or operation.get('operationId', f"{method}_{path.replace('/', '_').replace('{', '').replace('}', '')}"),
                    'description': build_tool_description(operation, method, path),
                    'instructions': operation.get('x-mcp-instructions', ''),
                    'path': path,
                    'method': method.upper(),
                    'parameters': operation.get('parameters', []),
                    'requestBody': operation.get('requestBody'),
                    'required_fields': operation.get('x-mcp-required-fields', []),
                    'include_fields': operation.get('x-mcp-tool-include-fields'),
                    'opaque_fields': operation.get('x-mcp-tool-opaque-fields'),
                    'include_query_params': operation.get('x-mcp-tool-include-query-params'),
                }
                tools.append(tool)

    # A tool name used to be an operationId, which OpenAPI already keeps unique.
    # `x-mcp-tool-name` gives that up, and a duplicate would be silent: a token's
    # `mcp:endpoints:<name>` resolves by first match, so which endpoint it reaches
    # would depend on the order of this file.
    names = [t['name'] for t in tools]
    duplicates = sorted({n for n in names if names.count(n) > 1})
    if duplicates:
        raise ValueError(f"duplicate MCP tool name(s): {', '.join(duplicates)}")

    return tools

def generate_typescript_code(tools: List[Dict[str, Any]], spec: Dict[str, Any], base_path: str = "") -> str:
    """Generate TypeScript code with MCP endpoint tools.

    This catalogue only feeds the frontend's MCP scope picker, which reads names and
    methods.
    """
    if not tools:
        return """// Auto-generated MCP tools from OpenAPI specification
// This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY

export interface EndpointTool {
    name: string;
    description: string;
    instructions: string;
    path: string;
    method: string;
    pathParamsSchema?: object;
    queryParamsSchema?: object;
    bodySchema?: object;
}

export const mcpEndpointTools: EndpointTool[] = [];
"""

    tool_definitions = []

    for tool in tools:
        tool_name = tool['name']
        description = tool['description'].replace('"', '\\"').replace('\n', '\\n')
        instructions = tool['instructions'].replace('"', '\\"').replace('\n', '\\n')
        path = tool['path']
        method = tool['method'].upper()

        # Generate separate schemas
        path_params_schema, query_params_schema, body_schema, query_field_renames, body_field_renames = extract_separate_schemas(
            tool['parameters'], tool['requestBody'], spec, tool['required_fields'], base_path,
            tool.get('include_fields'), tool.get('opaque_fields'), tool.get('include_query_params')
        )

        # Convert schemas to TypeScript - use 'as const' for better type inference
        path_params_ts = json.dumps(path_params_schema, indent=8) if path_params_schema else "undefined"
        query_params_ts = json.dumps(query_params_schema, indent=8) if query_params_schema else "undefined"
        body_schema_ts = json.dumps(body_schema, indent=8) if body_schema else "undefined"
        query_field_renames_ts = json.dumps(query_field_renames, indent=8) if query_field_renames else "undefined"
        body_field_renames_ts = json.dumps(body_field_renames, indent=8) if body_field_renames else "undefined"

        # Generate tool definition
        tool_def = f"""    {{
        name: "{tool_name}",
        description: "{description}",
        instructions: "{instructions}",
        path: "{path}",
        method: "{method}",
        pathParamsSchema: {path_params_ts},
        queryParamsSchema: {query_params_ts},
        bodySchema: {body_schema_ts},
        queryFieldRenames: {query_field_renames_ts},
        bodyFieldRenames: {body_field_renames_ts}
    }}"""
        tool_definitions.append(tool_def)

    # Combine everything
    tool_definitions_str = ",\n".join(tool_definitions)

    typescript_code = f"""// Auto-generated MCP tools from OpenAPI specification
// This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY

export interface EndpointTool {{
    name: string;
    description: string;
    instructions: string;
    path: string;
    method: string;
    pathParamsSchema?: object;
    queryParamsSchema?: object;
    bodySchema?: object;
    queryFieldRenames?: Record<string, string>;
    bodyFieldRenames?: Record<string, string>;
}}

export const mcpEndpointTools: EndpointTool[] = [
{tool_definitions_str}
];
"""

    return typescript_code

def generate_rust_code(tools: List[Dict[str, Any]], spec: Dict[str, Any], base_path: str = "") -> str:
    """Generate the complete Rust code with MCP tools."""
    if not tools:
        return f"""// No MCP tools found in the OpenAPI specification
{IMPORTS}
pub fn all_tools() -> Vec<EndpointTool> {{
    vec![]
}}
"""

    tool_definitions = []

    for tool in tools:
        tool_name = tool['name']
        description = tool['description']
        instructions = tool['instructions']
        path = tool['path']
        method = tool['method'].upper()

        # Generate separate schemas
        path_params_schema, query_params_schema, body_schema, query_field_renames, body_field_renames = extract_separate_schemas(
            tool['parameters'], tool['requestBody'], spec, tool['required_fields'], base_path,
            tool.get('include_fields'), tool.get('opaque_fields'), tool.get('include_query_params')
        )

        path_params_rust = schema_to_rust_value(path_params_schema)
        query_params_rust = schema_to_rust_value(query_params_schema)
        body_schema_rust = schema_to_rust_value(body_schema)
        query_field_renames_rust = schema_to_rust_value(query_field_renames if query_field_renames else None)
        body_field_renames_rust = schema_to_rust_value(body_field_renames if body_field_renames else None)

        # Generate tool definition
        tool_def = f"""    EndpointTool {{
        name: Cow::Borrowed("{tool_name}"),
        description: Cow::Borrowed("{description}"),
        instructions: Cow::Borrowed("{instructions}"),
        path: Cow::Borrowed("{path}"),
        method: Cow::Borrowed("{method}"),
        path_params_schema: {path_params_rust},
        query_params_schema: {query_params_rust},
        body_schema: {body_schema_rust},
        query_field_renames: {query_field_renames_rust},
        body_field_renames: {body_field_renames_rust},
    }}"""
        tool_definitions.append(tool_def)

    # Combine everything
    tool_definitions_str = ",\n".join(tool_definitions)

    rust_code = f"""// Auto-generated MCP tools from OpenAPI specification
// This file is generated by generate_mcp_tools.py - DO NOT EDIT MANUALLY
{IMPORTS}
pub fn all_tools() -> Vec<EndpointTool> {{
    vec![
{tool_definitions_str}
    ]
}}
"""

    return rust_code

def main():
    """Main function to parse OpenAPI and generate Rust and TypeScript code."""
    script_dir = Path(__file__).parent
    backend_dir = script_dir.parent
    project_dir = backend_dir.parent

    openapi_file = backend_dir / "windmill-api" / "openapi.yaml"
    rust_output_file = backend_dir / "windmill-api" / "src" / "mcp" / "auto_generated_endpoints.rs"
    ts_output_file = project_dir / "frontend" / "src" / "lib" / "mcpEndpointTools.ts"

    if not openapi_file.exists():
        print(f"OpenAPI file not found: {openapi_file}", file=sys.stderr)
        sys.exit(1)

    print(f"Loading OpenAPI specification from: {openapi_file}")
    spec = load_openapi_spec(str(openapi_file))

    print("Searching for endpoints with x-mcp-tool: true...")
    tools = find_mcp_tools(spec)

    if tools:
        print(f"Found {len(tools)} MCP tool(s):")
        for tool in tools:
            print(f"  - {tool['name']}: {tool['method']} {tool['path']}")
    else:
        print("No MCP tools found (no endpoints with x-mcp-tool: true)")

    # Generate and write Rust code
    print(f"Generating Rust code...")
    rust_code = generate_rust_code(tools, spec, str(openapi_file))

    print(f"Writing Rust code to: {rust_output_file}")
    rust_output_file.parent.mkdir(parents=True, exist_ok=True)
    with open(rust_output_file, 'w', encoding='utf-8') as f:
        f.write(rust_code)

    # Generate and write TypeScript code
    print(f"Generating TypeScript code...")
    typescript_code = generate_typescript_code(tools, spec, str(openapi_file))

    print(f"Writing TypeScript code to: {ts_output_file}")
    ts_output_file.parent.mkdir(parents=True, exist_ok=True)
    with open(ts_output_file, 'w', encoding='utf-8') as f:
        f.write(typescript_code)

    print("Done!")

if __name__ == "__main__":
    main()