# How to Implement Custom Code Interpreters with the InterpreterFactory in MathModelAgent

> Learn to implement custom code interpreters in MathModelAgent by subclassing BaseCodeInterpreter and registering with InterpreterFactory. Enhance your agent's capabilities today.

- Repository: [Sanjin/mathmodelagent](https://github.com/jihe520/mathmodelagent)
- Tags: how-to-guide
- Published: 2026-03-04

---

**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`](https://github.com/jihe520/mathmodelagent/blob/main/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`](https://github.com/jihe520/mathmodelagent/blob/main/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`](https://github.com/jihe520/mathmodelagent/blob/main/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__()`.

```python

# 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`](https://github.com/jihe520/mathmodelagent/blob/main/backend/app/tools/interpreter_factory.py) to import your new class and add a conditional branch that instantiates it when `kind="custom"` is requested.

```python

# 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.

```python
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`](https://github.com/jihe520/mathmodelagent/blob/main/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`](https://github.com/jihe520/mathmodelagent/blob/main/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`](https://github.com/jihe520/mathmodelagent/blob/main/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`](https://github.com/jihe520/mathmodelagent/blob/main/backend/app/tools/docker_interpreter.py)). Import this module in [`backend/app/tools/interpreter_factory.py`](https://github.com/jihe520/mathmodelagent/blob/main/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`.