mcp/server API

dnallm.mcp.server

DNALLM MCP Server Implementation.

This module implements the main MCP (Model Context Protocol) server using the FastMCP framework with Server-Sent Events (SSE) support for real-time DNA sequence prediction.

The server provides a comprehensive set of tools for DNA sequence analysis, including: - Single sequence prediction with specific models - Batch processing of multiple sequences - Multi-model prediction and comparison - Real-time streaming predictions with progress updates - Model management and health monitoring

Architecture

The server is built on top of the FastMCP framework, which provides MCP protocol implementation with multiple transport options (stdio, SSE, HTTP). The server manages DNA language models through a ModelManager and handles configuration through a ConfigManager.

Transport Protocols

• stdio: Standard input/output for CLI tools
• sse: Server-Sent Events for real-time web applications
• streamable-http: HTTP-based streaming protocol

Example

Basic server initialization:

server = DNALLMMCPServer("config/server_config.yaml")
await server.initialize()
server.start_server(host="127.0.0.1", port=8000, transport="sse")

Note

This server requires proper configuration files and model setup before initialization. See the configuration documentation for details.

Classes

DNALLMMCPServer

DNALLMMCPServer(config_path)

DNALLM MCP Server implementation using FastMCP framework with SSE support.

This class provides a comprehensive MCP server for DNA language model inference and analysis. It supports multiple transport protocols and provides real-time streaming capabilities for DNA sequence prediction tasks.

The server manages multiple DNA language models and provides various prediction modes including single sequence prediction, batch processing, and multi-model comparison. All operations support progress reporting through SSE for real-time user feedback.

Attributes:

Name	Type	Description
config_path	str	Path to the main server configuration file
config_manager	MCPConfigManager	Handles configuration management
model_manager	ModelManager	Manages model loading and prediction
app	FastMCP | None	Main FastMCP application instance
sse_app		SSE application instance (unused, FastMCP handles SSE internally)
_initialized	bool	Server initialization status flag

Example

Initialize and start the server:

# Create server instance
server = DNALLMMCPServer("config/mcp_server_config.yaml")

# Initialize asynchronously
await server.initialize()

# Start with SSE transport
server.start_server(host="0.0.0.0", port=8000,
                   transport="sse")

Note

The server must be initialized before starting. Configuration files must contain valid model and server settings.

Initialize the MCP server instance.

Sets up the server with configuration and model managers, but does not load models or start the server. Call initialize() and start_server() separately for complete setup.

Parameters:

Name	Type	Description	Default
config_path	str	Absolute or relative path to the main MCP server configuration file. This file should contain server settings, model configurations, and transport options.	required

Raises:

Type	Description
FileNotFoundError	If the configuration file doesn't exist
ConfigurationError	If the configuration file is invalid

Example

server = DNALLMMCPServer("/path/to/config.yaml")

Note

The configuration directory and filename are extracted separately to support the MCPConfigManager's directory-based configuration loading strategy.

Source code in dnallm/mcp/server.py

def __init__(self, config_path: str) -> None:
    """Initialize the MCP server instance.

    Sets up the server with configuration and model managers, but does not
    load models or start the server. Call initialize() and start_server()
    separately for complete setup.

    Args:
        config_path (str): Absolute or relative path to the main MCP
            server configuration file. This file should contain server
            settings, model configurations, and transport options.

    Raises:
        FileNotFoundError: If the configuration file doesn't exist
        ConfigurationError: If the configuration file is invalid

    Example:
        ```python
        server = DNALLMMCPServer("/path/to/config.yaml")
        ```

    Note:
        The configuration directory and filename are extracted
        separately to support the MCPConfigManager's directory-based
        configuration loading strategy.
    """
    self.config_path = config_path
    # Extract directory and filename from config file path for
    # ConfigManager. MCPConfigManager requires separate directory and
    # filename parameters
    config_path_obj = Path(config_path)
    config_dir = config_path_obj.parent
    config_filename = config_path_obj.name
    # Initialize core components
    self.config_manager = MCPConfigManager(
        str(config_dir), config_filename
    )
    self.model_manager = ModelManager(self.config_manager)

    # FastMCP application instances
    self.app: FastMCP | None = None  # Main MCP application
    self.sse_app = None  # Not used - FastMCP handles SSE internally

    # Server state tracking
    self._initialized = False  # Prevents double initialization

Functions

get_server_info

get_server_info()

Get server information.

Source code in dnallm/mcp/server.py

def get_server_info(self) -> dict[str, Any]:
    """Get server information."""
    server_config = self.config_manager.get_server_config()
    if not server_config:
        return {"error": "Server configuration not loaded"}

    return {
        "name": server_config.mcp.name,
        "version": server_config.mcp.version,
        "description": server_config.mcp.description,
        "host": server_config.server.host,
        "port": server_config.server.port,
        "loaded_models": self.model_manager.get_loaded_models(),
        "enabled_models": self.config_manager.get_enabled_models(),
        "initialized": self._initialized,
    }

initialize async

initialize()

Initialize the server and load all enabled models.

This method performs the complete server initialization process: 1. Checks if already initialized (idempotent operation) 2. Loads and validates server configuration 3. Creates the FastMCP application instance 4. Registers all MCP tools 5. Loads all enabled DNA language models

The initialization is asynchronous because model loading can be time-consuming, especially for large transformer models.

Raises:

Type	Description
RuntimeError	If server configuration cannot be loaded
ModelLoadError	If critical models fail to load
ConfigurationError	If configuration is invalid

Example

server = DNALLMMCPServer("config.yaml")
await server.initialize()  # Required before starting

Note

This method is idempotent - calling it multiple times has no additional effect after the first successful initialization.

Source code in dnallm/mcp/server.py

async def initialize(self) -> None:
    """Initialize the server and load all enabled models.

    This method performs the complete server initialization process:
    1. Checks if already initialized (idempotent operation)
    2. Loads and validates server configuration
    3. Creates the FastMCP application instance
    4. Registers all MCP tools
    5. Loads all enabled DNA language models

    The initialization is asynchronous because model loading can be
    time-consuming, especially for large transformer models.

    Raises:
        RuntimeError: If server configuration cannot be loaded
        ModelLoadError: If critical models fail to load
        ConfigurationError: If configuration is invalid

    Example:
        ```python
        server = DNALLMMCPServer("config.yaml")
        await server.initialize()  # Required before starting
        ```

    Note:
        This method is idempotent - calling it multiple times has no
        additional effect after the first successful initialization.
    """
    # Check for duplicate initialization
    if self._initialized:
        logger.info("Server already initialized")
        return

    logger.info("Initializing DNALLM MCP Server...")

    # Load and validate server configuration
    server_config = self.config_manager.get_server_config()
    if not server_config:
        raise RuntimeError("Failed to load server configuration")

    # Create FastMCP application with configuration
    self.app = FastMCP(
        name=server_config.mcp.name,
        instructions=server_config.mcp.description,
    )

    # Register all available MCP tools
    self._register_tools()

    # Load all enabled models asynchronously
    await self.model_manager.load_all_enabled_models()

    # SSE transport is built into FastMCP framework
    # No need for separate SSE application setup

    # Mark server as initialized
    self._initialized = True
    logger.info("DNALLM MCP Server initialized successfully")

shutdown async

shutdown()

Shutdown the server and cleanup resources.

Source code in dnallm/mcp/server.py

async def shutdown(self) -> None:
    """Shutdown the server and cleanup resources."""
    logger.info("Shutting down DNALLM MCP Server...")

    # Unload all models
    unloaded_count = self.model_manager.unload_all_models()
    logger.info(f"Unloaded {unloaded_count} models during shutdown")

    self._initialized = False
    logger.info("DNALLM MCP Server shutdown complete")

start_server

start_server(
    host="127.0.0.1", port=8000, transport="stdio"
)

Start the MCP server with the specified transport protocol.

This method starts the server using one of the supported transport protocols. The server must be initialized before calling this method. The transport protocol determines how the server communicates with clients:

• stdio: Standard input/output for CLI tools and automation
• sse: Server-Sent Events for real-time web applications
• streamable-http: HTTP-based streaming for REST API integration

Parameters:

Name	Type	Description	Default
host	str	Host address to bind the server to. Defaults to "127.0.0.1". Use "0.0.0.0" for all interfaces.	'127.0.0.1'
port	int	Port number to bind the server to. Defaults to 8000. Only used for HTTP-based transports.	8000
transport	str	Transport protocol to use. Choices: "stdio", "sse", "streamable-http". Defaults to "stdio".	'stdio'

Raises:

Type	Description
RuntimeError	If server is not initialized before starting
OSError	If port is already in use or host is invalid
ConfigurationError	If transport configuration is invalid

Example

# Start with SSE for real-time web apps
server.start_server(
    host="0.0.0.0",
    port=8000,
    transport="sse"
)

# Start with stdio for CLI tools
server.start_server(transport="stdio")

Note

This method is blocking and will run until the server is stopped. For SSE and HTTP transports, uvicorn handles graceful shutdown on SIGINT/SIGTERM signals.

Source code in dnallm/mcp/server.py

def start_server(
    self,
    host: str = "127.0.0.1",
    port: int = 8000,
    transport: str = "stdio",
) -> None:
    """Start the MCP server with the specified transport protocol.

    This method starts the server using one of the supported transport
    protocols.
    The server must be initialized before calling this method. The
    transport
    protocol determines how the server communicates with clients:

    - stdio: Standard input/output for CLI tools and automation
    - sse: Server-Sent Events for real-time web applications
    - streamable-http: HTTP-based streaming for REST API integration

    Args:
        host (str, optional): Host address to bind the server to.
            Defaults to "127.0.0.1". Use "0.0.0.0" for all interfaces.
        port (int, optional): Port number to bind the server to.
            Defaults to 8000. Only used for HTTP-based transports.
        transport (str, optional): Transport protocol to use.
            Choices: "stdio", "sse", "streamable-http".
            Defaults to "stdio".

    Raises:
        RuntimeError: If server is not initialized before starting
        OSError: If port is already in use or host is invalid
        ConfigurationError: If transport configuration is invalid

    Example:
        ```python
        # Start with SSE for real-time web apps
        server.start_server(
            host="0.0.0.0",
            port=8000,
            transport="sse"
        )

        # Start with stdio for CLI tools
        server.start_server(transport="stdio")
        ```

    Note:
        This method is blocking and will run until the server is stopped.
        For SSE and HTTP transports, uvicorn handles graceful shutdown
        on SIGINT/SIGTERM signals.
    """
    # Validate server initialization state
    if not self._initialized:
        raise RuntimeError(
            "Server not initialized. Call initialize() first."
        )

    # Override host/port from configuration if available
    server_config = self.config_manager.get_server_config()
    if server_config:
        host = server_config.server.host
        port = server_config.server.port

    logger.info(
        f"Starting DNALLM MCP Server on {host}:{port} with "
        f"{transport} transport"
    )

    # Dispatch to appropriate transport handler
    if transport == "sse":
        self._start_sse_server(host, port)
    elif transport == "streamable-http":
        self._start_http_server(host, port)
    else:
        # Default to stdio transport
        self._start_stdio_server()

Functions

initialize_mcp_server async

initialize_mcp_server(config_path)

Initialize the MCP server asynchronously.

This is a convenience function that creates and initializes a DNALLMMCPServer instance. It's designed to be called from asyncio.run() or other async contexts.

Parameters:

Name	Type	Description	Default
config_path	str	Path to the server configuration file	required

Returns:

Name	Type	Description
DNALLMMCPServer	DNALLMMCPServer	Fully initialized server instance ready to start

Raises:

Type	Description
ConfigurationError	If configuration is invalid
ModelLoadError	If model loading fails
RuntimeError	If initialization fails

Example

server = await initialize_mcp_server("config.yaml")
server.start_server(transport="sse")

Note

This function combines server creation and initialization for convenience in async entry points.

Source code in dnallm/mcp/server.py

async def initialize_mcp_server(config_path: str) -> DNALLMMCPServer:
    """Initialize the MCP server asynchronously.

    This is a convenience function that creates and initializes a
    DNALLMMCPServer instance. It's designed to be called from
    asyncio.run() or other async contexts.

    Args:
        config_path (str): Path to the server configuration file

    Returns:
        DNALLMMCPServer: Fully initialized server instance ready to start

    Raises:
        ConfigurationError: If configuration is invalid
        ModelLoadError: If model loading fails
        RuntimeError: If initialization fails

    Example:
        ```python
        server = await initialize_mcp_server("config.yaml")
        server.start_server(transport="sse")
        ```

    Note:
        This function combines server creation and initialization
        for convenience in async entry points.
    """
    server = DNALLMMCPServer(str(config_path))
    await server.initialize()
    return server

main

main()

Main entry point for the DNALLM MCP server CLI.

This function provides a command-line interface for starting the DNALLM MCP server with various configuration options. It handles argument parsing, configuration validation, server initialization, and graceful error handling.

The CLI supports multiple transport protocols and comprehensive configuration options for production deployment. It includes proper error handling and logging for troubleshooting.

Command Line Arguments

--config: Path to server configuration file --host: Host address to bind to (default: 0.0.0.0) --port: Port number to bind to (default: 8000) --transport: Protocol (stdio/sse/streamable-http, default: stdio) --log-level: Logging verbosity (DEBUG/INFO/WARNING/ERROR/CRITICAL) --version: Display version information

Example Usage

# Start with SSE transport
python server.py --config config.yaml --transport sse --port 8000

# Start with stdio (default)
python server.py --config config.yaml

# Start with debug logging
python server.py --config config.yaml --log-level DEBUG

Exit Codes

0: Successful execution 1: Configuration file not found or server error

Note

The server runs in blocking mode. Use Ctrl+C to stop gracefully. For SSE/HTTP transports, uvicorn handles signal processing automatically.

Source code in dnallm/mcp/server.py

def main():
    """Main entry point for the DNALLM MCP server CLI.

    This function provides a command-line interface for starting the DNALLM
    MCP server with various configuration options. It handles argument parsing,
    configuration validation, server initialization, and graceful error
    handling.

    The CLI supports multiple transport protocols and comprehensive
    configuration
    options for production deployment. It includes proper error handling and
    logging for troubleshooting.

    Command Line Arguments:
        --config: Path to server configuration file
        --host: Host address to bind to (default: 0.0.0.0)
        --port: Port number to bind to (default: 8000)
        --transport: Protocol (stdio/sse/streamable-http, default: stdio)
        --log-level: Logging verbosity (DEBUG/INFO/WARNING/ERROR/CRITICAL)
        --version: Display version information

    Example Usage:
        ```bash
        # Start with SSE transport
        python server.py --config config.yaml --transport sse --port 8000

        # Start with stdio (default)
        python server.py --config config.yaml

        # Start with debug logging
        python server.py --config config.yaml --log-level DEBUG
        ```

    Exit Codes:
        0: Successful execution
        1: Configuration file not found or server error

    Note:
        The server runs in blocking mode. Use Ctrl+C to stop gracefully.
        For SSE/HTTP transports, uvicorn handles signal processing
        automatically.
    """
    import asyncio
    import argparse
    import sys
    from pathlib import Path

    parser = argparse.ArgumentParser(
        description="Start the DNALLM MCP (Model Context Protocol) server",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  dnallm-mcp-server --config dnallm/mcp/configs/mcp_server_config.yaml
  dnallm-mcp-server --config dnallm/mcp/configs/mcp_server_config_2.yaml \
      --transport sse --port 8000
  dnallm-mcp-server --config dnallm/mcp/configs/mcp_server_config.yaml \
      --host 127.0.0.1 --port 9000
        """,
    )

    parser.add_argument(
        "--config",
        "-c",
        type=str,
        default="dnallm/mcp/configs/mcp_server_config.yaml",
        help="Path to MCP server configuration file (default: %(default)s)",
    )

    parser.add_argument(
        "--host",
        type=str,
        default="0.0.0.0",  # noqa: S104
        help="Host to bind the server to (default: %(default)s)",
    )

    parser.add_argument(
        "--port",
        type=int,
        default=8000,
        help="Port to bind the server to (default: %(default)s)",
    )

    parser.add_argument(
        "--log-level",
        type=str,
        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
        default="INFO",
        help="Logging level (default: %(default)s)",
    )

    parser.add_argument(
        "--transport",
        type=str,
        choices=["stdio", "sse", "streamable-http"],
        default="stdio",
        help="Transport protocol to use (default: %(default)s)",
    )

    parser.add_argument(
        "--version", action="version", version="DNALLM MCP Server 1.0.0"
    )

    args = parser.parse_args()

    # Check if config file exists
    config_path = Path(args.config)
    if not config_path.exists():
        logger.error(f"Configuration file not found: {config_path}")
        logger.error(
            "Please create a configuration file or specify the "
            "correct path with --config"
        )
        sys.exit(1)

    try:
        logger.info("Starting DNALLM MCP Server...")
        logger.info(f"Configuration: {config_path}")
        logger.info(f"Host: {args.host}")
        logger.info(f"Port: {args.port}")
        logger.info(f"Transport: {args.transport}")
        logger.info(f"Log Level: {args.log_level}")
        logger.info("-" * 50)

        # Initialize server in asyncio context
        server = asyncio.run(initialize_mcp_server(str(config_path)))

        # Get server info
        info = server.get_server_info()
        logger.info(f"Server initialized: {info['name']} v{info['version']}")
        logger.info(f"Loaded models: {info['loaded_models']}")
        logger.info(f"Enabled models: {info['enabled_models']}")
        logger.info("-" * 50)

        # Start server - let uvicorn handle signals for HTTP/SSE transports
        logger.info(
            f"Starting server on {args.host}:{args.port} with "
            f"{args.transport} transport"
        )
        logger.info("Press Ctrl+C to stop the server")

        # Start server (uvicorn will handle signals properly)
        server.start_server(
            host=args.host, port=args.port, transport=args.transport
        )

    except KeyboardInterrupt:
        logger.info("\nReceived interrupt signal, shutting down...")
    except Exception as e:
        logger.error(f"Server error: {e}")
        import traceback

        traceback.print_exc()
        sys.exit(1)
    finally:
        logger.info("Server stopped")