Skip to main content
This is a custom harness example. It runs Anthropic’s Claude Agent SDK inside an E2B sandbox. Dari owns the session, sandbox, events, and user-facing API. Your runtime server owns the agent loop and calls Claude Agent SDK directly. The runtime stores Claude’s SDK session ID in Dari state. On the next Dari turn, it passes that ID back to the SDK with resume, so Claude keeps its own conversational memory across messages in the same session.

​
Files

dari.yml
name: claude-sdk-runtime
harness:
  kind: customRuntimeServer
  protocol: dari-runtime-v1
  customRuntimeServer:
    command: sh -c 'uvicorn custom_runtime_server:app --host 0.0.0.0 --port "${PORT:-8787}"'
    port: 8787
    health_path: /health
    advance_path: /runtime/advance

instructions:
  system: prompts/system.md

sandbox:
  provider: e2b
  internet_access: true
  secrets:
    - ANTHROPIC_API_KEY
  setup:
    script: setup.sh

built_in_tools: []
custom_tools: []
setup.sh
#!/usr/bin/env bash
set -euo pipefail

python3 -m pip install --quiet claude-agent-sdk fastapi uvicorn
npm install -g @anthropic-ai/claude-code
prompts/system.md
You are a helpful agent running through Claude Agent SDK inside a Dari custom runtime server.
custom_runtime_server.py
from __future__ import annotations

import json
from typing import Any

from fastapi import FastAPI, Request

MODEL = "claude-sonnet-4-5"

app = FastAPI()


@app.get("/health")
async def health() -> dict[str, bool]:
    return {"ok": True}


@app.post("/runtime/advance")
async def advance(request: Request) -> dict[str, Any]:
    payload = await request.json()
    state = state_json(payload)

    text, claude_session_id = await ask_claude(
        user_text(payload.get("input")),
        resume=string_value(state, "claude_session_id"),
    )

    next_state = {
        "turns": int_value(state, "turns") + 1,
        "claude_session_id": claude_session_id,
    }

    return {"text": text, "state": next_state}


async def ask_claude(prompt: str, *, resume: str | None) -> tuple[str, str | None]:
    from claude_agent_sdk import ClaudeAgentOptions, query

    options = ClaudeAgentOptions(
        model=MODEL,
        max_turns=1,
        allowed_tools=[],
        cli_path="claude",
        permission_mode="dontAsk",
        resume=resume,
    )

    chunks: list[str] = []
    session_id = resume

    async for message in query(prompt=prompt, options=options):
        session_id = message_session_id(message) or session_id
        chunks.extend(message_text(message))

    return "".join(chunks).strip(), session_id


def state_json(payload: dict[str, Any]) -> dict[str, Any]:
    state_ref = payload.get("state_ref")
    if not isinstance(state_ref, dict):
        return {}
    value = state_ref.get("json")
    return value if isinstance(value, dict) else {}


def user_text(input_payload: object) -> str:
    if not isinstance(input_payload, dict):
        return ""
    if input_payload.get("type") != "user_message":
        return json.dumps(input_payload)

    message = input_payload.get("message")
    content = message.get("content") if isinstance(message, dict) else None
    if not isinstance(content, list):
        return ""

    parts: list[str] = []
    for item in content:
        if isinstance(item, dict) and isinstance(item.get("text"), str):
            parts.append(item["text"])
    return "\n".join(parts)


def message_text(message: Any) -> list[str]:
    content = getattr(message, "content", None)
    if not isinstance(content, list):
        return []

    parts: list[str] = []
    for item in content:
        text = getattr(item, "text", None)
        if isinstance(text, str):
            parts.append(text)
    return parts


def message_session_id(message: Any) -> str | None:
    value = getattr(message, "session_id", None)
    return value if isinstance(value, str) else None


def string_value(data: dict[str, Any], key: str) -> str | None:
    value = data.get(key)
    return value if isinstance(value, str) else None


def int_value(data: dict[str, Any], key: str) -> int:
    value = data.get(key)
    return value if isinstance(value, int) else 0

​
Deploy And Run

Store the Anthropic key before deploying: 
dari credentials add ANTHROPIC_API_KEY
Then deploy and send messages normally: 
dari deploy .
dari session create --agent <agent_id>
dari session send <session_id> "Remember this value: abc123."
dari session send <session_id> "What value did I ask you to remember?"
Dari passes the runtime’s returned state.json back as state_ref.json on the next advance. If a response omits state, the previous state is kept; "state": null clears it. In this example, claude_session_id is the memory handle used to resume the same Claude SDK conversation. If you need memory to survive a completely new sandbox after hard loss, persist the Claude transcript with the SDK’s session-store support or store your own compact transcript in state.