run_server.py

import json
import os
import logging
import sys
from pathlib import Path
from dotenv import load_dotenv
from adapter import ToolRegistry, APIExecutor, MCPServer, BasicAuth, NoAuth
from adapter.mcp import MCPTool
from adapter.parsing import CanonicalEndpoint

# Load environment variables from .env file in the script's directory
script_dir = Path(__file__).parent.absolute()
load_dotenv(dotenv_path=script_dir / ".env")

# Configure logging to stderr (MCP uses stdout for JSON-RPC)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    stream=sys.stderr
)
logger = logging.getLogger(__name__)


def load_config():
    """Load credentials from environment variables."""
    username = os.getenv("DATAFORSEO_USERNAME")
    password = os.getenv("DATAFORSEO_PASSWORD")
    base_url = os.getenv("DATAFORSEO_BASE_URL", "https://api.dataforseo.com")

    if username and password:
        return {
            "username": username,
            "password": password,
            "base_url": base_url
        }

    return None


def load_registry(registry_file: str):
    """Load registry from JSON file."""
    with open(registry_file) as f:
        data = json.load(f)
    
    registry = ToolRegistry(name=data.get("name", "dataforseo_researcher_toolkit"))
    
    for tool_data in data.get("tools", []):
        tool = MCPTool(
            name=tool_data["name"],
            description=tool_data["description"],
            inputSchema=tool_data["inputSchema"],
            metadata=tool_data.get("metadata")
        )
        registry.add_tool(tool)
    
    return registry


def load_endpoints(endpoints_file: str):
    """Load endpoints from JSON file."""
    with open(endpoints_file) as f:
        data = json.load(f)
    
    endpoints = []
    for ep_data in data:
        endpoint = CanonicalEndpoint(**ep_data)
        endpoints.append(endpoint)
    
    return endpoints


def main():
    logger.info("=" * 70)
    logger.info("DataForSEO API MCP Server")
    logger.info("=" * 70)

    # Check for registry files (use absolute paths)
    registry_file = script_dir / "dataforseo_researcher_toolkit.json"
    endpoints_file = script_dir / "dataforseo_researcher_endpoints.json"

    if not registry_file.exists() or not endpoints_file.exists():
        logger.error("Registry files not found!")
        logger.error("Please run generate_registry.py first:")
        logger.error("  python generate_registry.py")
        return

    # Load credentials
    config = load_config()
    if config:
        logger.info("Credentials loaded")
        logger.info(f"  Base URL: {config['base_url']}")
    else:
        logger.error("No credentials found!")
        logger.error("  Create a .env file with:")
        logger.error("    DATAFORSEO_USERNAME=your_username")
        logger.error("    DATAFORSEO_PASSWORD=your_password")
        return

    # Load registry
    logger.info(f"1. Loading registry from {registry_file}...")
    registry = load_registry(str(registry_file))
    logger.info(f"Loaded {registry.count()} tools")

    # Load endpoints
    logger.info(f"2. Loading endpoints from {endpoints_file}...")
    endpoints = load_endpoints(str(endpoints_file))
    logger.info(f"Loaded {len(endpoints)} endpoints")

    # Set up executor with Basic Auth
    logger.info("3. Setting up API executor...")
    auth = BasicAuth(
        username=config["username"],
        password=config["password"]
    )
    executor = APIExecutor(
        base_url=config["base_url"],
        auth=auth,
        timeout=30,
        max_retries=3
    )
    logger.info("Authentication: Basic Auth")

    # Create MCP server
    logger.info("4. Creating MCP server...")
    server = MCPServer(
        name="DataForSEO API",
        version="1.0.0",
        tool_registry=registry,
        executor=executor,
        endpoints=endpoints
    )

    logger.info("=" * 70)
    logger.info("Server ready!")
    logger.info("=" * 70)
    logger.info("Configure in Claude Desktop:")
    logger.info('{')
    logger.info('  "mcpServers": {')
    logger.info('    "dataforseo": {')
    logger.info(f'      "command": "{sys.executable}",')
    logger.info(f'      "args": ["{Path(__file__).absolute()}"]')
    logger.info('    }')
    logger.info('  }')
    logger.info('}')
    logger.info("=" * 70)
    logger.info("Starting server...")

    # Run server
    server.run()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        logger.info("Server stopped by user")
    except Exception as e:
        logger.error(f"Error: {e}")
        import traceback
        traceback.print_exc()