"""
Main FastMCP OpenAPI server implementation.
"""
import asyncio
import json
import logging
import sys
from typing import Dict, List, Optional, Any
import click
# Import MCP framework components
from mcp.server.fastmcp import FastMCP
from .config import AuthConfig, ServerConfig
from .models import OpenAPISpec
from .parser import OpenAPIParser
from .client import APIClient
from .tool_generator import ToolGenerator
def setup_logging(debug: bool = False):
"""Set up logging for the server."""
level = logging.DEBUG if debug else logging.INFO
logging.basicConfig(
level=level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
# Set specific loggers
logging.getLogger('fastmcp_openapi.client').setLevel(level)
logging.getLogger('fastmcp_openapi.server').setLevel(level)
class OpenAPIServer:
"""FastMCP server that generates tools from OpenAPI specifications."""
def __init__(self, name: str = "OpenAPI Server", debug: bool = False, port: int = 8000):
self.fastmcp = FastMCP(name=name, debug=debug, port=port)
# Storage for loaded APIs
self.apis: Dict[str, OpenAPISpec] = {}
self.api_clients: Dict[str, APIClient] = {}
self.tool_generators: Dict[str, ToolGenerator] = {}
self.parser = OpenAPIParser()
self.logger = logging.getLogger(__name__)
async def add_openapi_spec(
self,
name: str,
spec_url: str,
base_url: Optional[str] = None,
auth: Optional[AuthConfig] = None,
headers: Optional[Dict[str, str]] = None
) -> None:
"""Add an OpenAPI specification and generate tools for it."""
try:
self.logger.info(f"Loading OpenAPI spec for '{name}' from: {spec_url}")
# Load and parse the specification
spec_data = await self.parser.load_spec(spec_url)
openapi_spec = self.parser.parse_spec(spec_data)
# Determine base URL
if not base_url:
base_url = openapi_spec.get_base_url()
if not base_url:
raise ValueError(f"No base URL found for API '{name}'. Please provide one.")
# Create API client
api_client = APIClient(base_url=base_url, auth=auth, headers=headers)
# Create tool generator
tool_generator = ToolGenerator(api_client)
# Store everything
self.apis[name] = openapi_spec
self.api_clients[name] = api_client
self.tool_generators[name] = tool_generator
# Generate and register tools
self._register_tools_for_api(name, openapi_spec, tool_generator)
self.logger.info(f"Successfully loaded API '{name}' with {len(openapi_spec.operations)} operations")
except Exception as e:
self.logger.error(f"Failed to load OpenAPI spec '{name}': {e}")
raise
def _register_tools_for_api(
self,
api_name: str,
spec: OpenAPISpec,
tool_generator: ToolGenerator
) -> None:
"""Register MCP tools for all operations in an API."""
for operation in spec.operations:
# Generate tool function
tool_func = tool_generator.generate_tool_function(operation)
# Create a unique tool name (prefix with API name)
tool_name = f"{api_name}_{operation.operation_id}"
# Register with FastMCP
description = operation.summary or operation.description or f"API operation: {operation.operation_id}"
self.fastmcp.tool(
name=tool_name,
description=description
)(tool_func)
def run(self, transport: str = "stdio") -> None:
"""Run the FastMCP server."""
self.logger.info(f"Starting server with {len(self.apis)} APIs loaded")
self.fastmcp.run(transport=transport)
@click.command()
@click.option("--spec", multiple=True, help="OpenAPI specification URL or file path")
@click.option("--name", default="OpenAPI Server", help="Server name")
@click.option("--auth-header", multiple=True, help="Authorization header value (e.g., 'Bearer token123'). Must match order of --spec options.")
@click.option("--base-url", multiple=True, help="Override base URL for API calls. Must match order of --spec options.")
@click.option("--config", help="JSON config file with API specifications: {'apis': [{'spec': 'url', 'base_url': 'url', 'auth': 'header'}]}")
@click.option("--transport", type=click.Choice(["stdio", "streamable-http", "sse"]), default="stdio", help="Transport type")
@click.option("--port", type=int, default=8000, help="Port for HTTP transports")
@click.option("--debug", is_flag=True, help="Enable debug logging")
def main(
spec: List[str],
name: str,
auth_header: List[str],
base_url: List[str],
config: Optional[str],
transport: str,
port: int,
debug: bool
) -> None:
"""FastMCP OpenAPI server - Generate MCP tools from OpenAPI specifications."""
setup_logging(debug=debug)
# Validate that either --spec or --config is provided
if not config and not spec:
click.echo("Error: Must provide either --spec or --config option")
return
if config and spec:
click.echo("Error: Cannot use both --spec and --config options together")
return
# Create server
server = OpenAPIServer(name=name, debug=debug, port=port)
# Handle JSON config file if provided
if config:
try:
with open(config, 'r') as f:
config_data = json.load(f)
click.echo(f"Loading configuration from: {config}")
async def setup_from_config():
for i, api_config in enumerate(config_data.get('apis', [])):
api_name = api_config.get('name', f"api_{i + 1}")
spec_url = api_config['spec']
spec_base_url = api_config.get('base_url')
auth_value = api_config.get('auth')
# Parse auth
auth = None
if auth_value:
if auth_value.startswith("Bearer "):
auth = AuthConfig(type="bearer", token=auth_value[7:])
elif auth_value.startswith("Basic "):
auth = AuthConfig(type="basic", token=auth_value[6:])
else:
auth = AuthConfig(type="apikey", header_name="Authorization", token=auth_value)
try:
click.echo(f"Loading API '{api_name}' from: {spec_url}")
if spec_base_url:
click.echo(f" Using base URL: {spec_base_url}")
if auth:
click.echo(f" Using authentication: {auth.type}")
await server.add_openapi_spec(
name=api_name,
spec_url=spec_url,
base_url=spec_base_url,
auth=auth
)
except Exception as e:
click.echo(f"Error loading API {api_name}: {e}")
return
asyncio.run(setup_from_config())
except Exception as e:
click.echo(f"Error loading config file: {e}")
return
else:
# Handle command line arguments (positional matching)
async def setup_server():
"""Set up the server with provided specifications."""
for i, spec_url in enumerate(spec):
# Generate API name
api_name = "api" if len(spec) == 1 else f"api_{i + 1}"
# Get base URL for this spec (if provided)
spec_base_url = base_url[i] if i < len(base_url) else None
# Get auth header for this spec (if provided)
spec_auth_header = auth_header[i] if i < len(auth_header) else None
# Parse auth header for this spec
auth = None
if spec_auth_header:
if spec_auth_header.startswith("Bearer "):
auth = AuthConfig(type="bearer", token=spec_auth_header[7:])
elif spec_auth_header.startswith("Basic "):
auth = AuthConfig(type="basic", token=spec_auth_header[6:])
else:
auth = AuthConfig(type="apikey", header_name="Authorization", token=spec_auth_header)
try:
click.echo(f"Loading OpenAPI spec for '{api_name}' from: {spec_url}")
if spec_base_url:
click.echo(f" Using base URL: {spec_base_url}")
if auth:
click.echo(f" Using authentication: {auth.type}")
await server.add_openapi_spec(
name=api_name,
spec_url=spec_url,
base_url=spec_base_url,
auth=auth
)
except Exception as e:
click.echo(f"Error loading spec {spec_url}: {e}")
return
# Run setup
asyncio.run(setup_server())
# Start server
server.run(transport=transport)
if __name__ == "__main__":
main()
