agent-outbox

Wrap a tool so an agent can retry it without running the side effect twice.

Same session + tool + args → first call runs, later calls replay.

uv add agent-outbox
# or
pip install agent-outbox

Until the first PyPI release lands, install from GitHub:

uv add git+https://github.com/adp811/agent-outbox

LangGraph extra (only needed for the sample graph): uv add "agent-outbox[langgraph]"

From this repo:

uv sync
uv run agent-outbox-demo    # retries create_ticket 5 times, 1 ticket
uv run agent-outbox-graph   # same storm, as a LangGraph agent → tools loop
uv run agent-outbox-eval    # 23/23 retry-storm cases

Usage

The tool is a normal function or client method. You wrap the call with run:

from agent_outbox import ToolExecutor, OutboxStore

store = OutboxStore()  # or OutboxStore("outbox.db") to persist
ex = ToolExecutor(store)

class TicketClient:
    def create_ticket(
        self,
        title: str,
        severity: str,
        idempotency_key: str | None = None,
    ) -> dict:
        return httpx.post(
            "https://tickets.example/v1/tickets",
            json={"title": title, "severity": severity},
            headers={"Idempotency-Key": idempotency_key},
        ).json()

tickets = TicketClient()
args = {"title": "kserve replica crashloop", "severity": "high"}
session = "agent-session-1"

# first call: POSTs
a = ex.run("create_ticket", tickets.create_ticket, args, session)

# agent retries the same call: no second POST, same payload
b = ex.run("create_ticket", tickets.create_ticket, args, session)

run takes:

Returned Execution: ok, replay, result, error.

A new session, tool name, or arg value is a new call and will fire again.

If fn has an idempotency_key parameter, run fills it in so the downstream API can also dedupe.

LangGraph tool node

The graph is a normal agent → tools loop. The agent is scripted (it keeps emitting the same create_ticket call — no LLM). The tools node is the wrap:

from langgraph.graph import END, START, StateGraph
from agent_outbox import ToolExecutor, OutboxStore

def tools(state):
    execution = ex.run(
        "create_ticket",
        tickets.create_ticket,
        {"title": state["title"], "severity": state["severity"]},
        state["session_id"],
    )
    return {"replay": execution.replay, "result": execution.result}

graph = StateGraph(AgentState)
graph.add_node("agent", agent)   # decides to retry
graph.add_node("tools", tools)   # outbox absorbs the storm
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route, {"tools": "tools", "__end__": END})
graph.add_edge("tools", "agent")

Five trips through tools, one create_ticket side effect. uv run agent-outbox-graph runs it.

If the tool fails

The first attempt is recorded as failed. Retries return that error and do not call fn again.

If a retry lands while the first call is still running

Waiters poll the outbox until the owner finishes (default wait_timeout=10s) and then replay the same result. They do not start a second side effect.

If the process dies mid-call

Pending rows have a lease (lease_seconds=30 by default). Heartbeats refresh it while fn is running. After the lease expires, another worker can reclaim and run fn again.

That reclaim is at-least-once: if the original worker already performed the side effect and crashed before complete(), a reclaim will fire twice. Downstream Idempotency-Key is how you close that gap.

Optional: write spans

from agent_outbox import JsonlTracer, ToolExecutor, OutboxStore

tracer = JsonlTracer("spans.jsonl")
ex = ToolExecutor(OutboxStore("outbox.db"), tracer)

Each attempt logs tool, key, session_id, replay, status.

Tests

uv sync
uv run pytest
uv run agent-outbox-eval

CI runs pytest, the 23-case eval, and a wheel install on Python 3.11–3.13.

Release to PyPI

CI is automatic on push. Publishing is a GitHub Release (v0.1.0).

One-time PyPI setup: pypi.org → Publishing → pending publisher. Project name agent-outbox, owner adp811, repo agent-outbox, workflow publish.yml. Then create a GitHub release tagged v0.1.1.