esoul 0.7.0

Build, run, and durably resume multi-agent networks on ExternalSoul from Python — event-sourced, audit-logged workspace automation.

esoul — build agent networks in Python

esoul is the official Python SDK for the ExternalSoul platform. Its headline surface is the agent-network builder: declare a multi-agent network in a few lines of Python, get a clean auto-laid-out graph, create it in a workspace, run it, fan it out over your data, and durably resume it if it fails partway — no hand-written JSON, no overlapping nodes.

pip install esoul

import esoul
from esoul.agent_builder import AgentNetwork

client = esoul.Esoul()                         # auto-detects credentials

# Declare an agent network — nodes + edges, no positions, no JSON blob.
net = AgentNetwork("Researcher", model="anthropic/claude-3-5-haiku-20241022")
trigger = net.trigger()                        # fires on agent_builder/run
agent   = net.agent("Researcher", system_prompt="Answer concisely, then save to Notes.")
notes   = net.tool_binding("My Notes", application_type="note_editor")
trigger >> agent                               # wire edges with `>>`
agent   >> notes

# Auto-laid-out (no overlapping cards) + created as a workspace app.
app = net.create(client, workspace_id="ws_abc", instance_name="researcher")

# Run it and wait for the result.
result = client.agents.invoke(
    app.node_id, workspace_id="ws_abc",
    input="Summarise the Q3 metrics into My Notes.",
).wait()
print(result.text)

---

The builder — esoul.agent_builder.AgentNetwork

AgentNetwork produces the exact state the agent-builder UI persists, so a code-built network opens, edits, and runs identically to a hand-built one.

Nodes

Each constructor returns a NodeRef (its .id is the generated node id):

Method	Node
net.trigger(event=…, prompt_template=…, run_mode=…)	a trigger — what the network waits for (default agent_builder/run)
net.agent(name, system_prompt=…, model=…, description=…, context_apps=…)	an agent
net.tool_binding(app, application_type=…)	a Tool Binding — gives an agent a workspace app's full toolkit
net.workspace_tool(tool_names, label=…)	a Workspace Tool node (e.g. ["return_result_to_router"])
net.sub_agent_mapper(sub_agent_app_node_id=…, source_mode=…, source_config=…)	a Sub-Agent Mapper (fan-out — see below)
net.add_node(node_type, data, node_id=…)	generic escape hatch for any other node type (image_edit_agent, search, open_app, …)

Node ids are unique per network (a short token is stamped in), so two code-built networks never collide.

Edges

trigger >> agent >> mapper        # chainable; `a >> b` returns b
net.connect(router, billing, routing_prompt="billing questions")   # agent→agent routing label
net.connect(driver, mapper, mapper_meta={"iterationMode": "per_row", "label": "fill each row"})

Models

model= accepts a gateway string ("anthropic/claude-3-5-haiku-20241022"), a {"provider", "model"} dict, or None (network default). Set it per-network (AgentNetwork(..., model=…)) or per-agent (net.agent(..., model=…)).

Layout, serialise, create

net.auto_layout()          # clean, non-overlapping positions (also runs inside to_state)
state = net.to_state()     # the raw dict — pass to apps.create(initial_state=...) yourself
app   = net.create(client, workspace_id=ws, instance_name="my-net")   # …or one call

The layout is a pure, deterministic layered (Sugiyama-style) placement; from esoul.agent_builder import assert_no_overlap lets your tests prove a graph is clean.

---

Fan out over data — the Sub-Agent Mapper

Point a sub_agent_mapper node at a spreadsheet column and it runs one sub-agent per cell, in parallel:

sub  = AgentNetwork("Filler", model="anthropic/claude-3-5-haiku-20241022")
t = sub.trigger(); a = sub.agent("Filler", system_prompt="Fill one cell.")
t >> a; a >> sub.tool_binding("Cities", application_type="spreadsheet")
sub_app = sub.create(client, workspace_id=ws, instance_name="filler")

main = AgentNetwork("Driver", model="anthropic/claude-3-5-haiku-20241022")
driver = main.agent("Driver", system_prompt="Hand off to the mapper.")
mapper = main.sub_agent_mapper(
    sub_agent_app_node_id=sub_app.node_id,
    source_mode="spreadsheet",
    source_config={"spreadsheetName": "Cities", "inputColumnName": "City", "filters": []},
)
main.connect(driver, mapper, mapper_meta={"iterationMode": "per_row"})
main_app = main.create(client, workspace_id=ws, instance_name="driver")

client.agents.invoke(main_app.node_id, workspace_id=ws, input="go").wait()
inv = client.mapper.get_latest(workspace_id=ws, mapper_node_id=mapper.id)
print(inv.completed_count, "/", inv.expected_count)   # e.g. 100 / 100

Durable resume — recover an interrupted run without redoing work

A 100-sub-agent run that drops mid-flight (network error, a batch that errored) doesn't have to start over. Per-item state lives server-side, so resume re-dispatches only the unfinished items:

res = client.mapper.resume(inv.id, retry="incomplete")
#                                  ^ heuristic for which items re-run:
#   "incomplete" (default): everything not completed  — resume without loss
#   "errored":   only failures — retry, leave good work
#   "stalled":   only queued/running — a run that died mid-flight
print(res.resumed, res.pending)        # pending = items still not completed

Completed work is never redone; safe to call repeatedly.

---

Test agents — esoul.evals

Hermetically test an agent network before you publish: run it on graded inputs, grade with contains / regex / jsonSchema / llmJudge, and assert structural invariants over the event log. Same engine as the in-UI EVALS tab.

from esoul.evals import CastMember, run_eval, make_agent_judge

res = run_eval(
    client, workspace_id=ws,
    cast=[CastMember("pos", agent.node_id, "I love this!")],
    cases=[{"id": "p", "target": "pos", "expected": [{"graderId": "contains", "expected": "POSITIVE"}]}],
)
print(res.summary())   # PASS/FAIL + per-case ✓/✗

Use Haiku for tests; one case = one CastMember + one case spec. Agent output that ends on return_result_to_router is read from router_handoff (handled for you). To surface results in the in-UI EVALS tab, persist Eval/EvalRun/ EvalCaseResult rows keyed by agentRef.agentNodeId. Full how-to (graders, judge, invariants, persistence, the EVALS-tab hollow-circle meaning, ops): docs/esoul-llm-reference.md → "Testing harness". Worked example: tests/fairy_comprehensive.py + ../../scripts/eval-tests/persist-fairy-evals.ts.

---

Beyond agents: events + typed resources

ExternalSoul workspaces are event-sourced — every mutation is a typed event on a per-workspace timeline. The UI, agents, and your scripts all dispatch through the same reducers, so audit logging is automatic. Drop to the raw event layer or the typed resources whenever you need to:

client.describe()                               # what workspaces / apps / events exist
client.dispatch_event(app_id=…, event_name="spreadsheet_add_row", event_data={…})

Resource	What it does
client.spreadsheet	Create / seed / read spreadsheets — create, add_column, add_rows, set_cell, get_rows.
client.workspaces.apps	App lifecycle — list, create (with initial_state), rename, delete.
client.agents	Invoke agent_builder runs — invoke(...).wait().
client.mapper	Sub-Agent Mapper introspection (list_invocations, get_latest) + resume (resume, resume_latest).
client.questions	Human-in-the-loop — ask(...) against the workspace question queue.
client.drive	Google Drive ops on the workspace's connected Drive.
client.revert	Run-revert kernel (preview).

MCP server — drive ExternalSoul from Claude Desktop / Cursor / any MCP client

The whole SDK is also available as a Model Context Protocol server, so any MCP client can read app state, dispatch events, run and build agent networks, search, manage the human-in-the-loop queue, and scrub the timeline — using your PAT.

pip install "esoul[mcp]"
ESOUL_TOKEN=esoul_pat_… esoul-mcp                    # stdio (default)
ESOUL_TOKEN=… esoul-mcp --transport streamable-http --port 8765
ESOUL_TOKEN=… ESOUL_MCP_READONLY=1 esoul-mcp         # browse, don't touch

Add to your client (claude_desktop_config.json → mcpServers):

{
  "mcpServers": {
    "externalsoul": {
      "command": "esoul-mcp",
      "env": {
        "ESOUL_TOKEN": "esoul_pat_…",
        "ESOUL_WORKSPACE_ID": "6d735936-…"
      }
    }
  }
}

49 tools across every resource (21 in read-only mode). Set ESOUL_WORKSPACE_ID to omit workspace_id on most calls. Long-running work (agent runs, HIL questions) is fire-then-poll so calls stay short. Full reference: .cursor/rules/docs/esoul-mcp-server.md.

Authentication

esoul.Esoul() auto-detects credentials in this order:

1. Explicit token= kwarg
2. ESOUL_TOKEN environment variable
3. /var/run/esoul/token (the 0600 token every E2B sandbox writes at boot — Esoul() inside a sandbox Just Works)
4. ~/.config/esoul/credentials (TOML with a [default] section: token = "esoul_pat_...")

For off-platform use (laptop, CI), create a Personal Access Token in workspace settings → "Access Tokens", pick the workspaces it can mutate, then:

export ESOUL_TOKEN="esoul_pat_..."

Workspace isolation

A credential is scoped to specific workspaces. Cross-workspace dispatch is structurally impossible — a server invariant, not a check in your script.

Idempotency

Every write is automatically idempotent — the SDK generates a UUID per call and reuses it across retries; the server caches the response under (session, key) for 24h. Override with idempotency_key="…" for application-level retry control.

Async client

esoul.AsyncEsoul() mirrors the sync API exactly:

import asyncio, esoul

async def main():
    async with esoul.AsyncEsoul() as client:
        await client.mapper.resume_latest(workspace_id="ws_abc", mapper_node_id="…")

asyncio.run(main())

Typed errors

Failures surface as exception classes mapped from the server's error.code; esoul.EsoulError catches everything:

try:
    client.dispatch_event(...)
except esoul.WorkspaceAccessDenied as e:
    print(f"Token cannot reach workspace {e.details['workspaceId']}")
except esoul.RateLimitError as e:
    time.sleep(e.retry_after_seconds or 1)
except esoul.AuthError:
    print("Token expired or revoked")
except esoul.EsoulError as e:
    print(f"{e.code}: {e.message}")

Status

Beta. The agent-builder, mapper, spreadsheet, and low-level dispatch_event / read_state surfaces are stable and used in production automations. See CHANGELOG.md for release notes.

License

MIT.