Metadata-Version: 2.5
Name: doubleoh
Version: 0.4.0
Summary: The authority layer for AI agents: when an agent is stuck, a person fixes it once and the fix becomes a bounded skill.
License: MIT
License-File: LICENSE
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# doubleoh

**Your agents learn from every fix. Stop failing the same way twice.**

When your agent gets stuck on a real website, one of *your* people fixes it once in a live
browser. That fix compiles into a skill. The next time any agent in your fleet hits the same
wall, it follows the skill instead of asking a human.

```bash
pip install doubleoh
```

Zero dependencies — it is `urllib` and dataclasses.

## The loop, in code

```python
from doubleoh import DoubleOh

doubleoh = DoubleOh(api_key=os.environ["DOUBLEOH_API_KEY"])

# 1. Before your agent tries something it has failed at before, ask what has been learned.
skills = doubleoh.skills_for("check out on the supplier portal")
if skills and skills[0].times_worked > skills[0].times_failed:
    prompt += f"\n\nA colleague has done this before:\n{skills[0].instructions}"

# 2. Your agent runs. When it gets stuck, ask for a human.
fix = doubleoh.request_fix(url=page.url, task="Check out on the supplier portal")
if fix.deflected:
    # Nobody was paged: the fleet already knows this wall. Retry with the skill.
    prompt += fix.skill.instructions
elif fix.fix_url:
    slack.send(f"An agent needs a hand: {fix.fix_url}")
# fix.duplicate_of set? The wall is already waiting on a human — fix.blocked_runs says
# how many runs are stuck on it. Your retry loop needs no change; nobody gets re-paged.

# 3. Close the loop: say whether the skill worked. This is what makes the counts mean
#    something, and a skill followed three times that never works retires itself.
doubleoh.report_skill(skills[0].name, worked=True)
```

You decide what "stuck" means — a timeout, a retry count, a model that says it cannot
proceed. Only you know your agent.

## Guarded tools

The model can't bypass what it never holds. Rules and receipts only bind an agent that asks first, and in a benchmark Claude Haiku 4.5 called change_bank_details directly in three runs of three. A guarded tool asks on every call, whatever the model does: the dangerous function is wrapped, and the model only ever gets the wrapped one.

```python
from anthropic import Anthropic, beta_tool
from doubleoh import DoubleOh

doubleoh = DoubleOh()

@beta_tool
@doubleoh.guard(tool="change_bank_details", action="Change {vendor}'s IBAN to {iban}")
def change_bank_details(vendor: str, iban: str) -> str:
    """Change a vendor's IBAN."""
    return bank.change_details(vendor, iban)

runner = Anthropic().beta.messages.tool_runner(model="claude-opus-5-5", max_tokens=16000,
    tools=[change_bank_details], messages=messages)
```

`client.guard(fn, tool=..., action=..., target=..., amount=..., idempotency_key=..., evidence=..., wait_for_person_seconds=0, agent=None)` returns the same function, guarded, and works as a decorator. It raises `GuardRefused` (its message is the text for the model) and returns `GuardPending` or `GuardReplay` (strings, so any framework hands them to the model) instead of running the function. `action` and `target` are templates over the arguments by name (`{iban}`, `{tool}`, `{args}`) or callables; `amount` returns `(cents, currency)`.

`async def` functions stay async. `on_pending(pending, args)` hands you a waiting call's
arguments: after `client.wait_for_fix(pending.wall_id)`, call the guarded function again with them
and it runs once. The default idempotency key ignores case and whitespace in strings; pass
`idempotency_key` when the same arguments may rightly repeat.

```python
from doubleoh.langchain import guard_tool   # doubleoh.crewai and doubleoh.llamaindex have the same

tools = [
    guard_tool(change_bank_details, action="Change {vendor}'s IBAN to {iban}"),
    guard_tool(pay_invoice, amount=lambda a: (a["amount_cents"], a["currency"])),
    lookup_vendor,   # read-only: left as it is
]
agent = create_react_agent(model, tools)
```

OpenAI Agents SDK: use `guard_tool` in place of `@function_tool` (under a bare `@function_tool`,
a refusal would reach the model as "An error occurred ... Please try again"):

```python
from doubleoh.openai_agents import guard_tool

@guard_tool(action="Change {vendor}'s IBAN to {iban}")
def change_bank_details(vendor: str, iban: str) -> str:
    """Change a vendor's IBAN."""
    return bank.change_details(vendor, iban)

agent = Agent(name="ops", tools=[change_bank_details])
```

## Test without production

```python
# Unit tests: no key, no network. decide plays your rules; client.mock.answer plays the person.
client = DoubleOh(mode="mock", decide=lambda r: "needs_person" if (r.get("amountCents") or 0) > 50000 else "allow")
pay = client.guard(send_payment, amount=lambda a: (a["cents"], "USD"))
pending = pay("Acme", 90000)
assert isinstance(pending, GuardPending)
client.mock.answer(pending.receipt_id, True)
assert pay("Acme", 90000) == "paid"
```

`DOUBLEOH_MODE=dry` runs any agent against your real rules with nothing done: each guarded call
returns a `GuardDryRun` ("Dry run, not done: ... would wait for a person (...)") and the function
never runs. `DOUBLEOH_MODE=mock` is the same as `mode="mock"`.

## API

- `skills_for(task) -> list[LearnedSkill]` — what has been learned; put `instructions` in
  your agent's context. `times_worked` / `times_failed` are raw counts on purpose: 5 for 5
  is worth following, 1 for 1 is a guess that happened to work once.
- `request_fix(url, task) -> Intervention` — ask for help. Three possible shapes: a
  `fix_url` (a human was paged), `deflected=True` with `skill` (nobody was paged — you
  already know the answer), or `duplicate_of` with `blocked_runs` (already waiting; triage
  by the count). `url` must be a public page; private and loopback addresses are refused.
- `report_skill(name, worked)` — one boolean; wire it to whatever already tells you a run
  succeeded.
- `intervention(id)` / `interventions()` — status; `skill_name` is the receipt.
- `retire_skill(name, reason)` — stop serving a skill immediately.

Errors raise `DoubleOhError` carrying the server's own message, the HTTP `status`, a stable
`code` and `next`, the move to make. An unreachable service is `code == "unreachable"`, status 0.

## What is recorded

Fix sessions keep screenshots and the shape of each action. **Typed text is never
recorded** — a fix session is exactly where a password gets typed, so keystrokes stay off
the record entirely.

## Framework adapters

Every adapter serves the same three tools; pick your framework's line.

```python
from doubleoh import DoubleOh
client = DoubleOh(api_key=os.environ["DOUBLEOH_API_KEY"])

# LangChain / LangGraph
from doubleoh.langchain import doubleoh_tools
graph = create_react_agent(model, tools=doubleoh_tools(client))

# CrewAI
from doubleoh.crewai import doubleoh_tools
agent = Agent(role="operator", tools=doubleoh_tools(client))

# LlamaIndex
from doubleoh.llamaindex import doubleoh_tools
agent = ReActAgent.from_tools(doubleoh_tools(client), llm=llm)

# OpenAI Agents SDK, AutoGen, Pydantic AI — all accept plain functions:
from doubleoh.tools import build_tools
tools = build_tools(client)  # skills_for, request_fix, wait_for_fix, report_skill, check, open, confirm
agent = Agent(tools=[function_tool(f) for f in tools])
```

No framework is a dependency of this package: each adapter imports its framework lazily
and, if it is missing, raises one sentence naming what to `pip install`.

Using Claude or another MCP-capable agent? Skip the SDK entirely — `doubleoh-mcp` serves
the same tools over MCP.
