all-in-agents 0.2.15

A minimal, universal agent framework. Zero mandatory dependencies.

English | 中文

all-in-agents

A minimal, universal agent framework for Python. Zero mandatory dependencies.

pip install all-in-agents
pip install "all-in-agents[openai]"      # OpenAI GPT
pip install "all-in-agents[anthropic]"   # Anthropic Claude
pip install "all-in-agents[mcp]"         # MCP tool providers
pip install "all-in-agents[all]"         # all optional deps

Why all-in-agents

• 🪶 Zero dependencies — pure stdlib core; adapters are opt-in extras
• 🔌 Pluggable everything — swap LLM adapter, tools, history, or orchestration without touching other parts
• 🔍 Transparent by default — append-only NDJSON event log; every run is replayable
• 🛡️ Safe by default — dangerous tools require explicit approval; budget stops runaway agents

Quick Start

pip install "all-in-agents[openai]"      # or [anthropic]

from all_in_agents import Agent

agent = Agent.quick(model="gpt-4o", workspace=".")
result = agent.run_sync("Summarize README.md in three bullet points")
print(result.final_answer)

Or with full control:

from all_in_agents import Agent, OpenAIAdapter, ToolRegistry, BUILTIN_TOOLS, unsafe_defaults

llm = OpenAIAdapter(model="gpt-4o")     # reads OPENAI_API_KEY from env
tools = ToolRegistry(approval_callback=unsafe_defaults())
for t in BUILTIN_TOOLS:                  # read_file, write_file, bash, list_files, text_search
    tools.register(t)

agent = Agent(llm=llm, tools=tools, workspace_root=".")
result = agent.run_sync("Summarize README.md in three bullet points")
print(result.final_answer)

Jupyter Notebook or async framework? Use await agent.run(goal) directly.

Streaming

Use agent.stream(...) for typed agent events, or agent.stream_text(...) when you only need text deltas.

async for event in agent.stream("Summarize README.md"):
    if event.type == "text_delta":
        print(event.data["delta"], end="")
    elif event.type == "tool_called":
        print("\ncalling", event.data["name"])

async for text in agent.stream_text("Summarize README.md"):
    print(text, end="")

Stream events include run_started, llm_started, text_delta, tool_call_delta, control_decision, assistant_message, tool_called, tool_result, run_stopped, and error. OpenAIAdapter streams token deltas for both chat_completions and responses; adapters without native streaming fall back to one full-response chunk.

Initial Messages

Use initial_messages to seed a single run with structured context before the current goal. This is a lightweight per-run context injection point, not a persistent session store.

result = await agent.run(
    "Continue the analysis",
    initial_messages=[
        {"role": "user", "content": "Earlier requirement..."},
        {"role": "assistant", "content": "Earlier result..."},
    ],
)

initial_messages is also supported by run_sync(...), stream(...), and stream_text(...). Each message must contain role and content; content may be a string or a provider-neutral content block list.

Multimodal Input

Text, image, and file input blocks are provider-neutral. The OpenAI Chat, OpenAI Responses, and Anthropic adapters convert them into the provider-specific request shape when the provider supports the block type.

from all_in_agents import FileBase64Block, FileUrlBlock, ImageBase64Block, ImageUrlBlock, TextBlock

result = await agent.run(
    "Describe the image",
    initial_messages=[{
        "role": "user",
        "content": [
            TextBlock("Use this reference image."),
            ImageUrlBlock("https://example.com/image.png", detail="low"),
        ],
    }],
)

result = await agent.run(
    "Inspect this screenshot",
    initial_messages=[{
        "role": "user",
        "content": [
            TextBlock("What error is visible?"),
            ImageBase64Block(encoded_png, media_type="image/png"),
        ],
    }],
)

result = await agent.run(
    "Summarize the attached PDF",
    initial_messages=[{
        "role": "user",
        "content": [
            TextBlock("Use the PDF as source material."),
            FileUrlBlock("https://example.com/report.pdf", filename="report.pdf"),
        ],
    }],
)

Image and file blocks are currently input-only. Model-generated image/audio/file outputs are intentionally left to adapter-specific extensions for now. OpenAI Chat Completions supports FileBase64Block and FileIdBlock; use OpenAI Responses or Anthropic for FileUrlBlock.

Turn Gates

Use on_turn when callers need to inspect or control a completed model turn before tools run. The callback can be sync or async, so it can pause for human approval, policy checks, or an external evaluator.

from dataclasses import replace
from all_in_agents import Agent, AgentTurnDecision

async def gate(turn):
    if turn.metrics["tool_calls"] == 0 and not turn.response.tool_calls:
        return AgentTurnDecision.retry(
            "Use tools before giving the final answer.",
            max_retries=2,
        )

    if any(tc.name == "bash" for tc in turn.response.tool_calls):
        return AgentTurnDecision.stop(
            final_answer="Waiting for approval",
            stop_reason="human_gate",
        )

    # Replace the effective assistant response before tool dispatch.
    if len(turn.response.tool_calls) > 3:
        return AgentTurnDecision.replace(
            replace(turn.response, tool_calls=[], stop_reason="end_turn")
        )

agent = Agent(llm=llm, tools=tools, on_turn=gate)
result = await agent.run("Inspect the workspace")

Returning None continues normally. Returning AgentTurnDecision.stop(...) stops before tool execution. Returning AgentTurnDecision.replace(...) swaps the response that will be written to history and used for the next agent action. Returning AgentTurnDecision.retry(...) rejects the current response, writes an ASSISTANT_REJECTED audit event, injects a feedback message into history, and re-calls the LLM. AgentTurn.metrics exposes cumulative run metrics such as llm_calls, tool_calls, and token usage. The default retry cap is turn_max_retries=3; each retry decision can override it with max_retries=....

CLI

# Single-shot
python -m all_in_agents "Summarize README.md" --model gpt-4o --unsafe

# Interactive REPL
python -m all_in_agents --model gpt-4o --unsafe

Core Concepts

Node / Flow

Everything is a node. A flow is a graph of nodes.

from all_in_agents import BaseNode, Flow, NodeContext

class MyNode(BaseNode):
    async def prep(self, ctx: NodeContext):
        return ctx.state["input"]

    async def exec(self, prep_result, ctx: NodeContext):
        return prep_result.upper()

    async def post(self, ctx: NodeContext, exec_result) -> str:
        ctx.state["output"] = exec_result
        return "default"   # action name → next node

node_a = MyNode()
node_b = MyNode()
node_a >> node_b           # default edge
# or: (node_a - "custom_action") >> node_b

flow = Flow()
await flow.run(ctx, start=node_a)  # ctx is a RunContext

State contract: node execution receives a typed NodeContext. Cross-node runtime state belongs in ctx.state or typed fields on ctx.run_context; node instance fields may hold persistent node-local state. If you want stateless copied node instances, use Flow(copy_nodes=True).

Flows also support lifecycle hooks, conditional nodes, subflows, and flow-level error policies:

from all_in_agents import ConditionalNode, ErrorPolicy, Flow, FlowHooks, SubFlowNode

hooks = FlowHooks(
    on_node_start=lambda ctx: print("start", ctx.node_name),
    on_node_end=lambda ctx: print("end", ctx.node_name, ctx.action),
)

optional_node = ConditionalNode(node_a, lambda ctx: ctx.state.get("enabled", False))
subflow = SubFlowNode(start=node_b)
optional_node >> subflow

flow = Flow(
    hooks=hooks,
    error_policy=ErrorPolicy.retry(
        max_attempts=3,
        retry_exceptions=(TimeoutError,),
        base_delay_ms=250,
    ),
)
await flow.run(ctx, start=optional_node)

Use RetryPolicy on a Node when retry behavior belongs to that node rather than the whole flow. It supports exception filtering, exponential backoff, jitter, and retry_after_ms hints from provider errors.

Budget & Loop Detection

from all_in_agents import Budget

budget = Budget(
    max_llm_calls=40,
    max_tool_calls=80,
    max_wall_ms=1_800_000,       # 30 min wall-clock limit
    max_input_tokens_per_call=0,  # 0 = no artificial input cap; use model context window
    loop_same_action_limit=3,    # raise LoopDetectedError after 3 consecutive identical tool calls
)

agent = Agent(llm=llm, tools=tools, budget=budget)

Before each LLM call, the agent computes a prompt budget from the model context window, output reserve, system prompt, and active tool schemas. Only the remaining budget is given to history trimming. If max_input_tokens_per_call is set, it acts as a full prompt hard cap rather than a history-only cap.

Artifact Contracts

Use artifact contracts when a run must produce machine-checkable outputs. The agent can still work freely, but the framework marks the run incomplete if required artifacts are missing or invalid.

from all_in_agents import Agent, ArtifactContract

contract = ArtifactContract.files("research_plan.md", "observation.md")

agent = Agent.quick(
    model="gpt-4o",
    workspace=".",
    artifact_contract=contract,
)
result = agent.run_sync("Create the required research artifacts")

assert result.status == "success"

JSON artifacts can be schema-checked when the jsonschema extra is installed:

contract = ArtifactContract.json_files({
    "metrics.json": {
        "type": "object",
        "required": ["score"],
        "properties": {"score": {"type": "number"}},
    }
})

RunResult.status is the broad state (success, incomplete, error, budget_exhausted, interrupted). RunResult.stop_reason is the machine-readable recovery hint, such as goal_met, artifact_missing, validation_failed, model_unavailable, budget_exhausted, or loop_detected.

Checkpoint & Resume

Agent runs can checkpoint after each Flow node. A later process can resume from the run id or checkpoint path, using the same Flow graph and restored run/history/state context.

result = await agent.run("Long task", checkpoint=True)

resumed = await agent.run(
    "Long task",
    resume_from=result.run_id,  # or result.checkpoint_path
)

For custom flows, pass a JsonCheckpointStore to Flow.run. Use explicit checkpoint_id values on nodes when a graph contains multiple nodes of the same class and must be resumed across process restarts.

Tool Registry

from all_in_agents import Tool, ToolRegistry, SideEffectLevel, ToolResponse

async def my_tool(args: dict, run) -> ToolResponse:
    result = do_something(args["input"])
    return ToolResponse(status="success", content=result)

registry = ToolRegistry(
    approval_callback=my_approval_fn   # async (name, args) -> bool
)
registry.register(Tool(
    name="my_tool",
    description="Does something useful",
    input_schema={
        "type": "object",
        "properties": {"input": {"type": "string"}},
        "required": ["input"],
    },
    side_effect_level=SideEffectLevel.READ_ONLY,
    execute=my_tool,
))

DANGEROUS and WRITES_LOCAL tools call approval_callback before executing. By default, the callback denies all requests (safe by default). Use unsafe_defaults() for development or provide your own callback. Install jsonschema for automatic argument validation with type coercion.

Tool Selection

By default every visible tool schema is sent to the model. For large registries, pass a selector so each LLM call only exposes the tools that are relevant for that turn. ToolPolicy still applies after selection.

from all_in_agents import Agent, KeywordToolSelector, StaticToolsSelector

agent = Agent(
    llm=llm,
    tools=registry,
    tool_selector=StaticToolsSelector(["read_file", "text_search"]),
)

research_agent = Agent(
    llm=llm,
    tools=registry,
    tool_selector=KeywordToolSelector(
        {
            "search": ["text_search"],
            "file": ["read_file", "write_file"],
        },
        always_include=["read_file"],
    ),
)

MCP Tools

MCP support is optional. Install all-in-agents[mcp], connect to a stdio, SSE, or Streamable HTTP MCP server, and register its tools into the normal ToolRegistry.

from all_in_agents import (
    MCPToolProvider,
    SSEMCPServer,
    StdioMCPServer,
    StreamableHTTPMCPServer,
    ToolRegistry,
)

registry = ToolRegistry()
provider = MCPToolProvider(
    StdioMCPServer(
        command="uv",
        args=("run", "mcp-server-fetch"),
    ),
    name_prefix="mcp_",
)

await provider.register_tools(registry)

remote_provider = MCPToolProvider(
    StreamableHTTPMCPServer(
        url="http://localhost:8000/mcp",
        headers={"Authorization": "Bearer token"},
    ),
    name_prefix="remote_",
)

legacy_provider = MCPToolProvider(SSEMCPServer(url="http://localhost:8000/sse"))

Remote MCP tools default to SideEffectLevel.DANGEROUS, so the registry approval flow still protects execution. For a trusted server, pass a narrower side_effect_level such as SideEffectLevel.NETWORK or SideEffectLevel.READ_ONLY.

MCPToolProvider is a client-side bridge. It can connect to servers built with FastMCP or any other MCP-compliant server, but MCP stays out of the core runtime.

Skills

Project skills are prompt bundles stored as SKILL.md files:

skills/
  reviewer/
    SKILL.md
.skills/
  local-debug/
    SKILL.md

Load selected skills by name:

agent = Agent.quick(
    model="gpt-4o",
    workspace=".",
    skills=["reviewer"],
)

Or load every discovered skill:

agent = Agent.quick(model="gpt-4o", workspace=".", skills="all")

CLI usage:

python -m all_in_agents --skill reviewer "Review this code"
python -m all_in_agents --all-skills "Use the relevant project skill"
python -m all_in_agents --project-context "Follow AGENTS.md and project context"

Hidden .skills/ entries take precedence over skills/ entries with the same name. Skills are injected into the system prompt; they do not automatically register Python tools.

History & Compression

HistoryManager compresses conversation history when it exceeds a soft threshold. By default, that threshold is 70% of the model's context window; override it with history_compress_threshold_tokens on Agent or Agent.quick. The built-in compactor targets that same soft threshold, keeps recent turns verbatim, summarizes older turns into structured JSON (facts / decisions / open_threads), and falls back to deterministic snipping if summarization fails.

agent = Agent.quick(
    model="gpt-4o",
    history_compress_threshold_tokens=18_000,
    # compression_llm=cheap_llm,  # optionally use a separate summarizer model
)

Custom compaction strategies can implement compact_turns(llm, turns, *, max_context_tokens, target_tokens=None) and return CompactionResult.

RunResult.events_path always points to the NDJSON event log. If you need an in-memory trace for evaluation or orchestration, construct the agent with include_trajectory=True; the returned RunResult.trace is a typed RunTrace, and RunResult.trajectory returns a compact list view.

Event Store

Every run writes an append-only NDJSON log to ./runs/<run_id>/events.ndjson:

{"event_id": "...", "run_id": "...", "ts": "...", "type": "RUN_CREATED", "payload": {...}}
{"event_id": "...", "run_id": "...", "ts": "...", "type": "ASSISTANT_MESSAGE", "payload": {...}}
{"event_id": "...", "run_id": "...", "ts": "...", "type": "TOOL_RESULT", "payload": {...}}
{"event_id": "...", "run_id": "...", "ts": "...", "type": "RUN_STOPPED", "payload": {"reason": "goal_met"}}

Multi-Agent

from all_in_agents import MessageBus, TaskManager, MessageEnvelope, Task

bus = MessageBus(run_dir="./runs/session_1")
tm  = TaskManager(run_dir="./runs/session_1")

# coordinator creates tasks
task = await tm.create_task(goal="Analyze file X")

# worker claims and runs
available = await tm.get_available(agent_id="worker_1")
claimed   = await tm.claim_task(available[0].task_id, "worker_1")

# agents communicate
await bus.send(MessageEnvelope(
    msg_id="...", run_id="...",
    from_agent="worker_1", to_agent="coordinator",
    msg_type="TASK_DONE", payload={"result": "..."}, ts="...",
))

TaskManager uses file-based locking (fcntl on Unix, .lock file on Windows) for safe concurrent access. Tasks support dependency chains via dependencies: list[str].

LLM Adapters

Adapter	Install extra	Environment variable
OpenAIAdapter	all-in-agents[openai]	OPENAI_API_KEY
AnthropicAdapter	all-in-agents[anthropic]	ANTHROPIC_API_KEY

Both adapters classify errors (TRANSIENT, RATE_LIMITED, AUTH, INVALID_REQUEST, INTERNAL) and retry with exponential backoff. Rate-limited requests honor retry-after headers when available.

from all_in_agents import Agent, GenerationOptions, OpenAIAdapter, AnthropicAdapter

llm = OpenAIAdapter(model="gpt-4o-mini", max_retries=3)
llm = AnthropicAdapter(model="claude-sonnet-4-6", max_retries=3)

OpenAI requests support both Chat Completions and Responses API backends. Generation controls live on the adapter, keeping Agent independent from provider-specific request fields.

llm = OpenAIAdapter(
    model="gpt-5",
    api="responses",  # or "chat_completions" for OpenAI-compatible APIs
    response_format={"type": "json_object"},
    reasoning_effort="medium",
    temperature=0.2,
    model_kwargs={"metadata": {"app": "demo"}},
)

agent = Agent.quick(
    model="gpt-5",
    api="responses",
    response_format={"type": "json_object"},
    reasoning_effort="low",
)

await llm.generate(
    [{"role": "user", "content": "Return JSON."}],
    options=GenerationOptions(reasoning_effort="high"),
)

Architecture

📁 Directory Structure

all_in_agents/
├── cli.py       Lightweight CLI runner
├── core/
│   ├── context.py   RunContext · NodeContext
│   ├── checkpoint.py FlowCheckpoint · JsonCheckpointStore
│   ├── node.py      BaseNode · Node · BatchNode · ConditionalNode
│   ├── flow.py      Flow · FlowHooks
│   ├── errors.py    ErrorPolicy · ErrorDecision
│   ├── retry.py     RetryPolicy
│   ├── subflow.py   SubFlowNode
│   ├── budget.py    Budget · BudgetLedger
│   ├── trace.py     TraceEvent · RunTrace
│   └── run.py       Run · RunResult
├── adapters/
│   ├── base.py      LLMAdapter · LLMResponse · ToolCall · GenerationOptions · LLMError · ErrorClass
│   ├── anthropic.py AnthropicAdapter (error classification, prompt caching)
│   ├── openai.py    OpenAIAdapter (routing, lifecycle, retry)
│   ├── openai_chat.py
│   ├── openai_responses.py
│   └── openai_utils.py
├── tools/
│   ├── registry.py  ToolRegistry (safe-by-default, approval callbacks, jsonschema)
│   ├── policy.py    ToolPolicy · SideEffectLevel
│   ├── coerce.py    Schema-driven argument type coercion
│   ├── builtin.py   read_file · write_file · bash · list_files · text_search
│   └── mcp.py       MCPToolProvider · stdio/SSE/Streamable HTTP transports
├── history/
│   ├── manager.py   HistoryManager (dynamic threshold, LLM-based compression)
│   ├── compactor.py HistoryCompactor (micro-compact + summarize + fallback)
│   └── store.py     FileEventStore (append-only NDJSON, event callbacks)
└── agents/
    ├── base.py      Agent · AgentConfig · Agent.quick()
    ├── nodes.py     LLMCallNode · ToolDispatchNode
    ├── streaming.py AgentStreamEvent
    ├── harness.py   AGENTS.md / .context/ project context loader
    └── multi.py     MessageBus · TaskManager · MessageEnvelope · Task · TaskStatus

Package Naming

The PyPI package is all-in-agents, but the Python import name is all_in_agents:

pip install all-in-agents

from all_in_agents import Agent   # Python import name is 'all_in_agents'

The hyphen in the PyPI name can't be used in Python imports, so the module name uses underscores.

Design Goals

• Zero mandatory deps — pure stdlib core; adapters opt-in
• Small — ~120 LOC core loop, readable in one sitting
• Composable — every piece (Node, Tool, Adapter, History) is replaceable
• Safe by default — dangerous tools require approval; budget stops runaway agents

Requirements

Python 3.10+

Optional: anthropic, openai, jsonschema

License

MIT