MCP Server Usage Guide

This guide covers how to use the DNALLM MCP server, including API reference, client examples, and practical usage patterns.

Overview

The DNALLM MCP server provides DNA sequence analysis capabilities through the Model Context Protocol (MCP). It supports multiple transport protocols and offers both basic and streaming prediction modes.

Available Tools

Basic Prediction Tools

_dna_sequence_predict

Predict a single DNA sequence using a specific model.

Parameters: - sequence (string): DNA sequence to analyze (A, T, G, C only) - model_name (string): Name of the model to use

Returns: - Prediction results with confidence scores and labels

Example:

{
  "sequence": "ATCGATCGATCG",
  "model_name": "promoter_model",
  "result": {
    "prediction": "Core promoter",
    "confidence": 0.85,
    "probabilities": {
      "Not promoter": 0.15,
      "Core promoter": 0.85
    }
  }
}

_dna_batch_predict

Predict multiple DNA sequences using a single model.

Parameters: - sequences (array): List of DNA sequences to analyze - model_name (string): Name of the model to use

Returns: - Batch prediction results for all sequences

Example:

{
  "sequences": ["ATCGATCG", "GCTAGCTA", "TTAACCGG"],
  "model_name": "promoter_model",
  "results": [
    {
      "sequence": "ATCGATCG",
      "prediction": "Core promoter",
      "confidence": 0.85
    },
    {
      "sequence": "GCTAGCTA",
      "prediction": "Not promoter",
      "confidence": 0.72
    },
    {
      "sequence": "TTAACCGG",
      "prediction": "Core promoter",
      "confidence": 0.91
    }
  ]
}

_dna_multi_model_predict

Predict a single sequence using multiple models for comparison.

Parameters: - sequence (string): DNA sequence to analyze - model_names (array, optional): List of model names to use (uses all loaded models if not specified)

Returns: - Multi-model prediction results with per-model predictions (response keys: content, model_count, sequence; no consensus is computed)

Example:

{
  "sequence": "ATCGATCGATCG",
  "model_count": 2,
  "content": [
    {
      "type": "text",
      "text": "{'promoter_model': {'prediction': 'Core promoter', 'confidence': 0.85}, 'conservation_model': {'prediction': 'Conserved', 'confidence': 0.92}}"
    }
  ]
}

Streaming Tools (Real-time Progress)

_dna_stream_predict

Stream single sequence prediction with real-time progress updates.

Parameters: - sequence (string): DNA sequence to analyze - model_name (string): Name of the model to use - stream_progress (boolean, optional): Enable progress streaming (default: true)

Returns: - Streaming prediction results with progress updates

Progress Updates: - 0%: Starting prediction - 25%: Loading model and tokenizer - 75%: Processing prediction results - 100%: Prediction completed

_dna_stream_batch_predict

Stream batch prediction with progress updates.

Parameters: - sequences (array): List of DNA sequences to analyze - model_name (string): Name of the model to use - stream_progress (boolean, optional): Enable progress streaming (default: true)

Returns: - Streaming batch prediction results with per-sequence progress

_dna_stream_multi_model_predict

Stream multi-model prediction with progress updates.

Parameters: - sequence (string): DNA sequence to analyze - model_names (array, optional): List of model names to use - stream_progress (boolean, optional): Enable progress streaming (default: true)

Returns: - Streaming multi-model prediction results with per-model progress

Model Management Tools

_list_loaded_models

List all currently loaded models with their information.

Parameters: None

Returns: - List of loaded models with metadata

Example:

{
  "loaded_count": 1,
  "models": {
    "promoter_model": {
      "name": "promoter_model",
      "task_type": "binary",
      "num_labels": 2,
      "architecture": "DNABERT",
      "tokenizer": "BPE",
      "status": "loaded",
      "loaded": true
    }
  }
}

_get_model_info

Get detailed information about a specific model.

Parameters: - model_name (string): Name of the model

Returns: - Detailed model information including configuration and performance metrics

_list_models_by_task_type

List models filtered by task type.

Parameters: - task_type (string): Task type to filter by ("binary", "multiclass", "multilabel", "regression", "token", "mask", "generation", "classification")

Returns: - List of models matching the specified task type

_get_all_available_models

Get information about all available models (not just loaded ones).

Parameters: None

Returns: - Information about all models enabled in the server's mcp_server_config.yaml (not just loaded ones), returned as content, total_models, models

_health_check

Perform health check on the MCP server.

Parameters: None

Returns: - Server health status and statistics

Example:

from dnallm.mcp import DNALLMMCPClient

# Connect using streamable-http transport (recommended)
client = DNALLMMCPClient(transport="streamable-http", url="http://localhost:8000/mcp")

# Single sequence prediction
result = client.dna_sequence_predict("ATCGATCG", "dnabert-2")
print(result)

Mutation and Interpretation Tools

_dna_mutagenesis

Perform in silico mutagenesis on DNA sequences — evaluates the impact of sequence mutations (single/multi-base substitution, deletion, insertion, or exhaustive combinations) on model predictions.

Parameters: - model_name (string): Name of the model to use for prediction - sequence (string, optional): Single DNA sequence to mutate - sequences (array, optional): List of DNA sequences to mutate (either sequence or sequences must be provided) - mutation_type (string, optional): One of "single_base_substitution", "multi_base_substitution", "deletion", "insertion", "combo" (default: "single_base_substitution") - positions (array): 0-based positions to mutate (required, non-empty)

Returns: - Mutagenesis results containing original_prediction, mutated_prediction, delta, affected_positions, mutation_type, model_name (or error/isError on failure)

_dna_interpret

Interpret model predictions using attribution methods — computes attribution scores for each token in a DNA sequence using Captum methods.

Parameters: - sequence (string): DNA sequence to interpret (A, C, G, T, N only, case-insensitive) - model_name (string): Name of the model to use - method (string, optional): One of "lig", "deeplift", "occlusion", "feature_ablation", "layer_conductance", "gradient_shap", "noise_tunnel", "integrated_gradients" (default: "lig") - target_class (integer, optional): Target class index for attribution; auto-selects the class with maximum probability if not specified - max_length (integer, optional): Maximum token length for the tokenizer

Returns: - Interpretation results containing attributions, tokens, method, target_class, model_name, sequence (or error/isError on failure)

Python with Pydantic AI

import asyncio
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.ollama import OllamaProvider
from pydantic_ai.mcp import MCPServerStreamableHTTP

# Create MCP server connection
server = MCPServerStreamableHTTP("http://localhost:8000/mcp")

# Create agent with MCP server tools
agent = Agent(
    OpenAIChatModel(
        model_name="qwen3:latest",
        provider=OllamaProvider(base_url="http://localhost:11434/v1"),
    ),
    toolsets=[server],
    system_prompt="""You are a DNA analysis assistant with access to specialized DNA analysis tools via MCP server.

When analyzing a DNA sequence, you should:
1. First call _list_loaded_models to see what models are available
2. Then call _dna_multi_model_predict with the DNA sequence and appropriate model names
3. Interpret and explain the results in a comprehensive way

Available tools should include:
- _list_loaded_models: Lists available DNA analysis models
- _dna_multi_model_predict: Predicts DNA sequence properties using multiple models

Always use the tools to provide accurate analysis.""",
)


# Analyze DNA sequence
async def analyze_dna_sequence():
    async with agent:
        result = await agent.run(
            "What is the function of following DNA sequence? Please analyze it thoroughly using all available models: AGAAAAAACATGACAAGAAATCGATAATAATACAAAAGCTATGATGGTGTGCAATGTCCGTGTGCATGCGTGCACGCATTGCAACCGGCCCAAATCAAGGCCCATCGATCAGTGAATACTCATGGGCCGGCGGCCCACCACCGCTTCATCTCCTCCTCCGACGACGGGAGCACCCCCGCCGCATCGCCACCGACGAGGAGGAGGCCATTGCCGGCGGCGCCCCCGGTGAGCCGCTGCACCACGTCCCTGA"
        )
        return result


# Run the analysis
result = await analyze_dna_sequence()
print(result.output)

Python with MCP Client

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    # Connect to MCP server via STDIO
    server_params = StdioServerParameters(command="dnallm-mcp-server")

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the session
            await session.initialize()

            # List available models
            models = await session.call_tool("_list_loaded_models", {})
            print(f"Available models: {models}")

            # Predict DNA sequence
            result = await session.call_tool(
                "_dna_sequence_predict",
                {"sequence": "ATCGATCGATCG", "model_name": "promoter_model"},
            )
            print(f"Prediction result: {result}")

            # Multi-model prediction
            multi_result = await session.call_tool(
                "_dna_multi_model_predict",
                {
                    "sequence": "ATCGATCGATCG",
                    "model_names": ["promoter_model", "conservation_model"],
                },
            )
            print(f"Multi-model result: {multi_result}")


if __name__ == "__main__":
    asyncio.run(main())

Python with SSE Client

import asyncio
from mcp.client.sse import sse_client


async def main():
    # Connect to SSE server
    async with sse_client("http://localhost:8000/sse") as (read, write):
        # List available models
        models = await read.call_tool("_list_loaded_models", {})
        print(f"Available models: {models}")

        # Test streaming prediction
        result = await read.call_tool(
            "_dna_stream_predict",
            {
                "sequence": "ATCGATCGATCG",
                "model_name": "promoter_model",
                "stream_progress": True,
            },
        )
        print(f"Streaming prediction result: {result}")


if __name__ == "__main__":
    asyncio.run(main())

JavaScript/Node.js Client

const { EventSource } = require('eventsource');

// Connect to SSE server
const eventSource = new EventSource('http://localhost:8000/sse');

eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('Received:', data);
};

// Send MCP tool call
async function callTool(toolName, arguments) {
    const response = await fetch('http://localhost:8000/mcp/messages/?session_id=test', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
        },
        body: JSON.stringify({
            jsonrpc: "2.0",
            id: 1,
            method: "tools/call",
            params: {
                name: toolName,
                arguments: arguments
            }
        })
    });

    return await response.json();
}

// Example usage
callTool("_list_loaded_models", {})
    .then(result => console.log("Models:", result))
    .catch(error => console.error("Error:", error));

callTool("_dna_sequence_predict", {
    sequence: "ATCGATCGATCG",
    model_name: "promoter_model"
})
    .then(result => console.log("Prediction:", result))
    .catch(error => console.error("Error:", error));

cURL Examples

Health Check

curl -X POST "http://localhost:8000/mcp/messages/?session_id=test" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "_health_check", "arguments": {}}}'

List Models

curl -X POST "http://localhost:8000/mcp/messages/?session_id=test" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "_list_loaded_models", "arguments": {}}}'

Single Sequence Prediction

curl -X POST "http://localhost:8000/mcp/messages/?session_id=test" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "_dna_sequence_predict", "arguments": {"sequence": "ATCGATCGATCG", "model_name": "promoter_model"}}}'

Batch Prediction

curl -X POST "http://localhost:8000/mcp/messages/?session_id=test" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "_dna_batch_predict", "arguments": {"sequences": ["ATCGATCG", "GCTAGCTA"], "model_name": "promoter_model"}}}'

Multi-Model Prediction

async def basic_dna_analysis(sequence):
    async with sse_client("http://localhost:8000/sse") as (read, write):
        # 1. Check available models
        models = await read.call_tool("_list_loaded_models", {})
        print(f"Available models: {models}")

        # 2. Single model prediction
        result = await read.call_tool(
            "_dna_sequence_predict",
            {"sequence": sequence, "model_name": "promoter_model"},
        )
        print(f"Promoter prediction: {result}")

        # 3. Multi-model analysis
        multi_result = await read.call_tool("_dna_multi_model_predict", {"sequence": sequence})
        print(f"Multi-model analysis: {multi_result}")

        return multi_result

2. Batch Processing Workflow

async def batch_dna_analysis(sequences):
    async with sse_client("http://localhost:8000/sse") as (read, write):
        # Process sequences in batches
        batch_size = 10
        results = []

        for i in range(0, len(sequences), batch_size):
            batch = sequences[i : i + batch_size]

            # Use streaming for progress updates
            result = await read.call_tool(
                "_dna_stream_batch_predict",
                {
                    "sequences": batch,
                    "model_name": "promoter_model",
                    "stream_progress": True,
                },
            )
            results.extend(result.get("results", []))

        return results

3. Real-time Analysis with Progress

async def real_time_analysis(sequence):
    async with sse_client("http://localhost:8000/sse") as (read, write):
        # Use streaming prediction for real-time updates
        result = await read.call_tool(
            "_dna_stream_predict",
            {
                "sequence": sequence,
                "model_name": "promoter_model",
                "stream_progress": True,
            },
        )

        # Process streaming results
        if result.get("streamed"):
            print("Prediction completed with progress updates")

        return result

4. Model Comparison Workflow

async def model_comparison(sequence):
    async with sse_client("http://localhost:8000/sse") as (read, write):
        # Get all loaded models
        models = await read.call_tool("_list_loaded_models", {})
        model_names = list(models["models"].keys())

        # Compare all models
        comparison = await read.call_tool(
            "_dna_multi_model_predict",
            {"sequence": sequence, "model_names": model_names},
        )

        # Analyze per-model results
        results = comparison.get("results", {})

        print(f"Model comparison for sequence: {sequence}")
        for model_name, result in results.items():
            print(f"{model_name}: {result['prediction']} (confidence: {result['confidence']:.3f})")

        return comparison

Error Handling

Common Error Responses

{
  "error": "Model promoter_model not available or prediction failed",
  "isError": true
}

{
  "error": "Sequence contains invalid characters. Only A, C, G, T, N (case-insensitive) are allowed.",
  "isError": true
}

{
  "error": "Model unknown_model not found",
  "isError": true
}

Error Handling in Python

async def safe_prediction(sequence, model_name):
    try:
        async with sse_client("http://localhost:8000/sse") as (read, write):
            result = await read.call_tool(
                "_dna_sequence_predict",
                {"sequence": sequence, "model_name": model_name},
            )

            if result.get("isError"):
                print(f"Prediction error: {result.get('error')}")
                return None

            return result

    except Exception as e:
        print(f"Connection error: {e}")
        return None

Performance Tips

1. Batch Processing

• Use _dna_batch_predict for multiple sequences
• Process sequences in batches of 10-50 for optimal performance
• Use streaming for progress updates on large batches

2. Model Selection

• Use faster models (DNAMamba) for real-time applications
• Use more accurate models (DNABERT) for critical analysis
• Consider model-specific optimizations

3. Memory Management

• Monitor memory usage with large batches
• Use appropriate batch sizes for your hardware
• Consider CPU inference for memory-constrained environments

4. Network Optimization

• Use local models when possible
• Enable compression for SSE connections
• Use appropriate connection timeouts

Next Steps

• Configuration Guide for advanced setup
• Troubleshooting for common issues
• API Reference for detailed documentation