How to Implement Custom Code Interpreters with the InterpreterFactory in MathModelAgent

You can add custom code interpreters to MathModelAgent by subclassing BaseCodeInterpreter and registering your concrete implementation in the create_interpreter factory function located in backend/app/tools/interpreter_factory.py.

The jihe520/mathmodelagent repository isolates code execution behind a pluggable interpreter abstraction, allowing developers to swap execution backends without touching the task orchestration or WebSocket delivery systems. By following the factory pattern established in the backend, you can integrate Docker-based sandboxes, remote Jupyter servers, or specialized GPU environments as first-class citizens. This guide demonstrates the exact implementation steps using the source code structure from the MathModelAgent backend.

Core Architecture of the Interpreter System

The MathModelAgent backend decouples execution logic from workflow management through two primary components: the abstract base class that defines the contract, and the factory function that handles instantiation.

The BaseCodeInterpreter Contract

All interpreter implementations inherit from BaseCodeInterpreter defined in backend/app/tools/base_interpreter.py. This abstract class establishes four required methods that every concrete interpreter must implement:

• initialize() – Sets up the execution environment (e.g., launching containers, opening SSH sessions, or starting local kernels)
• execute_code(code: str) – Runs user-generated code and returns a standardized tuple of (text_to_gpt, error, error_message)
• get_created_images(section: str) – Detects and returns image files generated during execution
• cleanup() – Releases resources and terminates sessions

The base class also provides utilities for WebSocket push notifications, section handling, and text sanitization, which subclasses inherit automatically.

The InterpreterFactory Pattern

The create_interpreter function in backend/app/tools/interpreter_factory.py serves as the central dispatcher. It accepts a kind parameter (Literal["remote", "local"]) and inspects settings.E2B_API_KEY to determine whether to instantiate E2BCodeInterpreter or LocalCodeInterpreter. By extending this factory, you can introduce new execution backends while maintaining backward compatibility with existing code that depends only on the abstract interface.

Step-by-Step Implementation Guide

Follow these three steps to add a completely custom execution backend to the MathModelAgent system.

Step 1: Create a Concrete Interpreter Class

Create a new file in backend/app/tools/ that subclasses BaseCodeInterpreter and implements the four abstract methods. The constructor must accept task_id, work_dir, and notebook_serializer parameters and pass them to super().__init__().

# backend/app/tools/my_custom_interpreter.py

from app.tools.base_interpreter import BaseCodeInterpreter
from app.tools.notebook_serializer import NotebookSerializer
from app.utils.log_util import logger
from app.services.redis_manager import redis_manager
from app.schemas.response import SystemMessage, OutputItem, ResultModel

class MyCustomInterpreter(BaseCodeInterpreter):
    def __init__(self, task_id: str, work_dir: str, notebook_serializer: NotebookSerializer):
        super().__init__(task_id, work_dir, notebook_serializer)
        # Initialise fields specific to your backend here

        self.session = None

    async def initialize(self):
        logger.info("Starting MyCustomInterpreter session")
        # Example: launch a Docker container, open a remote SSH session, etc.

        self.session = await launch_my_backend(task_id=self.task_id, work_dir=self.work_dir)

    async def execute_code(self, code: str):
        logger.info(f"Executing code in MyCustomInterpreter: {code}")
        await redis_manager.publish_message(self.task_id, SystemMessage(content="Running custom code"))
        
        # Send the code to your backend and collect raw outputs

        raw = await self.session.run(code)
        
        # Transform raw output into the unified format expected by the UI

        text_to_gpt = [raw.stdout] if raw.stdout else []
        content: list[OutputItem] = [ResultModel(type="result", format="text", msg=raw.stdout)]
        
        # Push results to the front-end via WebSocket

        await self._push_to_websocket(content)
        return "\n".join(text_to_gpt), raw.error, raw.error_message

    async def get_created_images(self, section: str):
        # Detect new image files that the backend generated

        images = await self.session.list_images()
        self.add_section(section)
        self.section_output[section]["images"].extend(images)
        return images

    async def cleanup(self):
        logger.info("Cleaning up MyCustomInterpreter")
        if self.session:
            await self.session.close()

Step 2: Register with the Factory

Modify backend/app/tools/interpreter_factory.py to import your new class and add a conditional branch that instantiates it when kind="custom" is requested.

# backend/app/tools/interpreter_factory.py

from typing import Literal
from app.tools.my_custom_interpreter import MyCustomInterpreter  # New import

from app.tools.local_interpreter import LocalCodeInterpreter
from app.tools.e2b_interpreter import E2BCodeInterpreter
from app.core.config import settings

async def create_interpreter(
    kind: Literal["remote", "local", "custom"] = "local",
    *,
    task_id: str,
    work_dir: str,
    notebook_serializer,
    timeout=3000,
):
    # Existing auto-selection logic based on E2B_API_KEY

    if kind == "remote" or (kind == "local" and settings.E2B_API_KEY):
        from app.tools.e2b_interpreter import E2BCodeInterpreter
        interp = E2BCodeInterpreter(
            task_id=task_id,
            work_dir=work_dir,
            notebook_serializer=notebook_serializer,
            api_key=settings.E2B_API_KEY,
        )
    elif kind == "custom":
        # New custom backend registration

        interp = MyCustomInterpreter(
            task_id=task_id,
            work_dir=work_dir,
            notebook_serializer=notebook_serializer,
        )
        await interp.initialize()
        return interp
    else:
        interp = LocalCodeInterpreter(
            task_id=task_id,
            work_dir=work_dir,
            notebook_serializer=notebook_serializer,
            timeout=timeout,
        )
    
    await interp.initialize()
    return interp

The factory preserves its existing auto-selection behavior for "remote" and "local" kinds while allowing explicit instantiation of your custom interpreter via the kind parameter.

Step 3: Instantiate in Your Workflow

Any component requiring code execution—such as task workers or API endpoints—uses the factory without knowing which concrete interpreter is returned. Pass kind="custom" to trigger your new implementation.

from app.tools.interpreter_factory import create_interpreter

async def run_analysis_task(task):
    serializer = NotebookSerializer(task_id=task.id, work_dir=task.work_dir)
    
    # Request the custom interpreter explicitly

    interpreter = await create_interpreter(
        kind="custom",
        task_id=task.id,
        work_dir=task.work_dir,
        notebook_serializer=serializer,
    )
    
    try:
        output, error, msg = await interpreter.execute_code(task.user_code)
        # Process results...

    finally:
        await interpreter.cleanup()

Because the rest of the system (task orchestration, WebSocket updates, and notebook serialization) depends only on the BaseCodeInterpreter interface, no other files require modification.

Summary

• The abstract base class BaseCodeInterpreter in backend/app/tools/base_interpreter.py defines the contract that all interpreters must fulfill, including initialize, execute_code, get_created_images, and cleanup.

• The factory function create_interpreter in backend/app/tools/interpreter_factory.py centralizes instantiation logic and selects implementations based on the kind parameter and configuration settings.

• Custom implementations require subclassing BaseCodeInterpreter, implementing the four abstract methods, and adding a conditional branch to the factory that recognizes a new kind value (e.g., "custom").

• Existing workflows remain unchanged because they consume interpreters through the abstract interface, enabling seamless integration of Docker-based sandboxes, remote GPU clusters, or specialized execution environments.

Frequently Asked Questions

What methods must a custom code interpreter implement?

A custom interpreter must implement four abstract methods defined in BaseCodeInterpreter: initialize() to set up the environment, execute_code(code) to run user code and return standardized output, get_created_images(section) to collect generated artifacts, and cleanup() to release resources. The base class handles WebSocket communication and output formatting automatically.

How does the InterpreterFactory decide which interpreter to create?

The create_interpreter function in backend/app/tools/interpreter_factory.py inspects the kind parameter and settings.E2B_API_KEY. If kind="remote" or if an E2B API key exists, it returns E2BCodeInterpreter. If kind="local", it returns LocalCodeInterpreter. You can extend this logic to recognize custom kinds (e.g., kind="custom") and return your concrete subclass instead.

Can I use the custom interpreter alongside existing local and remote interpreters?

Yes. The factory pattern allows all interpreter types to coexist. Existing code that does not specify a kind parameter continues to receive the default local or remote interpreter based on configuration, while new workflows can explicitly request the custom implementation by passing kind="custom" to create_interpreter.

Where should I place the new interpreter file in the codebase?

Create the new interpreter file within the backend/app/tools/ directory (e.g., backend/app/tools/docker_interpreter.py). Import this module in backend/app/tools/interpreter_factory.py and reference the class within the create_interpreter function. This maintains consistency with the existing project structure used by LocalCodeInterpreter and E2BCodeInterpreter.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}