Tools API¶

This page provides comprehensive documentation for the DeepCritical tool system.

Tool Framework¶

ToolRunner¶

Abstract base class for all DeepCritical tools.

Key Methods: - run(parameters): Execute tool with given parameters - get_spec(): Get tool specification - validate_inputs(parameters): Validate input parameters

Attributes: - spec: Tool specification with metadata - category: Tool category for organization

ToolSpec¶

Defines tool metadata and interface specification.

Attributes: - name: Unique tool identifier - description: Human-readable description - category: Tool category (search, bioinformatics, etc.) - inputs: Input parameter specifications - outputs: Output specifications - metadata: Additional tool metadata

ToolRegistry¶

Central registry for tool management and execution.

Key Methods: - register_tool(spec, runner): Register a new tool - execute_tool(name, parameters): Execute tool by name - list_tools(): List all registered tools - get_tools_by_category(category): Get tools by category

Tool Categories¶

DeepCritical organizes tools into logical categories:

KNOWLEDGE_QUERY: Information retrieval tools
SEQUENCE_ANALYSIS: Bioinformatics sequence tools
STRUCTURE_PREDICTION: Protein structure tools
MOLECULAR_DOCKING: Drug-target interaction tools
DE_NOVO_DESIGN: Novel molecule design tools
FUNCTION_PREDICTION: Function annotation tools
RAG: Retrieval-augmented generation tools
SEARCH: Web and document search tools
ANALYTICS: Data analysis and visualization tools

Execution Framework¶

ExecutionResult¶

Results from tool execution.

Attributes: - success: Whether execution was successful - data: Main result data - metadata: Additional result metadata - execution_time: Time taken for execution - error: Error message if execution failed

ToolRequest¶

Request structure for tool execution.

Attributes: - tool_name: Name of tool to execute - parameters: Input parameters for the tool - metadata: Additional request metadata

ToolResponse¶

Response structure from tool execution.

Attributes: - success: Whether execution was successful - data: Tool output data - metadata: Response metadata - citations: Source citations if applicable

Domain Tools¶

Knowledge Query Tools¶

Web Search Tools¶

Bases: ToolRunner

Tool runner for web search operations.

Methods:

Name	Description
`run`	Execute web search operation.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/websearch_tools.py

def __init__(self):
    spec = ToolSpec(
        name="web_search",
        description="Search the web for information or fresh news, returning extracted content",
        inputs={"query": "TEXT", "search_type": "TEXT", "num_results": "INTEGER"},
        outputs={"content": "TEXT", "success": "BOOLEAN", "error": "TEXT"},
    )
    super().__init__(spec)

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Execute web search operation.

Source code in DeepResearch/src/tools/websearch_tools.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Execute web search operation."""
    try:
        # Validate inputs
        query = params.get("query", "")
        search_type = params.get("search_type", "search")
        num_results = params.get("num_results", 4)

        if not query:
            return ExecutionResult(
                success=False, error="Query parameter is required"
            )

        # Run async search
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            content = loop.run_until_complete(
                search_web(query, search_type, num_results)
            )
        finally:
            loop.close()

        # Check if search was successful
        success = not content.startswith("Error:")
        error = None if success else content

        return ExecutionResult(
            success=success,
            data={
                "content": content,
                "success": success,
                "error": error,
                "query": query,
                "search_type": search_type,
                "num_results": num_results,
            },
        )

    except Exception as e:
        return ExecutionResult(success=False, error=f"Web search failed: {e!s}")

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Bases: ToolRunner

Tool runner for chunked search operations.

Methods:

Name	Description
`run`	Execute chunked search operation.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/websearch_tools.py

def __init__(self):
    spec = ToolSpec(
        name="chunked_search",
        description="Search the web and return chunked content for RAG processing",
        inputs={
            "query": "TEXT",
            "search_type": "TEXT",
            "num_results": "INTEGER",
            "chunk_size": "INTEGER",
            "chunk_overlap": "INTEGER",
            "heading_level": "INTEGER",
            "min_characters_per_chunk": "INTEGER",
            "max_characters_per_section": "INTEGER",
            "clean_text": "BOOLEAN",
        },
        outputs={"chunks": "JSON", "success": "BOOLEAN", "error": "TEXT"},
    )
    super().__init__(spec)

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Execute chunked search operation.

Source code in DeepResearch/src/tools/websearch_tools.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Execute chunked search operation."""
    try:
        # Validate inputs
        query = params.get("query", "")
        search_type = params.get("search_type", "search")
        num_results = params.get("num_results", 4)
        chunk_size = params.get("chunk_size", 1000)
        chunk_overlap = params.get("chunk_overlap", 0)
        heading_level = params.get("heading_level", 3)
        min_characters_per_chunk = params.get("min_characters_per_chunk", 50)
        max_characters_per_section = params.get("max_characters_per_section", 4000)
        clean_text = params.get("clean_text", True)

        if not query:
            return ExecutionResult(
                success=False, error="Query parameter is required"
            )

        # Run async chunked search
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            chunks_json = loop.run_until_complete(
                search_and_chunk(
                    query=query,
                    search_type=search_type,
                    num_results=num_results,
                    tokenizer_or_token_counter="character",
                    chunk_size=chunk_size,
                    chunk_overlap=chunk_overlap,
                    heading_level=heading_level,
                    min_characters_per_chunk=min_characters_per_chunk,
                    max_characters_per_section=max_characters_per_section,
                    clean_text=clean_text,
                )
            )
        finally:
            loop.close()

        # Parse chunks
        try:
            chunks = json.loads(chunks_json)
            success = not (
                isinstance(chunks, list)
                and len(chunks) > 0
                and "error" in chunks[0]
            )
            error = None if success else chunks[0].get("error", "Unknown error")
        except json.JSONDecodeError:
            chunks = []
            success = False
            error = "Failed to parse chunks JSON"

        return ExecutionResult(
            success=success,
            data={
                "chunks": chunks,
                "success": success,
                "error": error,
                "query": query,
            },
        )

    except Exception as e:
        return ExecutionResult(success=False, error=f"Chunked search failed: {e!s}")

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Sequence Analysis Tools¶

Bioinformatics Tools¶

Bases: ToolRunner

Tool for processing GO annotations with PubMed context.

Methods:

Name	Description
`run`	Process GO annotations with PubMed context.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/bioinformatics_tools.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="go_annotation_processor",
            description="Process GO annotations with PubMed paper context for reasoning tasks",
            inputs={
                "annotations": "JSON",
                "papers": "JSON",
                "evidence_codes": "TEXT",
            },
            outputs={
                "processed_annotations": "JSON",
                "quality_score": "FLOAT",
                "annotation_count": "INTEGER",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Process GO annotations with PubMed context.

Source code in DeepResearch/src/tools/bioinformatics_tools.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Process GO annotations with PubMed context."""
    try:
        # Extract parameters
        annotations = params.get("annotations", [])
        papers = params.get("papers", [])
        evidence_codes = params.get("evidence_codes", "IDA,EXP").split(",")

        # Process annotations using deferred tool
        processed_annotations = go_annotation_processor(
            annotations, papers, evidence_codes
        )

        # Calculate quality score based on evidence codes
        quality_score = 0.9 if "IDA" in evidence_codes else 0.7

        return ExecutionResult(
            success=True,
            data={
                "processed_annotations": [
                    ann.model_dump() for ann in processed_annotations
                ],
                "quality_score": quality_score,
                "annotation_count": len(processed_annotations),
            },
            error=None,
        )

    except Exception as e:
        return ExecutionResult(
            success=False,
            data={},
            error=f"GO annotation processing failed: {e!s}",
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Bases: ToolRunner

Tool for retrieving PubMed papers.

Methods:

Name	Description
`run`	Retrieve PubMed papers.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/bioinformatics_tools.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="pubmed_retriever",
            description="Retrieve PubMed papers based on query with full text for open access papers",
            inputs={
                "query": "TEXT",
                "max_results": "INTEGER",
                "year_min": "INTEGER",
            },
            outputs={
                "papers": "JSON",
                "total_found": "INTEGER",
                "open_access_count": "INTEGER",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Retrieve PubMed papers.

Source code in DeepResearch/src/tools/bioinformatics_tools.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Retrieve PubMed papers."""
    try:
        # Extract parameters
        query = params.get("query", "")
        max_results = int(params.get("max_results", 100))
        year_min = params.get("year_min")

        if not query:
            return ExecutionResult(
                success=False,
                data={},
                error="No query provided for PubMed retrieval",
            )

        # Retrieve papers using deferred tool
        papers = pubmed_paper_retriever(query, max_results, year_min)

        # Count open access papers
        open_access_count = sum(1 for paper in papers if paper.is_open_access)

        return ExecutionResult(
            success=True,
            data={
                "papers": [paper.model_dump() for paper in papers],
                "total_found": len(papers),
                "open_access_count": open_access_count,
            },
            error=None,
        )

    except Exception as e:
        return ExecutionResult(
            success=False, data={}, error=f"PubMed retrieval failed: {e!s}"
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Deep Search Tools¶

Bases: ToolRunner

Main deep search tool that orchestrates the entire search process.

Methods:

Name	Description
`run`
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/deepsearch_tools.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="deep_search",
            description="Perform comprehensive deep search with multiple steps",
            inputs={"query": "TEXT", "max_steps": "NUMBER", "config": "TEXT"},
            outputs={"results": "TEXT", "search_history": "TEXT"},
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, str]) -> ExecutionResult

Source code in DeepResearch/src/tools/deepsearch_tools.py

def run(self, params: dict[str, str]) -> ExecutionResult:
    query = params.get("query", "")
    max_steps = int(params.get("max_steps", "10"))

    if not query:
        return ExecutionResult(success=False, error="No query provided")

    # Simulate deep search execution
    search_results = {
        "query": query,
        "steps_completed": min(max_steps, 5),  # Simulate some steps
        "results_found": 15,
        "final_answer": f"Deep search completed for query: {query}",
    }

    return ExecutionResult(
        success=True,
        data={
            "results": search_results,
            "search_history": f"Search history for: {query}",
        },
        metrics={"steps": max_steps, "results": 15},
    )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

RAG Tools¶

Bases: ToolRunner

Tool runner for RAG-compatible search operations.

Methods:

Name	Description
`run`	Execute RAG search operation.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/integrated_search_tools.py

def __init__(self):
    spec = ToolSpec(
        name="rag_search",
        description="Perform search optimized for RAG workflows with vector store integration",
        inputs={
            "query": "TEXT",
            "search_type": "TEXT",
            "num_results": "INTEGER",
            "chunk_size": "INTEGER",
            "chunk_overlap": "INTEGER",
        },
        outputs={
            "rag_query": "JSON",
            "documents": "JSON",
            "chunks": "JSON",
            "success": "BOOLEAN",
            "error": "TEXT",
        },
    )
    super().__init__(spec)

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Execute RAG search operation.

Source code in DeepResearch/src/tools/integrated_search_tools.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Execute RAG search operation."""
    try:
        # Extract parameters
        query = params.get("query", "")
        search_type = params.get("search_type", "search")
        num_results = params.get("num_results", 4)
        chunk_size = params.get("chunk_size", 1000)
        chunk_overlap = params.get("chunk_overlap", 0)

        if not query:
            return ExecutionResult(
                success=False, error="Query parameter is required"
            )

        # Create RAG query
        rag_query = RAGQuery(
            text=query,
            search_type=SearchType.SIMILARITY,
            top_k=num_results,
            filters={"search_type": search_type, "chunk_size": chunk_size},
        )

        # Use integrated search to get documents and chunks
        integrated_tool = IntegratedSearchTool()
        search_result = integrated_tool.run(
            {
                "query": query,
                "search_type": search_type,
                "num_results": num_results,
                "chunk_size": chunk_size,
                "chunk_overlap": chunk_overlap,
                "enable_analytics": True,
                "convert_to_rag": True,
            }
        )

        if not search_result.success:
            return ExecutionResult(
                success=False, error=f"RAG search failed: {search_result.error}"
            )

        return ExecutionResult(
            success=True,
            data={
                "rag_query": rag_query.model_dump(),
                "documents": search_result.data.get("documents", []),
                "chunks": search_result.data.get("chunks", []),
                "success": True,
                "error": None,
            },
        )

    except Exception as e:
        return ExecutionResult(success=False, error=f"RAG search failed: {e!s}")

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Code Execution Tools¶

Agent that generates code (bash commands or Python scripts) from natural language.

Initialize the code generation agent.

Parameters:

Name	Type	Description	Default
`model_name`	`str`	The model to use for code generation	`'anthropic:claude-sonnet-4-0'`
`max_retries`	`int`	Maximum number of generation retries	`3`
`timeout`	`float`	Timeout for generation	`60.0`

Methods:

Name	Description
`create_code_block`	Create a CodeBlock from generated code.
`generate_and_create_block`	Generate code and create a CodeBlock.
`generate_bash_command`	Generate a bash command from natural language description.
`generate_code`	Generate code from natural language description.
`generate_python_code`	Generate Python code from natural language description.

Attributes:

Name	Type	Description
`bash_agent`
`max_retries`
`model_name`
`python_agent`
`timeout`
`universal_agent`

Source code in DeepResearch/src/agents/code_generation_agent.py

def __init__(
    self,
    model_name: str = "anthropic:claude-sonnet-4-0",
    max_retries: int = 3,
    timeout: float = 60.0,
):
    """Initialize the code generation agent.

    Args:
        model_name: The model to use for code generation
        max_retries: Maximum number of generation retries
        timeout: Timeout for generation
    """
    self.model_name = model_name
    self.max_retries = max_retries
    self.timeout = timeout

    # Initialize Pydantic AI agents for different code types
    self.bash_agent = self._create_bash_agent()
    self.python_agent = self._create_python_agent()
    self.universal_agent = self._create_universal_agent()

Attributes¶

bash_agent `instance-attribute` ¶

bash_agent = _create_bash_agent()

max_retries `instance-attribute` ¶

max_retries = max_retries

model_name `instance-attribute` ¶

model_name = model_name

python_agent `instance-attribute` ¶

python_agent = _create_python_agent()

timeout `instance-attribute` ¶

timeout = timeout

universal_agent `instance-attribute` ¶

universal_agent = _create_universal_agent()

Functions¶

create_code_block ¶

create_code_block(code: str, language: str) -> CodeBlock

Create a CodeBlock from generated code.

Parameters:

Name	Type	Description	Default
`code`	`str`	The generated code	required
`language`	`str`	The language of the code	required

Returns:

Type	Description
`CodeBlock`	CodeBlock instance

Source code in DeepResearch/src/agents/code_generation_agent.py

def create_code_block(self, code: str, language: str) -> CodeBlock:
    """Create a CodeBlock from generated code.

    Args:
        code: The generated code
        language: The language of the code

    Returns:
        CodeBlock instance
    """
    return CodeBlock(code=code, language=language)

generate_and_create_block `async` ¶

generate_and_create_block(
    description: str, code_type: str | None = None
) -> tuple[str, CodeBlock]

Generate code and create a CodeBlock.

Parameters:

Name	Type	Description	Default
`description`	`str`	Natural language description	required
`code_type`	`str \| None`	Optional code type specification	`None`

Returns:

Type	Description
`tuple[str, CodeBlock]`	Tuple of (code_type, CodeBlock)

Source code in DeepResearch/src/agents/code_generation_agent.py

async def generate_and_create_block(
    self, description: str, code_type: str | None = None
) -> tuple[str, CodeBlock]:
    """Generate code and create a CodeBlock.

    Args:
        description: Natural language description
        code_type: Optional code type specification

    Returns:
        Tuple of (code_type, CodeBlock)
    """
    language, code = await self.generate_code(description, code_type)
    block = self.create_code_block(code, language)
    return language, block

generate_bash_command `async` ¶

generate_bash_command(description: str) -> str

Generate a bash command from natural language description.

Parameters:

Name	Type	Description	Default
`description`	`str`	Natural language description of the desired operation	required

Returns:

Type	Description
`str`	Generated bash command as string

Source code in DeepResearch/src/agents/code_generation_agent.py

async def generate_bash_command(self, description: str) -> str:
    """Generate a bash command from natural language description.

    Args:
        description: Natural language description of the desired operation

    Returns:
        Generated bash command as string
    """
    result = await self.bash_agent.run(
        f"Generate a bash command for: {description}"
    )
    return str(result.data).strip()

generate_code `async` ¶

generate_code(
    description: str, code_type: str | None = None
) -> tuple[str, str]

Generate code from natural language description.

Parameters:

Name	Type	Description	Default
`description`	`str`	Natural language description of the desired operation	required
`code_type`	`str \| None`	Type of code to generate ("bash", "python", or None for auto-detection)	`None`

Returns:

Type	Description
`tuple[str, str]`	Tuple of (code_type, generated_code)

Source code in DeepResearch/src/agents/code_generation_agent.py

async def generate_code(
    self, description: str, code_type: str | None = None
) -> tuple[str, str]:
    """Generate code from natural language description.

    Args:
        description: Natural language description of the desired operation
        code_type: Type of code to generate ("bash", "python", or None for auto-detection)

    Returns:
        Tuple of (code_type, generated_code)
    """
    if code_type == "bash":
        code = await self.generate_bash_command(description)
        return "bash", code
    if code_type == "python":
        code = await self.generate_python_code(description)
        return "python", code
    # Use universal agent to determine type and generate
    result = await self.universal_agent.run(
        f"Analyze and generate code for: {description}"
    )
    response = str(result.data).strip()

    # Parse response format: TYPE: [BASH|PYTHON]\nCODE: [code]
    lines = response.split("\n", 2)
    if len(lines) >= 2:
        type_line = lines[0]
        code_line = lines[1] if len(lines) > 1 else ""

        if type_line.startswith("TYPE:"):
            detected_type = type_line.split("TYPE:", 1)[1].strip().lower()
            if code_line.startswith("CODE:"):
                code = code_line.split("CODE:", 1)[1].strip()
                return detected_type, code

    # Fallback: try to infer from content
    if any(
        keyword in description.lower()
        for keyword in [
            "file",
            "directory",
            "list",
            "find",
            "copy",
            "move",
            "delete",
            "system",
        ]
    ):
        code = await self.generate_bash_command(description)
        return "bash", code
    code = await self.generate_python_code(description)
    return "python", code

generate_python_code `async` ¶

generate_python_code(description: str) -> str

Generate Python code from natural language description.

Parameters:

Name	Type	Description	Default
`description`	`str`	Natural language description of the desired operation	required

Returns:

Type	Description
`str`	Generated Python code as string

Source code in DeepResearch/src/agents/code_generation_agent.py

async def generate_python_code(self, description: str) -> str:
    """Generate Python code from natural language description.

    Args:
        description: Natural language description of the desired operation

    Returns:
        Generated Python code as string
    """
    result = await self.python_agent.run(f"Generate Python code for: {description}")
    return str(result.data).strip()

Agent that executes generated code using the AG2 execution framework.

Initialize the code execution agent.

Parameters:

Name	Type	Description	Default
`model_name`	`str`	Model for execution analysis	`'anthropic:claude-sonnet-4-0'`
`use_docker`	`bool`	Whether to use Docker for execution	`True`
`use_jupyter`	`bool`	Whether to use Jupyter for execution	`False`
`jupyter_config`	`dict[str, Any] \| None`	Jupyter connection configuration	`None`
`max_retries`	`int`	Maximum execution retries	`3`
`timeout`	`float`	Execution timeout	`60.0`

Methods:

Name	Description
`execute_code`	Execute code string directly.
`execute_code_block`	Execute a code block and return results.

Attributes:

Name	Type	Description
`docker_executor`
`jupyter_config`
`jupyter_executor`
`local_executor`
`max_retries`
`model_name`
`python_tool`
`timeout`
`use_docker`
`use_jupyter`

Source code in DeepResearch/src/agents/code_generation_agent.py

def __init__(
    self,
    model_name: str = "anthropic:claude-sonnet-4-0",
    use_docker: bool = True,
    use_jupyter: bool = False,
    jupyter_config: dict[str, Any] | None = None,
    max_retries: int = 3,
    timeout: float = 60.0,
):
    """Initialize the code execution agent.

    Args:
        model_name: Model for execution analysis
        use_docker: Whether to use Docker for execution
        use_jupyter: Whether to use Jupyter for execution
        jupyter_config: Jupyter connection configuration
        max_retries: Maximum execution retries
        timeout: Execution timeout
    """
    self.model_name = model_name
    self.use_docker = use_docker
    self.use_jupyter = use_jupyter
    self.jupyter_config = jupyter_config or {}
    self.max_retries = max_retries
    self.timeout = timeout

    # Import execution utilities
    from DeepResearch.src.utils.coding import (
        DockerCommandLineCodeExecutor,
        LocalCommandLineCodeExecutor,
    )
    from DeepResearch.src.utils.jupyter import JupyterCodeExecutor
    from DeepResearch.src.utils.python_code_execution import PythonCodeExecutionTool

    # Initialize executors
    self.docker_executor = (
        DockerCommandLineCodeExecutor(timeout=int(timeout)) if use_docker else None
    )

    self.local_executor = LocalCommandLineCodeExecutor(timeout=int(timeout))

    self.jupyter_executor = None
    if use_jupyter:
        from DeepResearch.src.utils.jupyter.base import JupyterConnectionInfo

        conn_info = JupyterConnectionInfo(**self.jupyter_config)
        self.jupyter_executor = JupyterCodeExecutor(conn_info)

    self.python_tool = PythonCodeExecutionTool(
        timeout=int(timeout), use_docker=use_docker
    )

Attributes¶

docker_executor `instance-attribute` ¶

docker_executor = (
    DockerCommandLineCodeExecutor(timeout=int(timeout))
    if use_docker
    else None
)

jupyter_config `instance-attribute` ¶

jupyter_config = jupyter_config or {}

jupyter_executor `instance-attribute` ¶

jupyter_executor = None

local_executor `instance-attribute` ¶

local_executor = LocalCommandLineCodeExecutor(
    timeout=int(timeout)
)

max_retries `instance-attribute` ¶

max_retries = max_retries

model_name `instance-attribute` ¶

model_name = model_name

python_tool `instance-attribute` ¶

python_tool = PythonCodeExecutionTool(
    timeout=int(timeout), use_docker=use_docker
)

timeout `instance-attribute` ¶

timeout = timeout

use_docker `instance-attribute` ¶

use_docker = use_docker

use_jupyter `instance-attribute` ¶

use_jupyter = use_jupyter

Functions¶

execute_code `async` ¶

execute_code(code: str, language: str) -> dict[str, Any]

Execute code string directly.

Parameters:

Name	Type	Description	Default
`code`	`str`	Code to execute	required
`language`	`str`	Language of the code	required

Returns:

Type	Description
`dict[str, Any]`	Dictionary with execution results

Source code in DeepResearch/src/agents/code_generation_agent.py

async def execute_code(self, code: str, language: str) -> dict[str, Any]:
    """Execute code string directly.

    Args:
        code: Code to execute
        language: Language of the code

    Returns:
        Dictionary with execution results
    """
    code_block = CodeBlock(code=code, language=language)
    return await self.execute_code_block(code_block)

execute_code_block `async` ¶

execute_code_block(code_block: CodeBlock) -> dict[str, Any]

Execute a code block and return results.

Parameters:

Name	Type	Description	Default
`code_block`	`CodeBlock`	CodeBlock to execute	required

Returns:

Type	Description
`dict[str, Any]`	Dictionary with execution results

Source code in DeepResearch/src/agents/code_generation_agent.py

async def execute_code_block(self, code_block: CodeBlock) -> dict[str, Any]:
    """Execute a code block and return results.

    Args:
        code_block: CodeBlock to execute

    Returns:
        Dictionary with execution results
    """
    executor = self._get_executor(code_block.language)

    try:
        if hasattr(executor, "run"):  # PythonCodeExecutionTool
            result = executor.run(
                {
                    "code": code_block.code,
                    "max_retries": self.max_retries,
                    "timeout": self.timeout,
                }
            )
            return {
                "success": result.success,
                "output": result.data.get("output", "") if result.success else "",
                "error": result.data.get("error", "") if not result.success else "",
                "exit_code": 0 if result.success else 1,
                "language": code_block.language,
                "executor": "python_tool"
                if code_block.language == "python"
                else "local",
            }
        # CodeExecutor interface
        result = executor.execute_code_blocks([code_block])
        return {
            "success": result.exit_code == 0,
            "output": result.output,
            "error": "" if result.exit_code == 0 else result.output,
            "exit_code": result.exit_code,
            "language": code_block.language,
            "executor": "jupyter"
            if self.use_jupyter
            else ("docker" if self.use_docker else "local"),
        }

    except Exception as e:
        return {
            "success": False,
            "output": "",
            "error": f"Execution failed: {e!s}",
            "exit_code": 1,
            "language": code_block.language,
            "executor": "unknown",
        }

Structure Prediction Tools¶

Molecular Docking Tools¶

De Novo Design Tools¶

Function Prediction Tools¶

RAG Tools¶

Search Tools¶

Analytics Tools¶

MCP Server Management Tools¶

Bases: ToolRunner

Tool for listing available MCP servers.

Methods:

Name	Description
`run`	List available MCP servers.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/mcp_server_management.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="mcp_server_list",
            description="List all available vendored MCP servers",
            inputs={
                "include_status": "BOOLEAN",
                "include_tools": "BOOLEAN",
            },
            outputs={
                "servers": "JSON",
                "count": "INTEGER",
                "success": "BOOLEAN",
                "error": "TEXT",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

List available MCP servers.

Source code in DeepResearch/src/tools/mcp_server_management.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """List available MCP servers."""
    try:
        include_status = params.get("include_status", True)
        include_tools = params.get("include_tools", True)

        servers = []
        for server_name, server_class in SERVER_IMPLEMENTATIONS.items():
            server_info = {
                "name": server_name,
                "type": getattr(server_class, "__name__", "Unknown"),
                "description": getattr(server_class, "__doc__", "").strip(),
            }

            if include_tools:
                try:
                    server_instance: MCPServerProtocol = server_class()  # type: ignore[assignment]
                    server_info["tools"] = server_instance.list_tools()
                except Exception as e:
                    server_info["tools"] = []
                    server_info["tools_error"] = str(e)

            if include_status:
                # Check if server is deployed
                try:
                    deployment = asyncio.run(
                        server_manager.get_server_status(server_name)
                    )
                    if deployment:
                        server_info["status"] = deployment.status
                        server_info["container_id"] = deployment.container_id
                    else:
                        server_info["status"] = "not_deployed"
                except Exception as e:
                    server_info["status"] = "unknown"
                    server_info["status_error"] = str(e)

            servers.append(server_info)

        return ExecutionResult(
            success=True,
            data={
                "servers": servers,
                "count": len(servers),
                "success": True,
                "error": None,
            },
        )

    except Exception as e:
        logger.error(f"Failed to list MCP servers: {e}")
        return ExecutionResult(
            success=False,
            error=f"Failed to list MCP servers: {e!s}",
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Bases: ToolRunner

Tool for deploying MCP servers using testcontainers.

Methods:

Name	Description
`run`	Deploy an MCP server.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/mcp_server_management.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="mcp_server_deploy",
            description="Deploy a vendored MCP server using testcontainers",
            inputs={
                "server_name": "TEXT",
                "server_type": "TEXT",
                "container_image": "TEXT",
                "environment_variables": "JSON",
                "volumes": "JSON",
                "ports": "JSON",
            },
            outputs={
                "deployment": "JSON",
                "container_id": "TEXT",
                "status": "TEXT",
                "success": "BOOLEAN",
                "error": "TEXT",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Deploy an MCP server.

Source code in DeepResearch/src/tools/mcp_server_management.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Deploy an MCP server."""
    try:
        server_name = params.get("server_name", "")
        if not server_name:
            return ExecutionResult(success=False, error="Server name is required")

        # Check if server implementation exists
        if server_name not in SERVER_IMPLEMENTATIONS:
            return ExecutionResult(
                success=False,
                error=f"Server '{server_name}' not found. Available servers: {', '.join(SERVER_IMPLEMENTATIONS.keys())}",
            )

        # Create server configuration
        server_config = MCPServerConfig(
            server_name=server_name,
            server_type=MCPServerType(params.get("server_type", "custom")),
            container_image=params.get("container_image", "python:3.11-slim"),
            environment_variables=params.get("environment_variables", {}),
            volumes=params.get("volumes", {}),
            ports=params.get("ports", {}),
        )

        # Convert to TestcontainersConfig
        testcontainers_config = TestcontainersConfig(
            image=server_config.container_image,
            working_directory=server_config.working_directory,
            auto_remove=server_config.auto_remove,
            network_disabled=server_config.network_disabled,
            privileged=server_config.privileged,
            environment_variables=server_config.environment_variables,
            volumes=server_config.volumes,
            ports=server_config.ports,
        )

        # Deploy server
        deployment = asyncio.run(
            server_manager.deploy_server(server_name, config=testcontainers_config)
        )

        return ExecutionResult(
            success=True,
            data={
                "deployment": deployment.model_dump(),
                "container_id": deployment.container_id or "",
                "status": deployment.status,
                "success": deployment.status == MCPServerStatus.RUNNING,
                "error": deployment.error_message or "",
            },
        )

    except Exception as e:
        logger.error(f"Failed to deploy MCP server: {e}")
        return ExecutionResult(
            success=False,
            error=f"Failed to deploy MCP server: {e!s}",
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Bases: ToolRunner

Tool for executing tools on deployed MCP servers.

Methods:

Name	Description
`run`	Execute a tool on an MCP server.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/mcp_server_management.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="mcp_server_execute",
            description="Execute a tool on a deployed MCP server",
            inputs={
                "server_name": "TEXT",
                "tool_name": "TEXT",
                "parameters": "JSON",
                "timeout": "INTEGER",
                "async_execution": "BOOLEAN",
            },
            outputs={
                "result": "JSON",
                "execution_time": "FLOAT",
                "success": "BOOLEAN",
                "error": "TEXT",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Execute a tool on an MCP server.

Source code in DeepResearch/src/tools/mcp_server_management.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Execute a tool on an MCP server."""
    try:
        server_name = params.get("server_name", "")
        tool_name = params.get("tool_name", "")
        parameters = params.get("parameters", {})
        timeout = params.get("timeout", 300)
        async_execution = params.get("async_execution", False)

        if not server_name:
            return ExecutionResult(success=False, error="Server name is required")

        if not tool_name:
            return ExecutionResult(success=False, error="Tool name is required")

        # Create execution request
        request = MCPToolExecutionRequest(
            server_name=server_name,
            tool_name=tool_name,
            parameters=parameters,
            timeout=timeout,
            async_execution=async_execution,
        )

        # Get server deployment
        deployment = asyncio.run(server_manager.get_server_status(server_name))
        if not deployment:
            return ExecutionResult(
                success=False, error=f"Server '{server_name}' not deployed"
            )

        if deployment.status != MCPServerStatus.RUNNING:
            return ExecutionResult(
                success=False,
                error=f"Server '{server_name}' is not running (status: {deployment.status})",
            )

        # Get server implementation
        server = SERVER_IMPLEMENTATIONS.get(server_name)
        if not server:
            return ExecutionResult(
                success=False,
                error=f"Server implementation for '{server_name}' not found",
            )

        # Execute tool
        if async_execution:
            result = asyncio.run(server().execute_tool_async(request))
        else:
            result = server().execute_tool(tool_name, **parameters)

        # Format result
        if hasattr(result, "model_dump"):
            result_data = result.model_dump()
        elif isinstance(result, dict):
            result_data = result
        else:
            result_data = {"result": str(result)}

        return ExecutionResult(
            success=True,
            data={
                "result": result_data,
                "execution_time": getattr(result, "execution_time", 0.0),
                "success": True,
                "error": None,
            },
        )

    except Exception as e:
        logger.error(f"Failed to execute MCP server tool: {e}")
        return ExecutionResult(
            success=False,
            error=f"Failed to execute MCP server tool: {e!s}",
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Bases: ToolRunner

Tool for checking MCP server deployment status.

Methods:

Name	Description
`run`	Check MCP server status.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/mcp_server_management.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="mcp_server_status",
            description="Check the status of deployed MCP servers",
            inputs={
                "server_name": "TEXT",
            },
            outputs={
                "status": "TEXT",
                "container_id": "TEXT",
                "deployment_info": "JSON",
                "success": "BOOLEAN",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Check MCP server status.

Source code in DeepResearch/src/tools/mcp_server_management.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Check MCP server status."""
    try:
        server_name = params.get("server_name", "")

        if server_name:
            # Check specific server
            deployment = asyncio.run(server_manager.get_server_status(server_name))
            if not deployment:
                return ExecutionResult(
                    success=False, error=f"Server '{server_name}' not deployed"
                )

            return ExecutionResult(
                success=True,
                data={
                    "status": deployment.status,
                    "container_id": deployment.container_id or "",
                    "deployment_info": deployment.model_dump(),
                    "success": True,
                },
            )
        # List all deployments
        deployments = asyncio.run(server_manager.list_servers())
        deployment_info = [d.model_dump() for d in deployments]

        return ExecutionResult(
            success=True,
            data={
                "status": "multiple",
                "deployments": deployment_info,
                "count": len(deployment_info),
                "success": True,
            },
        )

    except Exception as e:
        logger.error(f"Failed to check MCP server status: {e}")
        return ExecutionResult(
            success=False,
            error=f"Failed to check MCP server status: {e!s}",
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Bases: ToolRunner

Tool for stopping deployed MCP servers.

Methods:

Name	Description
`run`	Stop an MCP server.
`validate`

Attributes:

Name	Type	Description
`spec`	`ToolSpec`

Source code in DeepResearch/src/tools/mcp_server_management.py

def __init__(self):
    super().__init__(
        ToolSpec(
            name="mcp_server_stop",
            description="Stop a deployed MCP server",
            inputs={
                "server_name": "TEXT",
            },
            outputs={
                "success": "BOOLEAN",
                "message": "TEXT",
                "error": "TEXT",
            },
        )
    )

Attributes¶

spec `instance-attribute` ¶

spec: ToolSpec = spec

Functions¶

run ¶

run(params: dict[str, Any]) -> ExecutionResult

Stop an MCP server.

Source code in DeepResearch/src/tools/mcp_server_management.py

def run(self, params: dict[str, Any]) -> ExecutionResult:
    """Stop an MCP server."""
    try:
        server_name = params.get("server_name", "")
        if not server_name:
            return ExecutionResult(success=False, error="Server name is required")

        # Stop server
        success = asyncio.run(server_manager.stop_server(server_name))

        if success:
            return ExecutionResult(
                success=True,
                data={
                    "success": True,
                    "message": f"Server '{server_name}' stopped successfully",
                    "error": "",
                },
            )
        return ExecutionResult(
            success=False,
            error=f"Server '{server_name}' not found or already stopped",
        )

    except Exception as e:
        logger.error(f"Failed to stop MCP server: {e}")
        return ExecutionResult(
            success=False,
            error=f"Failed to stop MCP server: {e!s}",
        )

validate ¶

validate(params: dict[str, Any]) -> tuple[bool, str | None]

Source code in DeepResearch/src/tools/base.py

def validate(self, params: dict[str, Any]) -> tuple[bool, str | None]:
    for k, t in self.spec.inputs.items():
        if k not in params:
            return False, f"Missing required param: {k}"
        # basic type gate (string types only for placeholder)
        if t.endswith(("PATH", "ID")) or t in {"TEXT", "AA SEQUENCE"}:
            if not isinstance(params[k], str):
                return False, f"Invalid type for {k}: expected str for {t}"
    return True, None

Enhanced MCP Server Framework¶

DeepCritical implements a comprehensive MCP (Model Context Protocol) server framework that integrates Pydantic AI for enhanced tool execution and reasoning capabilities. This framework supports both patterns described in the Pydantic AI MCP documentation:

Agents acting as MCP clients: Pydantic AI agents can connect to MCP servers to use their tools for research workflows
Agents embedded within MCP servers: Pydantic AI agents are integrated within MCP servers for enhanced tool execution

Key Features¶

Pydantic AI Integration: All MCP servers include embedded Pydantic AI agents for reasoning and tool orchestration
Testcontainers Deployment: Isolated container deployment for secure, reproducible execution
Session Tracking: Tool call history and session management for debugging and optimization
Type Safety: Strongly-typed interfaces using Pydantic models
Error Handling: Comprehensive error handling with retry logic
Health Monitoring: Built-in health checks and resource management

Architecture¶

The enhanced MCP server framework consists of:

MCPServerBase: Base class providing Pydantic AI integration and testcontainers deployment
@mcp_tool decorator: Custom decorator that creates Pydantic AI-compatible tools
Session Management: MCPAgentSession for tracking tool calls and responses
Deployment Management: Testcontainers-based deployment with resource limits
Type System: Comprehensive Pydantic models for MCP operations

MCP Server Base Classes¶

MCPServerBase¶

Enhanced base class for MCP server implementations with Pydantic AI integration.

Key Features: - Pydantic AI agent integration for enhanced tool execution and reasoning - Testcontainers deployment support with resource management - Session tracking for tool call history and debugging - Async/await support for concurrent tool execution - Comprehensive error handling with retry logic - Health monitoring and automatic recovery - Type-safe interfaces using Pydantic models

Key Methods: - list_tools(): List all available tools on the server - get_tool_spec(tool_name): Get specification for a specific tool - execute_tool(tool_name, **kwargs): Execute a tool with parameters - execute_tool_async(request): Execute tool asynchronously with session tracking - deploy_with_testcontainers(): Deploy server using testcontainers - stop_with_testcontainers(): Stop server deployed with testcontainers - health_check(): Perform health check on deployed server - get_pydantic_ai_agent(): Get the embedded Pydantic AI agent - get_session_info(): Get session information and tool call history

Attributes: - name: Server name - server_type: Server type enum - config: Server configuration (MCPServerConfig) - tools: Dictionary of Pydantic AI Tool objects - pydantic_ai_agent: Embedded Pydantic AI agent for reasoning - session: MCPAgentSession for tracking interactions - container_id: Container ID when deployed with testcontainers

Available MCP Servers¶

DeepCritical includes 29 vendored MCP (Model Context Protocol) servers for common bioinformatics tools, deployed using testcontainers for isolated execution environments. The servers are built using Pydantic AI patterns and provide strongly-typed interfaces.

Quality Control & Preprocessing (7 servers)¶

FastQC Server¶

::: DeepResearch.src.tools.bioinformatics.fastqc_server.FastQCServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.fastqc_server import FastQCServer

FastQC is a quality control tool for high throughput sequence data. This MCP server provides strongly-typed access to FastQC functionality with Pydantic AI integration for enhanced quality control workflows.

Server Type: FASTQC | Capabilities: Quality control, sequence analysis, FASTQ processing, Pydantic AI reasoning Pydantic AI Integration: Embedded agent for automated quality assessment and report generation

Available Tools: - run_fastqc: Run FastQC quality control on FASTQ files with comprehensive parameter support - check_fastqc_version: Check the version of FastQC installed - list_fastqc_outputs: List FastQC output files in a directory

Samtools Server¶

::: DeepResearch.src.tools.bioinformatics.samtools_server.SamtoolsServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.samtools_server import SamtoolsServer

Samtools is a suite of utilities for interacting with high-throughput sequencing data. This MCP server provides strongly-typed access to SAM/BAM processing tools.

Server Type: SAMTOOLS | Capabilities: Sequence analysis, BAM/SAM processing, statistics

Available Tools: - samtools_view: Convert between SAM and BAM formats, extract regions - samtools_sort: Sort BAM file by coordinate or read name - samtools_index: Index a BAM file for fast random access - samtools_flagstat: Generate flag statistics for a BAM file - samtools_stats: Generate comprehensive statistics for a BAM file

Bowtie2 Server¶

::: DeepResearch.src.tools.bioinformatics.bowtie2_server.Bowtie2Server
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.bowtie2_server import Bowtie2Server

Bowtie2 is an ultrafast and memory-efficient tool for aligning sequencing reads to long reference sequences. This MCP server provides alignment and indexing capabilities.

Server Type: BOWTIE2 | Capabilities: Sequence alignment, index building, alignment inspection

Available Tools: - bowtie2_align: Align sequencing reads to a reference genome - bowtie2_build: Build a Bowtie2 index from a reference genome - bowtie2_inspect: Inspect a Bowtie2 index

MACS3 Server¶

::: DeepResearch.src.tools.bioinformatics.macs3_server.MACS3Server
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.macs3_server import MACS3Server

MACS3 (Model-based Analysis of ChIP-Seq) is a tool for identifying transcription factor binding sites and histone modifications from ChIP-seq data.

Server Type: MACS3 | Capabilities: ChIP-seq peak calling, transcription factor binding sites

Available Tools: - macs3_callpeak: Call peaks from ChIP-seq data using MACS3 - macs3_bdgcmp: Compare two bedGraph files to generate fold enrichment tracks - macs3_filterdup: Filter duplicate reads from BAM files

HOMER Server¶

::: DeepResearch.src.tools.bioinformatics.homer_server.HOMERServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

HOMER (Hypergeometric Optimization of Motif EnRichment) is a suite of tools for Motif Discovery and next-gen sequencing analysis.

Server Type: HOMER | Capabilities: Motif discovery, ChIP-seq analysis, NGS analysis

Available Tools: - homer_findMotifs: Find motifs in genomic regions using HOMER - homer_annotatePeaks: Annotate peaks with genomic features - homer_mergePeaks: Merge overlapping peaks

HISAT2 Server¶

::: DeepResearch.src.tools.bioinformatics.hisat2_server.HISAT2Server
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

HISAT2 is a fast and sensitive alignment program for mapping next-generation sequencing reads against a population of human genomes.

Server Type: HISAT2 | Capabilities: RNA-seq alignment, spliced alignment

Available Tools: - hisat2_build: Build HISAT2 index from genome FASTA file - hisat2_align: Align RNA-seq reads to reference genome

BEDTools Server¶

::: DeepResearch.src.tools.bioinformatics.bedtools_server.BEDToolsServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

BEDTools is a suite of utilities for comparing, summarizing, and intersecting genomic features in BED format.

Server Type: BEDTOOLS | Capabilities: Genomic interval operations, BED file manipulation

Available Tools: - bedtools_intersect: Find overlapping intervals between two BED files - bedtools_merge: Merge overlapping intervals in a BED file - bedtools_closest: Find closest intervals between two BED files

STAR Server¶

::: DeepResearch.src.tools.bioinformatics.star_server.STARServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

STAR (Spliced Transcripts Alignment to a Reference) is a fast RNA-seq read mapper with support for splice-junctions.

Server Type: STAR | Capabilities: RNA-seq alignment, transcriptome analysis, spliced alignment

Available Tools: - star_genomeGenerate: Generate STAR genome index from reference genome - star_alignReads: Align RNA-seq reads to reference genome using STAR

BWA Server¶

::: DeepResearch.src.tools.bioinformatics.bwa_server.BWAServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

BWA (Burrows-Wheeler Aligner) is a software package for mapping low-divergent sequences against a large reference genome.

Server Type: BWA | Capabilities: DNA sequence alignment, short read alignment

Available Tools: - bwa_index: Build BWA index from reference genome FASTA file - bwa_mem: Align DNA sequencing reads using BWA-MEM algorithm - bwa_aln: Align DNA sequencing reads using BWA-ALN algorithm

MultiQC Server¶

::: DeepResearch.src.tools.bioinformatics.multiqc_server.MultiQCServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

MultiQC is a tool to aggregate results from bioinformatics analyses across many samples into a single report.

Server Type: MULTIQC | Capabilities: Report generation, quality control visualization

Available Tools: - multiqc_run: Generate MultiQC report from bioinformatics tool outputs - multiqc_modules: List available MultiQC modules

Salmon Server¶

::: DeepResearch.src.tools.bioinformatics.salmon_server.SalmonServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Salmon is a tool for quantifying the expression of transcripts using RNA-seq data.

Server Type: SALMON | Capabilities: RNA-seq quantification, transcript abundance estimation

Available Tools: - salmon_index: Build Salmon index from transcriptome FASTA - salmon_quant: Quantify RNA-seq reads using Salmon pseudo-alignment

StringTie Server¶

::: DeepResearch.src.tools.bioinformatics.stringtie_server.StringTieServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

StringTie is a fast and highly efficient assembler of RNA-seq alignments into potential transcripts.

Server Type: STRINGTIE | Capabilities: Transcript assembly, quantification, differential expression

Available Tools: - stringtie_assemble: Assemble transcripts from RNA-seq alignments - stringtie_merge: Merge transcript assemblies from multiple runs

FeatureCounts Server¶

::: DeepResearch.src.tools.bioinformatics.featurecounts_server.FeatureCountsServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

FeatureCounts is a highly efficient general-purpose read summarization program that counts mapped reads for genomic features.

Server Type: FEATURECOUNTS | Capabilities: Read counting, gene expression quantification

Available Tools: - featurecounts_count: Count reads overlapping genomic features

TrimGalore Server¶

::: DeepResearch.src.tools.bioinformatics.trimgalore_server.TrimGaloreServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Trim Galore is a wrapper script to automate quality and adapter trimming as well as quality control.

Server Type: TRIMGALORE | Capabilities: Adapter trimming, quality filtering, FASTQ preprocessing

Available Tools: - trimgalore_trim: Trim adapters and low-quality bases from FASTQ files

Kallisto Server¶

::: DeepResearch.src.tools.bioinformatics.kallisto_server.KallistoServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Kallisto is a program for quantifying abundances of transcripts from RNA-seq data.

Server Type: KALLISTO | Capabilities: Fast RNA-seq quantification, pseudo-alignment

Available Tools: - kallisto_index: Build Kallisto index from transcriptome - kallisto_quant: Quantify RNA-seq reads using pseudo-alignment

HTSeq Server¶

::: DeepResearch.src.tools.bioinformatics.htseq_server.HTSeqServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

HTSeq is a Python package for analyzing high-throughput sequencing data.

Server Type: HTSEQ | Capabilities: Read counting, gene expression analysis

Available Tools: - htseq_count: Count reads overlapping genomic features using HTSeq

TopHat Server¶

::: DeepResearch.src.tools.bioinformatics.tophat_server.TopHatServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

TopHat is a fast splice junction mapper for RNA-seq reads.

Server Type: TOPHAT | Capabilities: RNA-seq splice-aware alignment, junction discovery

Available Tools: - tophat_align: Align RNA-seq reads to reference genome

Picard Server¶

::: DeepResearch.src.tools.bioinformatics.picard_server.PicardServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Picard is a set of command line tools for manipulating high-throughput sequencing data.

Server Type: PICARD | Capabilities: SAM/BAM processing, duplicate marking, quality control

Available Tools: - picard_mark_duplicates: Mark duplicate reads in BAM files - picard_collect_alignment_summary_metrics: Collect alignment summary metrics

BCFtools Server¶

::: DeepResearch.src.tools.bioinformatics.bcftools_server.BCFtoolsServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.bcftools_server import BCFtoolsServer

BCFtools is a suite of programs for manipulating variant calls in the Variant Call Format (VCF) and its binary counterpart BCF. This MCP server provides strongly-typed access to BCFtools with Pydantic AI integration for variant analysis workflows.

Server Type: BCFTOOLS | Capabilities: Variant analysis, VCF processing, genomics, Pydantic AI reasoning Pydantic AI Integration: Embedded agent for automated variant filtering and analysis

Available Tools: - bcftools_view: View, subset and filter VCF/BCF files - bcftools_stats: Parse VCF/BCF files and generate statistics - bcftools_filter: Filter VCF/BCF files using arbitrary expressions

BEDTools Server¶

::: DeepResearch.src.tools.bioinformatics.bedtools_server.BEDToolsServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.bedtools_server import BEDToolsServer

BEDtools is a suite of utilities for comparing, summarizing, and intersecting genomic features in BED format. This MCP server provides strongly-typed access to BEDtools with Pydantic AI integration for genomic interval analysis.

Server Type: BEDTOOLS | Capabilities: Genomics, BED operations, interval arithmetic, Pydantic AI reasoning Pydantic AI Integration: Embedded agent for automated genomic analysis workflows

Available Tools: - bedtools_intersect: Find overlapping intervals between genomic features - bedtools_merge: Merge overlapping/adjacent intervals

Cutadapt Server¶

::: DeepResearch.src.tools.bioinformatics.cutadapt_server.CutadaptServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.cutadapt_server import CutadaptServer

Cutadapt is a tool for removing adapter sequences, primers, and poly-A tails from high-throughput sequencing reads. This MCP server provides strongly-typed access to Cutadapt with Pydantic AI integration for sequence preprocessing workflows.

Server Type: CUTADAPT | Capabilities: Adapter trimming, sequence preprocessing, FASTQ processing, Pydantic AI reasoning Pydantic AI Integration: Embedded agent for automated adapter detection and trimming

Available Tools: - cutadapt_trim: Remove adapters and low-quality bases from FASTQ files

Fastp Server¶

::: DeepResearch.src.tools.bioinformatics.fastp_server.FastpServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.fastp_server import FastpServer

Fastp is an ultra-fast all-in-one FASTQ preprocessor that can perform quality control, adapter trimming, quality filtering, per-read quality pruning, and many other operations. This MCP server provides strongly-typed access to Fastp with Pydantic AI integration.

Server Type: FASTP | Capabilities: FASTQ preprocessing, quality control, adapter trimming, Pydantic AI reasoning Pydantic AI Integration: Embedded agent for automated quality control workflows

Available Tools: - fastp_process: Comprehensive FASTQ preprocessing and quality control

BUSCO Server¶

::: DeepResearch.src.tools.bioinformatics.busco_server.BUSCOServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

from DeepResearch.src.tools.bioinformatics.busco_server import BUSCOServer

BUSCO (Benchmarking Universal Single-Copy Orthologs) assesses genome assembly and annotation completeness by searching for single-copy orthologs. This MCP server provides strongly-typed access to BUSCO with Pydantic AI integration for genome quality assessment.

Server Type: BUSCO | Capabilities: Genome completeness assessment, ortholog detection, quality metrics, Pydantic AI reasoning Pydantic AI Integration: Embedded agent for automated genome quality analysis

Available Tools: - busco_run: Assess genome assembly completeness using BUSCO

DeepTools Server¶

::: DeepResearch.src.tools.bioinformatics.deeptools_server.DeepToolsServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

deepTools is a suite of user-friendly tools for the exploration of deep-sequencing data.

Server Type: DEEPTOOLS | Capabilities: NGS data analysis, visualization, quality control

Available Tools: - deeptools_bamCoverage: Generate coverage tracks from BAM files - deeptools_computeMatrix: Compute matrices for heatmaps from BAM files

FreeBayes Server¶

::: DeepResearch.src.tools.bioinformatics.freebayes_server.FreeBayesServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

FreeBayes is a Bayesian genetic variant detector designed to find small polymorphisms.

Server Type: FREEBAYES | Capabilities: Variant calling, SNP detection, indel detection

Available Tools: - freebayes_call: Call variants from BAM files using FreeBayes

Flye Server¶

::: DeepResearch.src.tools.bioinformatics.flye_server.FlyeServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Flye is a de novo assembler for single-molecule sequencing reads.

Server Type: FLYE | Capabilities: Genome assembly, long-read assembly

Available Tools: - flye_assemble: Assemble genome from long-read sequencing data

MEME Server¶

::: DeepResearch.src.tools.bioinformatics.meme_server.MEMEServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

MEME (Multiple EM for Motif Elicitation) is a tool for discovering motifs in a group of related DNA or protein sequences.

Server Type: MEME | Capabilities: Motif discovery, sequence analysis

Available Tools: - meme_discover: Discover motifs in DNA or protein sequences

Minimap2 Server¶

::: DeepResearch.src.tools.bioinformatics.minimap2_server.Minimap2Server
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Minimap2 is a versatile pairwise aligner for nucleotide sequences.

Server Type: MINIMAP2 | Capabilities: Sequence alignment, long-read alignment

Available Tools: - minimap2_align: Align sequences using minimap2 algorithm

Qualimap Server¶

::: DeepResearch.src.tools.bioinformatics.qualimap_server.QualimapServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Qualimap is a platform-independent application written in Java and R that provides both a Graphical User Interface (GUI) and a command-line interface to facilitate the quality control of alignment sequencing data.

Server Type: QUALIMAP | Capabilities: Quality control, alignment analysis, RNA-seq analysis

Available Tools: - qualimap_bamqc: Generate quality control report for BAM files - qualimap_rnaseq: Generate RNA-seq quality control report

Seqtk Server¶

::: DeepResearch.src.tools.bioinformatics.seqtk_server.SeqtkServer
    handler: python
    options:
      docstring_style: google
      show_category_heading: true

Seqtk is a fast and lightweight tool for processing sequences in the FASTA or FASTQ format.

Server Type: SEQTK | Capabilities: FASTA/FASTQ processing, sequence manipulation

Available Tools: - seqtk_seq: Convert and manipulate FASTA/FASTQ files - seqtk_subseq: Extract subsequences from FASTA/FASTQ files

Deployment¶

from DeepResearch.src.tools.bioinformatics.fastqc_server import FastQCServer
from DeepResearch.datatypes.mcp import MCPServerConfig

config = MCPServerConfig(
    server_name="fastqc-server",
    server_type="fastqc",
    container_image="python:3.11-slim",
)

server = FastQCServer(config)
deployment = await server.deploy_with_testcontainers()

Available Servers by Category¶

Quality Control & Preprocessing: - FastQC, TrimGalore, Cutadapt, Fastp, MultiQC, Qualimap, Seqtk

Sequence Alignment: - Bowtie2, BWA, HISAT2, STAR, TopHat, Minimap2

RNA-seq Quantification & Assembly: - Salmon, Kallisto, StringTie, FeatureCounts, HTSeq

Genome Analysis & Manipulation: - Samtools, BEDTools, Picard, DeepTools

ChIP-seq & Epigenetics: - MACS3, HOMER, MEME

Genome Assembly: - Flye

Genome Assembly Assessment: - BUSCO

Variant Analysis: - BCFtools, FreeBayes

Enhanced MCP Server Management Tools¶

DeepCritical provides comprehensive tools for managing MCP server deployments using testcontainers with Pydantic AI integration:

MCPServerListTool¶

Lists all available vendored MCP servers.

Features: - Lists all 29 MCP servers with descriptions and capabilities - Shows deployment status and available tools - Supports filtering and detailed information

MCPServerDeployTool¶

Deploys vendored MCP servers using testcontainers.

Features: - Deploys any of the 29 MCP servers in isolated containers - Supports custom configurations and resource limits - Provides detailed deployment information

MCPServerExecuteTool¶

Executes tools on deployed MCP servers.

Features: - Executes specific tools on deployed MCP servers - Supports synchronous and asynchronous execution - Provides comprehensive error handling and retry logic - Returns detailed execution results

MCPServerStatusTool¶

Checks deployment status of MCP servers.

Features: - Checks deployment status of individual servers or all servers - Provides container and deployment information - Supports health monitoring

MCPServerStopTool¶

Stops deployed MCP servers.

Features: - Stops and cleans up deployed MCP server containers - Provides confirmation of stop operations - Handles resource cleanup

TestcontainersDeployer¶

Deployer for MCP servers using testcontainers with integrated code execution.

Methods:

Name	Description
`cleanup_server`	Clean up a deployed server and its files.
`create_deployment_config`	Create deployment configuration for a server.
`create_server_files`	Create necessary files for server deployment.
`deploy_server`	Enhanced deployment with Pydantic AI integration.
`execute_code`	Execute code using the deployed server's container environment.
`execute_code_blocks`	Execute multiple code blocks using the deployed server's environment.
`execute_tool`	Execute a tool on a deployed server.
`get_server_status`	Get the status of a deployed server.
`health_check`	Perform health check on a deployed server.
`list_servers`	List all deployed servers.
`stop_server`	Stop a deployed MCP server.

Attributes:

Name	Type	Description
`code_executors`	`dict[str, DockerCommandLineCodeExecutor]`
`containers`	`dict[str, Any]`
`deployments`	`dict[str, MCPServerDeployment]`
`python_execution_tools`	`dict[str, PythonCodeExecutionTool]`
`server_implementations`

Source code in DeepResearch/src/utils/testcontainers_deployer.py

def __init__(self):
    self.deployments: dict[str, MCPServerDeployment] = {}
    self.containers: dict[
        str, Any
    ] = {}  # Would hold testcontainers container objects
    self.code_executors: dict[str, DockerCommandLineCodeExecutor] = {}
    self.python_execution_tools: dict[str, PythonCodeExecutionTool] = {}

    # Map server types to their implementations
    self.server_implementations = {
        "fastqc": FastQCServer,
        "samtools": SamtoolsServer,
        "bowtie2": Bowtie2Server,
    }

Attributes¶

code_executors `instance-attribute` ¶

code_executors: dict[
    str, DockerCommandLineCodeExecutor
] = {}

containers `instance-attribute` ¶

containers: dict[str, Any] = {}

deployments `instance-attribute` ¶

deployments: dict[str, MCPServerDeployment] = {}

python_execution_tools `instance-attribute` ¶

python_execution_tools: dict[
    str, PythonCodeExecutionTool
] = {}

server_implementations `instance-attribute` ¶

server_implementations = {
    "fastqc": FastQCServer,
    "samtools": SamtoolsServer,
    "bowtie2": Bowtie2Server,
}

Functions¶

cleanup_server `async` ¶

cleanup_server(server_name: str) -> bool

Clean up a deployed server and its files.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def cleanup_server(self, server_name: str) -> bool:
    """Clean up a deployed server and its files."""
    try:
        # Stop the server
        await self.stop_server(server_name)

        # Remove from deployments
        if server_name in self.deployments:
            del self.deployments[server_name]

        # Remove container reference
        if server_name in self.containers:
            del self.containers[server_name]

        logger.info("Cleaned up MCP server '%s'", server_name)
        return True

    except Exception:
        logger.exception("Failed to cleanup server '%s'", server_name)
        return False

create_deployment_config ¶

create_deployment_config(
    server_name: str, **kwargs
) -> TestcontainersConfig

Create deployment configuration for a server.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

def create_deployment_config(
    self, server_name: str, **kwargs
) -> TestcontainersConfig:
    """Create deployment configuration for a server."""
    base_config = TestcontainersConfig()

    # Customize based on server type
    if server_name in self.server_implementations:
        server = self.server_implementations[server_name]

        # Add server-specific environment variables
        base_config.environment_variables.update(
            {
                "MCP_SERVER_NAME": server_name,
                "MCP_SERVER_VERSION": getattr(server, "version", "1.0.0"),
                "PYTHONPATH": "/workspace",
            }
        )

        # Add server-specific volumes for data
        base_config.volumes.update(
            {
                f"/tmp/mcp_{server_name}": "/workspace/data",
            }
        )

    # Apply customizations from kwargs
    for key, value in kwargs.items():
        if hasattr(base_config, key):
            setattr(base_config, key, value)

    return base_config

create_server_files `async` ¶

create_server_files(
    server_name: str, output_dir: str
) -> list[str]

Create necessary files for server deployment.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def create_server_files(self, server_name: str, output_dir: str) -> list[str]:
    """Create necessary files for server deployment."""
    files_created = []

    try:
        # Create temporary directory for server files
        server_dir = Path(output_dir) / f"mcp_{server_name}"
        server_dir.mkdir(parents=True, exist_ok=True)

        # Create server script
        server_script = server_dir / f"{server_name}_server.py"

        # Generate server code based on server type
        server_code = self._generate_server_code(server_name)

        with open(server_script, "w") as f:
            f.write(server_code)

        files_created.append(str(server_script))

        # Create requirements file
        requirements_file = server_dir / "requirements.txt"
        requirements_content = self._generate_requirements(server_name)

        with open(requirements_file, "w") as f:
            f.write(requirements_content)

        files_created.append(str(requirements_file))

        logger.info("Created server files for '%s' in %s", server_name, server_dir)
        return files_created

    except Exception:
        logger.exception("Failed to create server files for '%s'", server_name)
        return files_created

deploy_server `async` ¶

deploy_server(
    server_name: str,
    config: TestcontainersConfig | None = None,
    **kwargs,
) -> MCPServerDeployment

Enhanced deployment with Pydantic AI integration.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def deploy_server(
    self, server_name: str, config: TestcontainersConfig | None = None, **kwargs
) -> MCPServerDeployment:
    """Enhanced deployment with Pydantic AI integration."""
    deployment = MCPServerDeployment(
        server_name=server_name,
        status=MCPServerStatus.DEPLOYING,
    )

    try:
        # Get server implementation
        server = self._get_server_implementation(server_name)
        if not server:
            msg = f"Server implementation for '{server_name}' not found"
            raise ValueError(msg)

        # Use testcontainers deployment method if available
        if hasattr(server, "deploy_with_testcontainers"):
            deployment = await server.deploy_with_testcontainers()
        else:
            # Fallback to basic deployment
            deployment = await self._deploy_server_basic(
                server_name, config, **kwargs
            )

        # Update deployment registry
        self.deployments[server_name] = deployment
        self.server_implementations[server_name] = server

        return deployment

    except Exception as e:
        deployment.status = MCPServerStatus.FAILED
        deployment.error_message = str(e)
        self.deployments[server_name] = deployment
        raise

execute_code `async` ¶

execute_code(
    server_name: str,
    code: str,
    language: str = "python",
    timeout: int = 60,
    max_retries: int = 3,
    **kwargs,
) -> dict[str, Any]

Execute code using the deployed server's container environment.

Parameters:

Name	Type	Description	Default
`server_name`	`str`	Name of the deployed server to use for execution	required
`code`	`str`	Code to execute	required
`language`	`str`	Programming language of the code	`'python'`
`timeout`	`int`	Execution timeout in seconds	`60`
`max_retries`	`int`	Maximum number of retry attempts	`3`
`**kwargs`		Additional execution parameters	`{}`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing execution results

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def execute_code(
    self,
    server_name: str,
    code: str,
    language: str = "python",
    timeout: int = 60,
    max_retries: int = 3,
    **kwargs,
) -> dict[str, Any]:
    """Execute code using the deployed server's container environment.

    Args:
        server_name: Name of the deployed server to use for execution
        code: Code to execute
        language: Programming language of the code
        timeout: Execution timeout in seconds
        max_retries: Maximum number of retry attempts
        **kwargs: Additional execution parameters

    Returns:
        Dictionary containing execution results
    """
    deployment = self.deployments.get(server_name)
    if not deployment:
        raise ValueError(f"Server '{server_name}' not deployed")

    if deployment.status != "running":
        raise ValueError(
            f"Server '{server_name}' is not running (status: {deployment.status})"
        )

    # Get or create code executor for this server
    if server_name not in self.code_executors:
        # Create a code executor using the same container
        try:
            # In a real implementation, we'd create a DockerCommandLineCodeExecutor
            # that shares the container with the MCP server
            # For now, we'll use the Python execution tool
            self.python_execution_tools[server_name] = PythonCodeExecutionTool(
                timeout=timeout,
                work_dir=f"/tmp/{server_name}_code_exec",
                use_docker=True,
            )
        except Exception:
            logger.exception(
                "Failed to create code executor for server '%s'", server_name
            )
            raise

    # Execute the code
    tool = self.python_execution_tools[server_name]
    result = tool.run(
        {
            "code": code,
            "timeout": timeout,
            "max_retries": max_retries,
            "language": language,
            **kwargs,
        }
    )

    return {
        "server_name": server_name,
        "success": result.success,
        "output": result.data.get("output", ""),
        "error": result.data.get("error", ""),
        "exit_code": result.data.get("exit_code", -1),
        "execution_time": result.data.get("execution_time", 0.0),
        "retries_used": result.data.get("retries_used", 0),
    }

execute_code_blocks `async` ¶

execute_code_blocks(
    server_name: str, code_blocks: list[CodeBlock], **kwargs
) -> dict[str, Any]

Execute multiple code blocks using the deployed server's environment.

Parameters:

Name	Type	Description	Default
`server_name`	`str`	Name of the deployed server to use for execution	required
`code_blocks`	`list[CodeBlock]`	List of code blocks to execute	required
`**kwargs`		Additional execution parameters	`{}`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing execution results for all blocks

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def execute_code_blocks(
    self, server_name: str, code_blocks: list[CodeBlock], **kwargs
) -> dict[str, Any]:
    """Execute multiple code blocks using the deployed server's environment.

    Args:
        server_name: Name of the deployed server to use for execution
        code_blocks: List of code blocks to execute
        **kwargs: Additional execution parameters

    Returns:
        Dictionary containing execution results for all blocks
    """
    deployment = self.deployments.get(server_name)
    if not deployment:
        raise ValueError(f"Server '{server_name}' not deployed")

    if server_name not in self.code_executors:
        # Create code executor if it doesn't exist
        self.code_executors[server_name] = DockerCommandLineCodeExecutor(
            image=deployment.configuration.image
            if hasattr(deployment.configuration, "image")
            else "python:3.11-slim",
            timeout=kwargs.get("timeout", 60),
            work_dir=f"/tmp/{server_name}_code_blocks",
        )

    executor = self.code_executors[server_name]
    result = executor.execute_code_blocks(code_blocks)

    return {
        "server_name": server_name,
        "success": result.exit_code == 0,
        "output": result.output,
        "exit_code": result.exit_code,
        "command": getattr(result, "command", ""),
        "image": getattr(result, "image", None),
    }

execute_tool `async` ¶

execute_tool(
    server_name: str, tool_name: str, **kwargs
) -> dict[str, Any]

Execute a tool on a deployed server.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def execute_tool(
    self, server_name: str, tool_name: str, **kwargs
) -> dict[str, Any]:
    """Execute a tool on a deployed server."""
    deployment = self.deployments.get(server_name)
    if not deployment:
        msg = f"Server '{server_name}' not deployed"
        raise ValueError(msg)

    if deployment.status != "running":
        msg = f"Server '{server_name}' is not running (status: {deployment.status})"
        raise ValueError(msg)

    # Get server implementation
    server = self.server_implementations.get(server_name)
    if not server:
        msg = f"Server implementation for '{server_name}' not found"
        raise ValueError(msg)

    # Check if tool exists
    available_tools = server.list_tools()
    if tool_name not in available_tools:
        msg = f"Tool '{tool_name}' not found on server '{server_name}'. Available tools: {', '.join(available_tools)}"
        raise ValueError(msg)

    # Execute tool
    try:
        return server.execute_tool(tool_name, **kwargs)
    except Exception as e:
        msg = f"Tool execution failed: {e}"
        raise ValueError(msg)

get_server_status `async` ¶

get_server_status(
    server_name: str,
) -> MCPServerDeployment | None

Get the status of a deployed server.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def get_server_status(self, server_name: str) -> MCPServerDeployment | None:
    """Get the status of a deployed server."""
    return self.deployments.get(server_name)

health_check `async` ¶

health_check(server_name: str) -> bool

Perform health check on a deployed server.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def health_check(self, server_name: str) -> bool:
    """Perform health check on a deployed server."""
    deployment = self.deployments.get(server_name)
    if not deployment:
        return False

    if deployment.status != "running":
        return False

    try:
        # In a real implementation, this would check if the container is healthy
        # For now, we'll just check if the deployment exists and is running
        return True
    except Exception:
        logger.exception("Health check failed for server '%s'", server_name)
        return False

list_servers `async` ¶

list_servers() -> list[MCPServerDeployment]

List all deployed servers.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def list_servers(self) -> list[MCPServerDeployment]:
    """List all deployed servers."""
    return list(self.deployments.values())

stop_server `async` ¶

stop_server(server_name: str) -> bool

Stop a deployed MCP server.

Source code in DeepResearch/src/utils/testcontainers_deployer.py

async def stop_server(self, server_name: str) -> bool:
    """Stop a deployed MCP server."""
    if server_name not in self.deployments:
        logger.warning("Server '%s' not found in deployments", server_name)
        return False

    deployment = self.deployments[server_name]

    try:
        # In a real implementation, this would stop the testcontainers container
        deployment.status = "stopped"
        deployment.finished_at = None  # Would be set by testcontainers

        # Clean up container reference
        if server_name in self.containers:
            del self.containers[server_name]

        logger.info("Stopped MCP server '%s'", server_name)
        return True

    except Exception as e:
        logger.exception("Failed to stop MCP server '%s'", server_name)
        deployment.status = "failed"
        deployment.error_message = str(e)
        return False

Core deployment infrastructure for MCP servers using testcontainers with integrated code execution.

Features: - MCP Server Deployment: Deploy bioinformatics servers (FastQC, SAMtools, Bowtie2) in isolated containers - Testcontainers Integration: Isolated container environments for secure, reproducible execution - Code Execution: AG2-style code execution within deployed containers - Health Monitoring: Built-in health checks and automatic recovery - Resource Management: Configurable CPU, memory, and timeout limits - Multi-Server Support: Deploy multiple servers simultaneously with resource optimization

Key Methods: - deploy_server(): Deploy MCP servers with custom configurations - execute_code(): Execute code within deployed server containers - execute_code_blocks(): Execute multiple code blocks with container isolation - health_check(): Perform health monitoring on deployed servers - stop_server(): Gracefully stop and cleanup deployed servers

Configuration:

# Testcontainers configuration
testcontainers:
  image: "python:3.11-slim"
  working_directory: "/workspace"
  auto_remove: true
  privileged: false
  environment_variables:
    PYTHONPATH: "/workspace"
  volumes:
    /tmp/mcp_data: "/workspace/data"

Usage Examples¶

Creating a Custom Tool¶

from deepresearch.tools import ToolRunner, ToolSpec, ToolCategory
from deepresearch.datatypes import ExecutionResult

class CustomAnalysisTool(ToolRunner):
    """Custom tool for data analysis."""

    def __init__(self):
        super().__init__(ToolSpec(
            name="custom_analysis",
            description="Performs custom data analysis",
            category=ToolCategory.ANALYTICS,
            inputs={
                "data": "dict",
                "analysis_type": "str",
                "parameters": "dict"
            },
            outputs={
                "result": "dict",
                "statistics": "dict"
            }
        ))

    def run(self, parameters: Dict[str, Any]) -> ExecutionResult:
        """Execute the analysis.

        Args:
            parameters: Tool parameters including data, analysis_type, and parameters

        Returns:
            ExecutionResult with analysis results
        """
        try:
            data = parameters["data"]
            analysis_type = parameters["analysis_type"]

            # Perform analysis
            result = self._perform_analysis(data, analysis_type, parameters)

            return ExecutionResult(
                success=True,
                data={
                    "result": result,
                    "statistics": self._calculate_statistics(result)
                }
            )
        except Exception as e:
            return ExecutionResult(
                success=False,
                error=str(e),
                error_type=type(e).__name__
            )

    def _perform_analysis(self, data: Dict, analysis_type: str, params: Dict) -> Dict:
        """Perform the actual analysis logic."""
        # Implementation here
        return {"analysis": "completed"}

    def _calculate_statistics(self, result: Dict) -> Dict:
        """Calculate statistics for the result."""
        # Implementation here
        return {"stats": "calculated"}

Registering and Using Tools¶

from deepresearch.tools import ToolRegistry

# Get global registry
registry = ToolRegistry.get_instance()

# Register custom tool
registry.register_tool(
    tool_spec=CustomAnalysisTool().get_spec(),
    tool_runner=CustomAnalysisTool()
)

# Use the tool
result = registry.execute_tool("custom_analysis", {
    "data": {"key": "value"},
    "analysis_type": "statistical",
    "parameters": {"confidence": 0.95}
})

if result.success:
    print(f"Analysis result: {result.data}")
else:
    print(f"Analysis failed: {result.error}")

Tool Categories and Organization¶

from deepresearch.tools import ToolCategory

# Available categories
categories = [
    ToolCategory.KNOWLEDGE_QUERY,    # Information retrieval
    ToolCategory.SEQUENCE_ANALYSIS,  # Bioinformatics sequence tools
    ToolCategory.STRUCTURE_PREDICTION, # Protein structure tools
    ToolCategory.MOLECULAR_DOCKING,  # Drug-target interaction
    ToolCategory.DE_NOVO_DESIGN,     # Novel molecule design
    ToolCategory.FUNCTION_PREDICTION, # Function annotation
    ToolCategory.RAG,               # Retrieval-augmented generation
    ToolCategory.SEARCH,            # Web and document search
    ToolCategory.ANALYTICS,         # Data analysis and visualization
    ToolCategory.CODE_EXECUTION,    # Code execution environments
]

Tools API¶

Tool Framework¶

ToolRunner¶

ToolSpec¶

ToolRegistry¶

Tool Categories¶

Execution Framework¶

ExecutionResult¶

ToolRequest¶

ToolResponse¶

Domain Tools¶

Knowledge Query Tools¶

Web Search Tools¶

Attributes¶

spec instance-attribute ¶

Functions¶

run ¶

validate ¶

Attributes¶

spec instance-attribute ¶

Functions¶

run ¶

validate ¶

Sequence Analysis Tools¶

Bioinformatics Tools¶

Attributes¶

spec instance-attribute ¶

Functions¶

run ¶

validate ¶

Attributes¶

spec instance-attribute ¶

Functions¶

run ¶

validate ¶

Deep Search Tools¶

Attributes¶

spec instance-attribute ¶

Functions¶

run ¶

validate ¶

RAG Tools¶

Attributes¶

spec instance-attribute ¶

Functions¶

run ¶

validate ¶

Code Execution Tools¶

Attributes¶

bash_agent instance-attribute ¶

max_retries instance-attribute ¶

model_name instance-attribute ¶

python_agent instance-attribute ¶

timeout instance-attribute ¶

universal_agent instance-attribute ¶

Functions¶

create_code_block ¶

generate_and_create_block async ¶

generate_bash_command async ¶

generate_code async ¶

generate_python_code async ¶

Attributes¶

docker_executor instance-attribute ¶

jupyter_config instance-attribute ¶

jupyter_executor instance-attribute ¶

local_executor instance-attribute ¶

max_retries instance-attribute ¶

model_name instance-attribute ¶

python_tool instance-attribute ¶

timeout instance-attribute ¶

use_docker instance-attribute ¶

use_jupyter instance-attribute ¶

Functions¶

execute_code async ¶

execute_code_block async ¶

Structure Prediction Tools¶

Molecular Docking Tools¶

De Novo Design Tools¶

Function Prediction Tools¶

RAG Tools¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

bash_agent `instance-attribute` ¶

max_retries `instance-attribute` ¶

model_name `instance-attribute` ¶

python_agent `instance-attribute` ¶

timeout `instance-attribute` ¶

universal_agent `instance-attribute` ¶

generate_and_create_block `async` ¶

generate_bash_command `async` ¶

generate_code `async` ¶

generate_python_code `async` ¶

docker_executor `instance-attribute` ¶

jupyter_config `instance-attribute` ¶

jupyter_executor `instance-attribute` ¶

local_executor `instance-attribute` ¶

max_retries `instance-attribute` ¶

model_name `instance-attribute` ¶

python_tool `instance-attribute` ¶

timeout `instance-attribute` ¶

use_docker `instance-attribute` ¶

use_jupyter `instance-attribute` ¶

execute_code `async` ¶

execute_code_block `async` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶

spec `instance-attribute` ¶