Metadata-Version: 2.4
Name: autoagent-core
Version: 0.11.0
Summary: LLM agent library with tool use, dynamic tool generation, and software evolution capabilities
License: MIT
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: jsonschema>=4.0
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20; extra == "otel"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: pytest-timeout>=2.1; extra == "dev"
Requires-Dist: hypothesis>=6.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: mypy>=1.0; extra == "dev"
Dynamic: license-file

# autoagent

**A minimal, auditable LLM agent core for Python. Not a framework.**

`autoagent` is a small library that gives you the agent loop — LLM ↔ tools, done right —
and the safety rails around it, without pulling in a framework's worldview. You can read
the entire core in an hour, and every boundary your agent has is **Python code you wrote**,
not a prompt you hope it respects.

```python
from autoagent import Agent

agent = Agent.from_model("gemini", "gemini-3.5-flash")

@agent.tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

result = agent.run("What is 21 + 21?")
print(result.output)   # "42"
```

That's the whole API for the common case. The JSON schema for `add` is generated from the
type annotations. The loop, retries, and provider wire formats are handled for you.

Need live output? Streaming is a plain sync iterator — no async ceremony:

```python
for ev in agent.run_stream("What is 21 + 21?"):
    if ev.type == "text":
        print(ev.text, end="", flush=True)      # token deltas as they arrive
    elif ev.type == "tool_start":
        print(f"\n[calling {ev.tool_name}…]")
```

---

## Why another agent library?

Because most of them are **frameworks**: they want to own your prompts, your memory, your
control flow, and your dependency tree. When something misbehaves at 2 a.m., you're reading
someone else's abstraction stack instead of your own code.

`autoagent`'s theses:

1. **The agent must be readable.** The core loop is a few hundred lines of plain Python.
   No metaclasses, no callbacks-on-callbacks, no YAML.
2. **Bounding is code, not prompts.** File access goes through `ProjectWorkspace`
   (allowlists, anti path-traversal, write history + rollback). Generated tools run in a
   **sandbox** (Docker isolation when available, hardened AST denylist otherwise) and are
   promoted to native only through a **hash-based manifest a human approves**.
3. **Zero dependencies for the core.** Python ≥ 3.10, `urllib`, `dataclasses`. No SDKs —
   each provider adapter speaks the wire format directly (~100 lines each).
4. **Multi-provider without ceremony.** OpenAI, Anthropic, DeepSeek, Gemini. A provider is
   one method: `complete(LLMRequest) -> LLMResponse`. Write your own in 50 lines.
5. **Synchronous and deterministic — with real streaming.** The loop is sync by design —
   *you* choose your concurrency model (`threading`, `asyncio.to_thread`, a queue). No
   colored functions imposed on your codebase. Streaming is a plain **sync iterator**:
   `for event in agent.run_stream(prompt): …` yields text deltas and tool events as they
   happen (SSE under the hood, all four providers).

## What's in the box

| Capability | What it gives you |
|---|---|
| **Tool schema autogen** | `@agent.tool` reads annotations + docstring → strict JSON schema (`Literal` → enums, `additionalProperties: false` by default) |
| **`ProjectWorkspace`** | bounded reads/writes: extension allowlists, anti path-traversal, change history, rollback |
| **Dynamic tools + sandbox** | the agent can write its own tools; they run in Docker (or a hardened subprocess), never natively without human promotion via a hash manifest |
| **`TraceEmitter`** | typed lifecycle events (`run_start`, `tool_call_*`, `run_end`…) with span/parent IDs → JSONL and/or callback; **secret redaction built in** (Bearer tokens, API keys never hit your logs) |
| **Streaming** | `run_stream()` / `run_messages_stream()` yield `StreamEvent`s (text deltas, `tool_start`/`tool_end`, corrections) as plain sync iterators — SSE wire streaming in all four providers |
| **`Memory` protocol** | two methods (`compact`, `recall`) — duck-typed, `@runtime_checkable`. Built-ins: `BufferMemory` (hard cap) and **`SummarizingMemory`** (old turns folded into an *incremental* LLM summary — bounded context without losing established facts — plus dependency-free lexical `recall`). Ship your own vector store for semantic recall; `agent.register_recall_tool()` lets the agent query its memory as a tool |
| **`RoutingProvider`** | per-request dispatch across providers: text-only turns go to a cheap model, turns carrying images route to a vision-capable one — behind the standard provider interface, invisible to `Agent` |
| **`post_turn_hook`** | host-side verification loop: inspect what the agent did, inject a correction message, hard-capped iterations — validation in code, not vibes |
| **`cancel_token`** | cooperative cancellation between LLM turns (`threading.Event`) |
| **Multimodal** | `ImageAttachment` on messages, serialized per provider |
| **`agent.as_tool()`** | the minimal multi-agent primitive: expose an agent as a tool of another (supervisor/specialist in two lines); stateless delegation, failures surface as tool errors, delegation cost (`tokens`, `steps`) reported to the parent |
| **`token_budget`** | hard cap on a run's cumulative token usage (`TokenBudgetExceeded`); per-call `TokenUsage` on results and `done` events — never invented, only what the provider reports |
| **`parallel_tool_calls`** | opt-in: multiple tool calls in one turn execute on a thread pool (I/O-bound latency win), transcript order stays deterministic |
| **`Orchestrator`** | for host-driven flows (guided forms, CATI questionnaires): *your* state machine decides every step; the LLM only interprets answers and rephrases prompts |
| **`MCPClient`** | mount any MCP server's tools as ordinary agent tools (stdio transport, zero dependencies): `mcp.mount(agent, prefix=, include=)` — server schemas validated by the registry before every call |
| **`OTelTraceExporter`** | the trace tree becomes real OpenTelemetry spans (`agent.run → llm → tool.<name>`) for Jaeger / Tempo / Langfuse / Phoenix; `opentelemetry-api` is an *optional* extra, the core stays dependency-free |
| **`RunState` + `resume`** | durable runs: a JSON checkpoint after every completed step; resume after a crash, a restart, or past a raised `max_steps` / `token_budget` (`exc.state` is ready to resume) |
| **`tool_policy`** | one hook for allow / deny / **human approval** / quota-audit, checked for every tool call *before* any side effect; a crashing policy denies (fail-closed); `ApprovalRequired` pauses the run with a resumable snapshot |
| **`EvolutionRuntime`** | let an agent modify a live project: read state, propose a module, run validation, roll back on failure |

## Quickstart

```bash
pip install autoagent-core            # installs as `import autoagent`
# or from source:
git clone <this-repo>
cd autoagent
pip install jsonschema        # the core's only extra; examples may need more
export GEMINI_API_KEY=...     # or OPENAI_API_KEY / ANTHROPIC_API_KEY / DEEPSEEK_API_KEY
python examples/demo_autoagent.py      # the demo below (French prompts/comments)
```

The demo that carries the argument — one scenario, two files:

**[`examples/demo_autoagent.py`](examples/demo_autoagent.py)** (55 lines of code) is a
**three-agent hierarchy**: an orchestrator delegates two log files to two specialist
agents (`as_tool()`), **cross-checks their findings against each other** (each
gateway-502 server error must match one FAILED payment), audits the raw files with its
own tools when in doubt, then **saves its validated report through a `ProjectWorkspace`**
fenced to `_out/` + markdown only. The whole delegation tree lands in one trace file via
a shared `TraceEmitter` — and the script ends by proving the fence deterministically,
attempting what an agent might: `../demo_autoagent.py` → *Path escapes workspace*,
`virus.exe` → *extension blocked*, `C:/Windows/x.md` → *absolute paths not allowed*.
Boundaries you can demo, because they're code.

**[`examples/demo_pure_python.py`](examples/demo_pure_python.py)** (164 lines of code) is
the **same system with no library** — same model, same three agents, same validated
answer. Everything the library did for free, hand-rolled and annotated: the generic agent
loop, every tool schema, the provider wire format + retries, delegation with its failure
contract, the write fence, and a trace tree (without secret redaction).

| same behavior, same answer | with autoagent | pure Python |
|---|---|---|
| lines of code | **55** | **164** |
| …that you must maintain per provider | no (4 providers included) | yes |

### A tool-using agent with memory, tracing, and a verification hook

```python
import threading
from autoagent import Agent, BufferMemory, Message, ModelConfig, TraceEmitter, create_provider

provider = create_provider(ModelConfig(provider="openai", model="gpt-4o-mini"))

def must_have_saved(ctx) -> Message | None:
    """Host-side check: force another turn if the agent never wrote the file."""
    if not any(tc.name == "write_file" for tc in ctx.tool_calls):
        return Message(role="user", content="You never called write_file. Save your work.")
    return None

with TraceEmitter(file="run.jsonl") as trace:
    agent = Agent(
        provider,
        system_prompt="You are a careful refactoring assistant.",
        max_steps=12,
        memory=BufferMemory(max_messages=30),
        trace=trace,
        post_turn_hook=must_have_saved,
        max_corrections_per_run=1,
    )
    result = agent.run("Refactor ./api.py", cancel_token=threading.Event())

print(result.output, result.steps)
```

### Verdicts as tools (a pattern we use in production)

Instead of asking the model for "strict JSON" and parsing it with regexes, expose the
decision as tools — the model *cannot* answer malformed:

```python
verdict = {"decided": False}

@agent.tool
def approve(reason: str = "") -> dict:
    """The last exchange is fine."""
    verdict.update(decided=True, ok=True, reason=reason)
    return {"recorded": True}

@agent.tool
def request_fix(problem: str, instruction: str) -> dict:
    """Something is actually wrong; give the voice agent a short corrective instruction."""
    verdict.update(decided=True, ok=False, problem=problem, instruction=instruction)
    return {"recorded": True}

agent.run(state_and_transcript)
if verdict.get("ok") is False:
    deliver_instruction(verdict["instruction"])
```

This exact pattern supervises a real-time **phone bot** (Gemini Live voice loop): the
supervisor agent runs in a thread, checks every turn against the call state and the
caller's past-call memory (via `register_recall_tool`), and injects corrections — without
ever blocking the audio. The sync core made that trivial: `asyncio.to_thread(supervisor.review, …)`.

### Compose agents in two lines (`as_tool`)

No crew DSL, no choreography framework — an agent is just a tool of another agent:

```python
expert = Agent(cheap_provider, system_prompt="You are a traffic-count analyst…")
supervisor.add_tool(expert.as_tool(
    name="analyze_counts",
    description="Delegate traffic-count questions to the analyst.",
))
# Delegation is stateless, failures surface as tool errors (never crash the parent),
# and the parent sees the cost: {"output": …, "steps": 3, "tokens": 812}.
# Share one TraceEmitter and the whole hierarchy lands in a single trace tree.
```

### Keep costs bounded

```python
agent = Agent(provider, token_budget=50_000)   # hard cap per run — TokenBudgetExceeded beyond
result = agent.run("…")
print(result.usage.total_tokens)               # provider-reported, never invented
```

### Mount an MCP server's tools (two lines)

The entire MCP tool ecosystem, without wrappers — the server runs as a local
subprocess (stdio) and each of its tools becomes a regular autoagent tool,
validated against the server's own JSON Schema before anything is sent:

```python
from autoagent import MCPClient

with MCPClient(["npx", "-y", "@modelcontextprotocol/server-filesystem", "."]) as mcp:
    mcp.mount(agent, prefix="fs_", include={"read_text_file", "list_directory"})
    agent.run("List the project files and summarize the README.")
# isError results surface as tool errors (the model reacts, nothing crashes);
# transport failures raise MCPError with the server's stderr attached.
# Prefer include={...}: mounting 3 precise tools beats mounting 40.
```

### Survive crashes: checkpoint / resume

```python
import json, pathlib
from autoagent import RunState

CKPT = pathlib.Path("run_state.json")
result = agent.run(mission, checkpoint=lambda s: CKPT.write_text(json.dumps(s.to_dict())))

# …process died? restart and continue where it stopped:
state = RunState.from_dict(json.loads(CKPT.read_text()))
result = agent.resume(state)

# Budget ran out mid-run? The exception carries a ready-to-resume snapshot:
# except TokenBudgetExceeded as exc: agent.token_budget *= 2; agent.resume(exc.state)
```

### Approval gates: pause on sensitive tools, resume after a human decides

```python
from autoagent import Agent, ApprovalRequired

def policy(ctx):                                   # every tool call passes here first
    if "filesystem.write" not in (ctx.spec.permissions if ctx.spec else []):
        return None                                # allow
    if ctx.call.id not in approvals:               # your store, keyed by call id
        raise ApprovalRequired(f"{ctx.call.name}({ctx.call.arguments})")

agent = Agent(provider, tool_policy=policy)
try:
    result = agent.run("Clean up the old logs.")
except ApprovalRequired as exc:
    save(exc.state.to_dict())                      # nothing executed yet — resumable JSON
    notify_operator(exc.calls)
# …operator approves → agent.resume(state) runs the tool exactly once and finishes.
# Return a str from the policy to DENY with a reason the model can react to.
# A crashing policy DENIES (fail-closed): this hook is a security boundary.
```

### See your runs in Jaeger / Langfuse (OpenTelemetry)

```python
from autoagent import TraceEmitter, OTelTraceExporter   # pip install autoagent[otel]

with OTelTraceExporter() as exporter:                     # uses the global OTel tracer
    trace = TraceEmitter(file="trace.jsonl", on_event=exporter)   # JSONL + OTel spans
    agent = Agent(provider, trace=trace)
    agent.run("…")
# agent.run → llm → tool.<name>, with durations, statuses and redacted payloads.
# A broken OTel backend can never break the agent loop.
```

## Recipes

### A file-editing agent that can't escape its box

```python
from autoagent import Agent, ProjectWorkspace

ws = ProjectWorkspace("./my_app/src", allowed_write_extensions={".py", ".json"})

@agent.tool
def read_file(path: str) -> dict:
    return ws.read_file(path)

@agent.tool(permissions=["filesystem.write"])
def write_file(path: str, content: str, reason: str = "") -> dict:
    return ws.write_file(path, content, reason)     # history kept per change

@agent.tool(permissions=["filesystem.write"])
def rollback() -> dict:
    return ws.rollback_last_change()                # tests failed? undo.

agent.run("Rename the config loader and update its imports.")
```

If the model tries `write_file("/etc/passwd", …)` or `../../secrets.env`, the workspace
raises — the model sees `{"error": "Path escapes workspace"}` and course-corrects. Every
write is journaled (`ws.list_changes()`) and individually revertible. **The boundary is
code you can unit-test, not a system-prompt plea.**

### A chat that never forgets (in a bounded context)

```python
from autoagent import Agent, SummarizingMemory

agent = Agent(
    provider,
    memory=SummarizingMemory(cheap_provider, max_messages=40, keep_recent=12),
)
agent.register_recall_tool()      # the model gets a `recall(query)` tool

# 500 messages later: old turns live in an *incrementally updated* LLM summary
# (one cheap call per compaction — never a full re-synthesis), the last 12 stay
# verbatim, and when the user asks "what was that flag I mentioned yesterday?"
# the model calls recall("flag") over the folded history by itself.
```

If the summarizer call ever fails, compaction is skipped for that turn — context grows
temporarily instead of being silently truncated. Failure modes are boring on purpose.

### Route images to a vision model, keep text on the cheap one

```python
from autoagent.providers.routing import RoutingProvider

agent = Agent(RoutingProvider(
    default=create_provider(ModelConfig(provider="deepseek", model="deepseek-chat")),
    vision=create_provider(ModelConfig(provider="gemini",  model="gemini-3.5-flash")),
))
# Text turns → DeepSeek. A turn carrying an ImageAttachment → Gemini.
# Past image parts are stripped for the text model (which would crash on them).
# The Agent never knows: it's just an LLMProvider.
```

Custom policies are one lambda away: `router=lambda req: big if is_long(req) else small`.

### The agent writes its own tools — without owning your machine

```python
from autoagent import Agent, DynamicToolBuilder

agent = Agent(manager_provider, max_dynamic_tools_per_run=3)
agent.enable_dynamic_tools(DynamicToolBuilder(coder_provider, tools_dir="./tools_dyn"))

agent.run("Read ./access.log and give me the top 5 most-hit URLs.")
# The model decides it needs a counter → calls create_python_tool(...)
# → a second LLM writes the code → AST denylist screens it (no eval/exec,
#   imports filtered by declared permissions) → it runs in a sandbox
#   (Docker when available, hardened subprocess otherwise)
# → promotion to native execution requires a HUMAN adding its hash
#   to the tool manifest. Convenience without the YOLO.
```

### Host-driven flows the LLM cannot derail

For questionnaires, guided forms, onboarding — where skipping a step is a bug, not
creativity — `Orchestrator` inverts the roles: **your state machine** owns the flow, the
LLM only interprets answers and rephrases prompts:

```python
from autoagent.orchestrator import Orchestrator, Step

answers = {}
def current_steps():          # your code decides what's next — always
    todo = [f for f in ("name", "age", "city") if f not in answers]
    return [Step(id=f, payload={"ask": f}) for f in todo[:2]]

def record(step_id, value):   # your code validates — return an error string to reject
    answers[step_id] = value
    return None

orch = Orchestrator(provider, current_steps=current_steps, record=record)
for ev in orch.turn("I'm Ana and I'm 30"):
    if ev.type == "text":     print(ev.text, end="")     # streamed, rephrased nicely
    elif ev.type == "recorded": log(ev.step_id, ev.value)  # name AND age, one utterance
```

We run this pattern in production against a live phone line: the deterministic flow
records the answers, a supervisor agent (the "verdicts as tools" pattern above) audits
every turn in parallel, and `SummarizingMemory`-style per-caller memory makes the bot
recognize people who call back.

## How it compares

Honest positioning — these tools optimize for different things, and several of them are
excellent at what they do:

| | **autoagent** | LangChain / LangGraph | CrewAI | AutoGen | OpenAI Agents SDK | smolagents |
|---|---|---|---|---|---|---|
| **Core size** | ~6k LOC total, core loop readable in an hour | very large | large | large | medium | small |
| **Core dependencies** | **0** (stdlib) + `jsonschema` | many | many | many | `openai` sdk | `huggingface_hub` etc. |
| **Providers** | OpenAI, Anthropic, DeepSeek, Gemini — raw wire, no SDKs | very many (via integrations) | via LiteLLM | via extensions | OpenAI-first | via LiteLLM |
| **Control flow** | your Python (`run` / `run_messages` / `Orchestrator` for host-driven flows) | graphs/chains DSL | role/crew abstraction | multi-agent conversation | handoffs | code-as-actions |
| **Security model** | **code-level**: bounded workspace, Docker/AST sandbox, hash-manifest tool approval | per-integration | limited | limited | guardrails (model-level) | sandboxed code exec |
| **Observability** | typed trace events + **built-in secret redaction**, zero deps | LangSmith (SaaS) | ext. | ext. | OpenAI tracing | basic |
| **Streaming** | ✅ sync iterators (`run_stream`, text deltas + tool events, SSE in all providers) | ✅ | ✅ | ✅ | ✅ | partial |
| **Async (asyncio)** | ❌ sync by design — wrap with threads (`asyncio.to_thread`) | ✅ | ✅ | ✅ | ✅ | partial |
| **Multi-agent** | ✅ minimal primitive: `agent.as_tool()` (supervisor → specialist delegation, shared trace tree); no crew/choreography DSL | ✅ graphs | ✅ core feature | ✅ core feature | ✅ handoffs | ✅ `managed_agents` (hierarchical) |
| **Memory / RAG** | buffer + **incremental summarizing memory** + lexical recall built-in; vector store = bring your own (2-method protocol) | ✅ full RAG stack | ✅ | partial | partial | partial |
| **Multi-provider routing** | ✅ per-request (text → cheap model, images → vision model) | via config | via LiteLLM | via config | ❌ OpenAI-first | via LiteLLM |
| **MCP tools** | ✅ zero-dep stdio client, server schemas validated locally | ✅ | ✅ | ✅ | ✅ | ✅ |
| **Durable runs (checkpoint/resume)** | ✅ JSON `RunState` per step, `resume()` after crash or raised budget | ✅ LangGraph checkpointers | ✅ Flows `@persist` | partial (state save/load) | ✅ `RunState` to/from JSON + HITL interruptions | ❌ |
| **OpenTelemetry export** | ✅ optional extra, spans mirror the trace tree | via ext. | via ext. | ✅ | via ext. (tracing processors) | ✅ via openinference |
| **Best when** | you want to **own and audit** the loop; embed agents in an existing app; strict tool bounding | you want the ecosystem | role-played crews fast | conversational multi-agent research | you're all-in on OpenAI | HF ecosystem, code agents |

### When you should NOT use autoagent

- You need a **native asyncio pipeline** (the loop and streaming are sync iterators — wrapping
  them in threads is easy, but if your whole stack is `async/await`, friction adds up).
- You want **turn-key vector RAG** or hundreds of prebuilt integrations — LangChain's ecosystem is unmatched.
- You need rich **multi-agent choreography** (roles, negotiation, group chats) as a first-class
  framework feature — here you get one honest primitive (`as_tool`) and compose the rest.
- You don't want to write any Python around your agent.

If, instead, your agent is a **component inside a real application** — where you need to
know exactly what it can touch, log exactly what it did, and debug it by reading code —
that's the niche this library is built for.

## Design notes

- **One loop to understand**: send history + tool specs → LLM answers text (done) or tool
  calls → execute locally → append results → repeat, hard-capped by `max_steps`.
- **Errors are data**: a tool that raises returns `{"error": "…"}` to the model, which
  gets a chance to recover. A broken `Memory` or hook is logged and bypassed — never fatal.
- **Every preview is redacted**: trace payloads and recall snippets pass through the same
  secret-scrubbing filter (`Bearer …`, `api_key`, `?key=` patterns).
- **Providers are boring on purpose**: request in, response out. `reasoning_content`
  round-tripping (DeepSeek thinking / o-series) and `max_completion_tokens` quirks are
  handled inside the adapter so your code stays clean.

## Project status

Used in production internally (survey/phone-bot supervision at Alyce). API surface is
small and stable; version-tagged features are documented in [the developer guide](autoagent-dev-doc.md).
Contributions welcome — especially provider adapters, `Memory` backends, and sandbox hardening.

## License

MIT (see `LICENSE`).
