Metadata-Version: 2.5
Name: agent-outbox
Version: 0.1.1
Summary: Idempotent execution layer for LLM tool calls: outbox + dedupe keys.
Project-URL: Homepage, https://github.com/adp811/agent-outbox
Project-URL: Repository, https://github.com/adp811/agent-outbox
Project-URL: Issues, https://github.com/adp811/agent-outbox/issues
Author-email: Aryan Patel <aryanpatel02@icloud.com>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,idempotency,langgraph,llm,outbox,tools
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.11
Provides-Extra: langgraph
Requires-Dist: langgraph>=0.6.0; extra == 'langgraph'
Description-Content-Type: text/markdown

# 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.

[![CI](https://github.com/adp811/agent-outbox/actions/workflows/ci.yml/badge.svg)](https://github.com/adp811/agent-outbox/actions/workflows/ci.yml)

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

Until the first PyPI release lands, install from GitHub:

```bash
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:

```bash
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`:

```python
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:

| arg | meaning |
|---|---|
| `tool` | name, part of the key |
| `fn` | the function to call once |
| `args` | kwargs passed to `fn` (key order does not matter) |
| `session_id` | one agent run / conversation |

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:

```python
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

```python
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

```bash
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](https://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`.

