Epsilon SDK¶

The Epsilon SDK lets you keep Epsilon's orchestration layer and swap in your own implementation. That implementation can be anything — a LangChain agent, a LlamaIndex pipeline, a plain Python function, or an external process. Epsilon does not care what runs inside the adapter. It only cares that the adapter follows a small contract.

The contract¶

Your adapter does three things:

Read the task from input["task"]
Write files into input["workspace"]
Return a result dict with status and summary

That is the entire interface. If your code can do those three things, it works with Epsilon.

Smallest example¶

def run(input, **kwargs):
    task = input["task"]
    workspace = input["workspace"]

    # do your work
    result = my_agent.execute(task)

    # write output
    from pathlib import Path
    Path(workspace, "output.txt").write_text(result)

    return {"status": "ok", "summary": "wrote output.txt", "artifact": "output.txt"}

Run it:

epsilon runs create \
  --topology dag \
  --task "Write hello.txt with hello world" \
  --implementation python:my_adapter.py:run

Everything else — topology, coordination, retries, logging, artifacts — is handled by Epsilon. Your adapter just does the work.

What your adapter receives¶

The input dict contains:

Field	Type	Description
`task`	`str`	The task description for this step
`workspace`	`str`	Path to the shared workspace directory. Write your output files here.
`agent_id`	`str`	Unique identifier for this agent instance
`topics`	`list[str]`	Message subscription topics (for multi-agent messaging)

What your adapter should return¶

Return a dict with at least status and summary:

{
    "status": "ok",          # "ok" or "error"
    "summary": "wrote result.md",
    "artifact": "result.md",  # optional: name of the main output file
}

If your function returns a plain string, Epsilon wraps it automatically.

Registering implementations¶

If you plan to reuse an adapter, register it once and refer to it by name:

epsilon implementations register my-worker \
  --from python:examples/epsilon_sdk/function_adapter_example.py:run \
  --description "Function-based starter adapter"

# now use it by name
epsilon runs create \
  --topology dag \
  --task "Write hello.txt with hello world" \
  --implementation my-worker

Manage registered implementations:

epsilon implementations list
epsilon implementations get my-worker

Using the session object¶

Your adapter receives an optional session parameter — an AdapterSession instance that provides logging and messaging:

def run(input, *, session=None, **kwargs):
    if session:
        session.log("starting work")

    # do work...

    if session:
        session.send_message("ready for review")

    return {"status": "ok", "summary": "done"}

You do not need to use session. It is there if you want logging or inter-agent messaging.

Starter files¶

Copy one of these and modify it:

Function-style (use with python:path/to/file.py:run):

examples/epsilon_sdk/function_adapter_example.py — minimal starter
examples/epsilon_sdk/langchain_simple_chat.py — LangChain chat
examples/epsilon_sdk/langchain_workspace_file_agent.py — LangChain with file output
examples/epsilon_sdk/llamaindex_simple_chat.py — LlamaIndex chat
examples/epsilon_sdk/llamaindex_workspace_file_agent.py — LlamaIndex with file output

Process-style (use with python3 path/to/file.py, no :run suffix):

examples/epsilon_sdk/process_adapter_example.py — stdin/stdout JSON protocol

Framework guides¶

Step-by-step integration guides for specific frameworks:

Going deeper¶

The adapter protocol, environment variables, and architecture details are in the Technical Reference.