Autoagent

autoagent

[](https://github.com/laazizi/autoagent/actions/workflows/tests.yml) [](https://pypi.org/project/autoagent-core/) [](https://laazizi.github.io/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.

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.

Prefer assembling visually? Try the visual builder online — drag blocks (agent, tools, memory, MCP, checkpoint…), get runnable Python. Fully offline, also ships in the repo as constructeur_autoagent.html.

[](https://laazizi.github.io/autoagent/)

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

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.

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

1. Zero dependencies for the core. Python ≥ 3.10, urllib, dataclasses. No SDKs —

each provider adapter speaks the wire format directly (~100 lines each).

1. Multi-provider without ceremony. OpenAI, Anthropic, DeepSeek, Gemini — plus any

OpenAI-compatible endpoint (Kimi/Moonshot, Groq, Ollama, vLLM…) via ModelConfig(base_url=…). A provider is one method: complete(LLMRequest) -> LLMResponse. Write your own in 50 lines.

1. 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 StreamEvents (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), SummarizingMemory (old turns folded into an incremental LLM summary), and FactMemory (atomic facts kept up to date via add/update/delete consolidation — a contradiction replaces the stale fact; human-readable JSON store per identity; sleep-time consolidation with background=True — compact() returns in ) 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 maxsteps / tokenbudget (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

pip install autoagent-core            # installs as `import autoagent`
# or from source:
git clone https://github.com/laazizi/autoagent.git
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 (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 (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

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:

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:

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

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:

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

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

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)

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

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)

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.

Measured, not promised: the repo ships a behavioral memory eval ([evals/](evals/)) — 12 multi-session scenarios (facts established in past calls, then contradicted or made stale, then queried in a fresh session). FactMemory + recall: 12/12 · no memory: 0/12 · rolling summary: 0/12 (a rolling summary is conversation memory by design — it doesn't survive a new session; FactMemory is identity memory). Run it yourself: python evals/eval_memoire.py.

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

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.

…

Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: laazizi
• Source: laazizi/autoagent
• License: MIT
• Homepage: https://pypi.org/project/autoagent-core/

Install and usage instructions live in the source repository linked above.