Skip to main content
MentatLab docs

MentatLab Agent SDK

This guide covers everything you need to build agents for the MentatLab orchestration platform.

Overview

Agents are containerized units of work that:

  • Execute tasks within a DAG (directed acyclic graph) workflow
  • Communicate via NDJSON events over stdout
  • Run in Kubernetes Jobs or subprocess mode
  • Can be registered and discovered via the Agent Registry API

Agent Manifest Schema

Every agent must be registered with a manifest. Here's the complete schema:

{
  "id": "myorg.my-agent",
  "name": "My Agent",
  "version": "1.0.0",
  "image": "registry.example.com/my-agent:v1.0.0",
  "command": ["python", "-m", "my_agent"],
  "capabilities": ["text-processing", "summarization"],
  "description": "Processes text and generates summaries",
  "author": "Your Name <[email protected]>",
  "schema": {
    "input": {
      "type": "object",
      "properties": {
        "text": { "type": "string" },
        "max_length": { "type": "integer", "default": 500 }
      },
      "required": ["text"]
    },
    "output": {
      "type": "object",
      "properties": {
        "summary": { "type": "string" },
        "word_count": { "type": "integer" }
      }
    }
  },
  "metadata": {
    "category": "nlp",
    "cost_per_run": "0.001"
  }
}

Required Fields

FieldTypeDescription
idstringUnique identifier (e.g., myorg.agent-name)
namestringHuman-readable display name
versionstringSemantic version (e.g., 1.0.0)

Optional Fields

FieldTypeDescription
imagestringContainer image URL for K8s execution
commandstring[]Entrypoint command override
capabilitiesstring[]Tags for filtering/discovery
schemaobjectJSON Schema for input/output validation
descriptionstringAgent description
authorstringAuthor name/email
metadataobjectCustom key-value pairs

Event Protocol (NDJSON)

Agents communicate with the orchestrator via newline-delimited JSON (NDJSON) on stdout. Each line is a complete JSON object.

Event Structure

{
  "type": "log|checkpoint|metric|output|result|error|progress|heartbeat",
  "level": "debug|info|warn|error",
  "message": "Human-readable message",
  "data": { "key": "value" },
  "correlation_id": "run-123-node-abc",
  "ts": "2024-01-15T10:30:00.000Z"
}

Event Types

1. Log Events

Used for general logging and debugging.

{"type":"log","level":"info","message":"Processing started","data":{"input_size":1024},"ts":"2024-01-15T10:30:00Z"}
{"type":"log","level":"error","message":"Failed to parse input","data":{"error":"invalid JSON"},"ts":"2024-01-15T10:30:01Z"}

2. Checkpoint Events

Report progress through execution stages.

{"type":"checkpoint","data":{"stage":"start","progress":0.0},"ts":"2024-01-15T10:30:00Z"}
{"type":"checkpoint","data":{"stage":"processing","progress":0.5,"items_processed":50},"ts":"2024-01-15T10:30:05Z"}
{"type":"checkpoint","data":{"stage":"processing","progress":0.5,"state":{"cursor":"page-42"}},"ts":"2024-01-15T10:30:05Z"}
{"type":"checkpoint","data":{"stage":"end","progress":1.0},"ts":"2024-01-15T10:30:10Z"}

Checkpoint state is optional resumable JSON. The orchestrator persists the latest state per run node, rejects states larger than 1 MiB, and injects the last state as context.resume_state plus RESUME_STATE when retrying the node.

3. Metric Events

Emit metrics for monitoring and cost tracking.

{"type":"metric","data":{"name":"tokens_used","value":1500,"unit":"tokens"},"ts":"2024-01-15T10:30:10Z"}
{"type":"metric","data":{"name":"api_latency_ms","value":250},"ts":"2024-01-15T10:30:10Z"}

4. Output Events

Emit structured outputs for downstream nodes in a DAG. The orchestrator captures these and stores them via runstore.SetNodeOutputs(). Downstream nodes access values through the expression environment as inputs.nodeId.key.

{"type":"output","data":{"key":"summary","value":"The quick brown fox..."},"ts":"2024-01-15T10:30:10Z"}
{"type":"output","data":{"key":"word_count","value":150},"ts":"2024-01-15T10:30:10Z"}

Each output event must have key (string) and value (any JSON type) in its data field.

The nested data form above is the canonical protocol and is described by schemas/agent-output-event.schema.json. For migration compatibility, the orchestrator also accepts the historical flat form below and normalizes it to the same stored output:

{"type":"output","key":"summary","value":"The quick brown fox..."}

The flat form is deprecated for new agents. A JSON output event with a missing key, missing value, non-string/empty key, or non-object data payload is not silently treated as output. The orchestrator emits an error event with code INVALID_OUTPUT_EVENT, and the malformed value is excluded from downstream inputs.<node>.<key> data.

5. Result Events

Final output of the agent (optional, can also use exit code + stdout).

{
  "type": "result",
  "data": { "summary": "The quick brown fox...", "word_count": 150 },
  "ts": "2024-01-15T10:30:10Z"
}

6. Error Events

Emit structured errors to classify failures as transient (retryable) or permanent. The orchestrator uses the retryable hint to decide whether to retry the node or fail permanently.

{"type":"error","level":"error","message":"Model not ready","data":{"code":"MODEL_NOT_READY","message":"Model not ready","retryable":true},"ts":"2024-01-15T10:30:10Z"}
{"type":"error","level":"error","message":"Invalid input","data":{"code":"INVALID_INPUT","message":"Missing required field 'text'","retryable":false},"ts":"2024-01-15T10:30:10Z"}

Error fields (inside data):

FieldTypeDescription
codestringMachine-readable error code (e.g., MODEL_NOT_READY, TIMEOUT, INVALID_INPUT)
messagestringHuman-readable description
retryablebooleantrue = transient failure (scheduler retries), false = permanent failure
detailsobjectOptional additional context

Retry contract: When an agent emits {"type":"error","data":{"retryable":true}} and then exits with a non-zero code, the orchestrator rewrites the exit code to 3 (the transient-retry convention), which triggers the node's retry policy. Non-retryable errors leave the exit code unchanged and the node fails permanently.

7. Progress Events

Emit runtime progress for long-running work. Mission Control can render these as node progress bars.

{"type":"progress","level":"info","message":"Processing batch 4/10","data":{"percent":40,"message":"Processing batch 4/10","eta_seconds":18},"ts":"2024-01-15T10:30:10Z"}

Progress fields (inside data):

FieldTypeDescription
percentnumberRequired completion percentage from 0 to 100
messagestringOptional human-readable progress status
eta_secondsnumberOptional estimated seconds remaining
currentnumberOptional current step for compatibility
totalnumberOptional total steps for compatibility

8. Heartbeat Events

Emit heartbeat events during long-running work so the orchestrator can detect stalled agents after the configured node heartbeat timeout.

{"type":"heartbeat","ts":"2024-01-15T10:30:10Z"}

Python SDK

Installation

# The common module is included in the MentatLab repository
# For standalone agents, copy agents/common/emit.py

Basic Usage

#!/usr/bin/env python3
from agents.common.emit import (
    log_info,
    log_error,
    checkpoint,
    emit_error,
    emit_heartbeat,
    emit_progress,
    emit_event,
    set_correlation_id,
)

def main():
    # Set correlation ID for all events (optional, passed via --cid=...)
    set_correlation_id("run-123")

    # Report start
    checkpoint("start", 0.0, {"args": sys.argv[1:]})
    log_info("Agent starting", {"version": "1.0.0"})

    try:
        # Do work...
        result = process_input()

        # Report progress
        emit_progress(percent=50, message="Processing inputs", eta_seconds=10)
        emit_heartbeat()

        # Emit final result
        emit_event(type="result", data={"output": result})
        checkpoint("end", 1.0)

    except Exception as e:
        emit_error(
            "INTERNAL_ERROR",
            str(e),
            retryable=False,
            details={"traceback": traceback.format_exc()},
        )
        log_error(f"Agent failed: {e}", {"traceback": traceback.format_exc()})
        checkpoint("error", 0.0, {"error": str(e)})
        return 1

    return 0

if __name__ == "__main__":
    sys.exit(main())

API Reference

# Set default correlation ID for all subsequent events
set_correlation_id(correlation_id: Optional[str]) -> None

# Emit a raw event
emit_event(
    type: str,                          # Required: event type
    data: Optional[Dict[str, Any]],     # Payload data
    level: Optional[str],               # Log level (for type="log")
    message: Optional[str],             # Human-readable message
    correlation_id: Optional[str],      # Override default correlation ID
    ts: Optional[str],                  # ISO8601 timestamp (auto-generated)
) -> None

# Convenience functions
log_info(message: str, data: Optional[Dict] = None) -> None
log_error(message: str, data: Optional[Dict] = None) -> None
checkpoint(stage: str, progress: float, extra: Optional[Dict] = None, state: Optional[Any] = None) -> None
emit_progress(
    current: Optional[int] = None,
    total: Optional[int] = None,
    percent: Optional[float] = None,
    message: Optional[str] = None,
    eta_seconds: Optional[float] = None,
) -> None
emit_heartbeat() -> None
emit_error(
    code: str,
    message: str,
    retryable: bool = False,
    details: Optional[Dict] = None,
) -> None

Go SDK

package main

import (
    "encoding/json"
    "fmt"
    "os"
    "time"
)

type Event struct {
    Type          string                 `json:"type"`
    Level         string                 `json:"level,omitempty"`
    Message       string                 `json:"message,omitempty"`
    Data          map[string]interface{} `json:"data,omitempty"`
    CorrelationID string                 `json:"correlation_id,omitempty"`
    Timestamp     string                 `json:"ts"`
}

func emit(e Event) {
    if e.Timestamp == "" {
        e.Timestamp = time.Now().UTC().Format(time.RFC3339Nano)
    }
    data, _ := json.Marshal(e)
    fmt.Println(string(data))
}

func logInfo(message string, data map[string]interface{}) {
    emit(Event{Type: "log", Level: "info", Message: message, Data: data})
}

func emitError(code, message string, retryable bool, details map[string]interface{}) {
    payload := map[string]interface{}{
        "code": code,
        "message": message,
        "retryable": retryable,
    }
    if len(details) > 0 {
        payload["details"] = details
    }
    emit(Event{Type: "error", Level: "error", Message: message, Data: payload})
}

func checkpoint(stage string, progress float64, extra map[string]interface{}) {
    data := map[string]interface{}{
        "stage":    stage,
        "progress": progress,
    }
    for k, v := range extra {
        data[k] = v
    }
    emit(Event{Type: "checkpoint", Data: data})
}

func emitOutput(key string, value interface{}) {
    emit(Event{Type: "output", Data: map[string]interface{}{"key": key, "value": value}})
}

func main() {
    checkpoint("start", 0.0, nil)
    logInfo("Processing started", map[string]interface{}{"args": os.Args[1:]})

    // Permanent failure example
    if len(os.Args) < 2 {
        emitError("INVALID_INPUT", "missing required CLI argument", false, nil)
        os.Exit(1)
    }

    // Transient failure example:
    // emitError("MODEL_NOT_READY", "model is still loading", true, map[string]interface{}{"model": "my-model"})
    // os.Exit(1)

    // Do work...
    emitOutput("result", "processed data")

    checkpoint("end", 1.0, nil)
}

TypeScript/Node.js SDK

The TypeScript SDK lives in sdk/typescript and publishes as @mentatlab/agent-sdk. It provides typed NDJSON event helpers plus a createAgent({ onInput, onCancel }) factory for Node.js agents.

import {
  createAgent,
  emitHeartbeat,
  emitOutput,
  emitProgress,
} from "@mentatlab/agent-sdk";

const agent = createAgent({
  agentId: "example.typescript",
  version: "0.1.0",
  async onInput(spec, context, runtime) {
    emitProgress({ percent: 25, message: "Starting work" });

    if (runtime.signal.aborted) {
      return { cancelled: true };
    }

    emitHeartbeat();
    emitOutput("result", { prompt: spec.prompt ?? null });

    return {
      output: `Processed: ${spec.prompt ?? ""}`,
      executionId: runtime.executionId ?? context.execution_id ?? null,
    };
  },
});

agent.run().then((code) => process.exit(code));
cd sdk/typescript
npm install
npm test

Dockerfile Template

FROM python:3.11-slim

WORKDIR /app

# Copy agent code
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Run as non-root user
RUN useradd -m agent
USER agent

# Default entrypoint
ENTRYPOINT ["python", "-m", "my_agent"]

Best Practices

  1. Use slim base images - Reduces image size and attack surface
  2. Run as non-root - Required for security in K8s
  3. No interactive input - Agents must run non-interactively
  4. Flush stdout - Always flush after emitting events
  5. Handle signals - Gracefully handle SIGTERM for cancellation

Agent Registration API

Register an Agent

curl -X POST http://orchestrator:7070/api/v1/agents \
  -H "Content-Type: application/json" \
  -d '{
    "id": "myorg.summarizer",
    "name": "Text Summarizer",
    "version": "1.0.0",
    "image": "registry.example.com/summarizer:v1.0.0",
    "capabilities": ["text-processing", "summarization"]
  }'

List Agents

# List all agents
curl http://orchestrator:7070/api/v1/agents

# Filter by capability
curl "http://orchestrator:7070/api/v1/agents?capabilities=summarization"

Get Agent by ID

curl http://orchestrator:7070/api/v1/agents/myorg.summarizer

Schedule Agent Execution

curl -X POST http://orchestrator:7070/api/v1/agents/schedule \
  -H "Content-Type: application/json" \
  -d '{
    "agent_manifest": {
      "id": "myorg.summarizer",
      "version": "1.0.0",
      "image": "registry.example.com/summarizer:v1.0.0"
    },
    "inputs": {
      "text": "Long text to summarize...",
      "max_length": 200
    },
    "execution_id": "run-12345"
  }'

Debugging Agents

Local Testing

# Run agent locally with test input
echo '{"text": "Hello world"}' | python -m my_agent

# With correlation ID
python -m my_agent --cid=test-run-001

# View NDJSON output
python -m my_agent 2>&1 | jq -c '.'

Common Issues

IssueCauseSolution
Events not appearingstdout not flushedCall sys.stdout.flush() after each emit
Invalid JSON errorsNon-JSON in stdoutEnsure only NDJSON on stdout, use stderr for debug
Agent timeoutLong-running without checkpointsEmit checkpoints regularly
K8s OOMKilledMemory limit exceededSet appropriate resource limits

Viewing Logs in K8s

# View agent job logs
kubectl logs job/agent-run-12345 -n mentatlab

# Follow logs
kubectl logs -f job/agent-run-12345 -n mentatlab

Example Agents

Echo Agent (Python)

The simplest possible agent - echoes input arguments:

#!/usr/bin/env python3
from agents.common.emit import log_info, checkpoint

def main():
    import sys
    args = sys.argv[1:]

    checkpoint("start", 0.0, {"args_count": len(args)})

    for i, arg in enumerate(args):
        log_info("echo", {"index": i, "value": arg})

    checkpoint("end", 1.0)
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

LLM Agent (Python with OpenAI)

#!/usr/bin/env python3
import os
import sys
from openai import OpenAI
from agents.common.emit import log_info, log_error, checkpoint, emit_event

def main():
    checkpoint("start", 0.0)

    prompt = sys.stdin.read() if not sys.stdin.isatty() else " ".join(sys.argv[1:])
    if not prompt:
        log_error("No input provided")
        return 1

    log_info("Processing prompt", {"length": len(prompt)})
    checkpoint("processing", 0.3)

    try:
        client = OpenAI()
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
        )

        result = response.choices[0].message.content
        tokens_used = response.usage.total_tokens

        emit_event(type="metric", data={"name": "tokens_used", "value": tokens_used})
        emit_event(type="result", data={"response": result})

        checkpoint("end", 1.0, {"tokens": tokens_used})

    except Exception as e:
        log_error(f"API call failed: {e}")
        checkpoint("error", 0.0, {"error": str(e)})
        return 1

    return 0

if __name__ == "__main__":
    sys.exit(main())

Architecture

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Frontend      │────▶│    Gateway      │────▶│  Orchestrator   │
│   (React)       │     │    (Go)         │     │    (Go)         │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                                                        │
                                                        ▼
                                               ┌─────────────────┐
                                               │   Kubernetes    │
                                               │   Job Scheduler │
                                               └─────────────────┘
                                                        │
                        ┌───────────────────────────────┼───────────────────────────────┐
                        ▼                               ▼                               ▼
               ┌─────────────────┐             ┌─────────────────┐             ┌─────────────────┐
               │  Agent Pod 1    │             │  Agent Pod 2    │             │  Agent Pod N    │
               │  (container)    │             │  (container)    │             │  (container)    │
               └─────────────────┘             └─────────────────┘             └─────────────────┘
                        │                               │                               │
                        └───────────────────────────────┼───────────────────────────────┘
                                                        │
                                                        ▼ NDJSON events
                                               ┌─────────────────┐
                                               │     Redis       │
                                               │   (Pub/Sub)     │
                                               └─────────────────┘
                                                        │
                                                        ▼
                                               ┌─────────────────┐
                                               │   WebSocket     │
                                               │   Clients       │
                                               └─────────────────┘

Workflow Patterns

Gate Nodes (Manual Approval)

Gate nodes pause execution until an external signal (approve/reject). Use them for human-in-the-loop workflows.

{
  "nodes": [
    { "id": "process", "agent_id": "myorg.processor" },
    {
      "id": "review",
      "type": "gate",
      "gate": {
        "description": "Review output before publishing",
        "timeout": 3600,
        "auto_reject": true
      }
    },
    { "id": "publish", "agent_id": "myorg.publisher" }
  ],
  "edges": [
    { "from": "process", "to": "review" },
    { "from": "review", "to": "publish" }
  ]
}

When the gate node is reached, its status becomes waiting_approval. Approve or reject via the API:

# Approve
curl -X POST http://orchestrator:7070/api/v1/runs/{runId}/nodes/review/approve

# Reject
curl -X POST http://orchestrator:7070/api/v1/runs/{runId}/nodes/review/reject

Webhook Triggers

Create a webhook to allow external systems to trigger runs:

# Create webhook for a flow
curl -X POST http://orchestrator:7070/api/v1/webhooks \
  -H "Content-Type: application/json" \
  -d '{"flow_id": "my-flow-id"}'

# Response includes a token
# {"flow_id": "my-flow-id", "webhook_token": "abc123...", "webhook_url": "/api/v1/webhooks/trigger/my-flow-id"}

# Trigger a run via webhook
curl -X POST http://orchestrator:7070/api/v1/webhooks/trigger/my-flow-id \
  -H "X-Webhook-Token: abc123..." \
  -H "Content-Type: application/json" \
  -d '{"input_params": {"key": "value"}}'

Cron Scheduled Runs

Create cron-style schedules for recurring workflow execution:

# Create a schedule (runs every hour)
curl -X POST http://orchestrator:7070/api/v1/schedules \
  -H "Content-Type: application/json" \
  -d '{
    "flow_id": "my-flow-id",
    "cron": "0 * * * *",
    "input_params": {"source": "scheduled"},
    "enabled": true
  }'

# List schedules
curl http://orchestrator:7070/api/v1/schedules

# Delete a schedule
curl -X DELETE http://orchestrator:7070/api/v1/schedules/{id}

Cron expressions use standard 5-field format: minute hour day-of-month month day-of-week.

Run Cloning

Clone a previous run or create runs directly from flows:

# Clone a run (reuses same plan)
curl -X POST http://orchestrator:7070/api/v1/runs/{runId}/clone \
  -H "Content-Type: application/json" \
  -d '{"auto_start": true}'

# Create a run from a flow
curl -X POST http://orchestrator:7070/api/v1/flows/{flowId}/run \
  -H "Content-Type: application/json" \
  -d '{"timeout": "30m"}'

Per-Node Retry Policies

Configure retry behavior per node with different backoff strategies:

{
  "id": "flaky-api-call",
  "agent_id": "myorg.api-caller",
  "retry_policy": {
    "max_retries": 5,
    "backoff_type": "exponential",
    "backoff_base": 2000000000,
    "backoff_max": 60000000000
  }
}

Supported backoff_type values: fixed, exponential, linear. Durations are in nanoseconds (Go time.Duration).

Run-Level Timeouts

Set a timeout on the entire run plan:

{
  "plan": {
    "nodes": [...],
    "edges": [...],
    "timeout": "30m"
  }
}

The server default is configured via ORCH_DEFAULT_RUN_TIMEOUT (e.g., 30m). Plan-level timeout overrides the default.

See Also

MentatLab Agent SDK | MentatLab docs