agentwerk (Python)

Why use agentwerk?

• Minimal interface: create agents with a few lines of code.
• Complex workflows: allow agents to interact through shared knowledge and tickets.
• Deep observability: inspect every request, message and failure.
• Ease of integration: apply agents as simple as HTTP calls.
• Facilitate training: collect trajectories for fine-tuning models.

Installation

Python

pip install agentwerk

Also see: Rust implementation.

Quick Start

import asyncio
from agentwerk import Agent, GrepTool, ReadFileTool


async def main():
    agent = (
        Agent.from_env()
        .role("You are a Rust developer who explores source files to answer questions.")
        .tool(ReadFileTool())
        .tool(GrepTool())
        .build()
    )

    agent.task(
        "Find every `pub trait` defined under src/ and explain each in one sentence."
    )
    work = agent.start()
    results = await work.finish_all()

    print(results[-1])


asyncio.run(main())

Use Cases

Example projects built with agentwerk:

• Terminal REPL: minimal interactive chat
• Divide and Conquer: arithmetic problem shared across agents, ported in examples/divide_and_conquer.py
• Deep Research: deep research pipeline (requires BRAVE_API_KEY)
• Malware Scanner: identify indicators of compromise in a software package
• Apparat Fabrik: a shift on the line of an apparatus works

Configure an LLM provider first (see Environment).

python examples/divide_and_conquer.py 200 4 2

API

The API, section by section:

• Agents: Define roles, behavior and actions.
• Tickets: Coordinate complex work across agents.
• Tools: Define accessible tooling.
• Events: Requests, tool usage, failures and more.
• Stats: Metrics about tickets, tokens and time.
• Knowledge: Notes agents can share for collaboration.
• Sessions: Directory layout of data agents create.

Agents

An Agent is the core entity of agentwerk. It has access to tools for solving tasks in the form of tickets.

from agentwerk import Agent, ReadFileTool

agent = (
    Agent.from_env()
    .label("math")
    .role("You are an arithmetic agent. Compute step by step and show your work.")
    .tool(ReadFileTool())
    .build()
)

tickets.agent(agent)
tickets.task("Compute (47 * 92) / 8, then round to the nearest integer.")

- Ticket: TICKET-7
- Date: 2026-05-06
- Working directory: /Users/caro
- Platform: darwin 25.1.0
- Turns remaining: 8
- Input tokens remaining: 95000
- Output tokens remaining: 12000
- Time remaining: 240s

Providers

Connect to a Provider to give agents access to LLMs. agentwerk supports: Anthropic, OpenAI, Mistral, and a LiteLLM proxy.

from agentwerk import Agent, AnthropicProvider

agent = (
    Agent()
    .provider(AnthropicProvider(key))
    .model("claude-sonnet-4-20250514")
)

# Or read both from the environment.
agent = Agent.from_env()

Models

You can configure models to set a custom context window size or the applied reasoning:

from agentwerk import Agent, Model

agent = Agent().model(
    Model("my-local-model").context_window(128_000).reasoning_effort("high")
)

Claude, GPT, Mistral, and Qwen families are pre-configured.

Tickets

The TicketQueue is the core data structure of agentwerk allowing to coordinate complex interactions.

analyst = (
    Agent.from_env()
    .label("analysis")
    .build()
)

tickets.agent(analyst)
tickets.ticket(Ticket("Rank all products by value.", labels=["analysis"]))

Execution

tickets.start()
answer = (await tickets.finish_all())[-1]

Results

Access the results of the agents' work:

await tickets.finish_all()

answers = tickets.results()
if answers:
    print(answers[-1])

for ticket in tickets.tickets():
    print(f"{ticket.key}: {ticket.status}")

Each Ticket carries a result as free text or JSON validated by schemas:

ticket = tickets.find_ticket(lambda t: t.has_label("analysis"))
print(ticket.result["title"])

Schemas

A Schema constrains the result an agent produces for a ticket. A violation triggers a retry until max_schema_retries is exhausted.

from agentwerk import Schema, Ticket

schema = Schema(
    {
        "type": "object",
        "properties": {"title": {"type": "string"}},
        "required": ["title"],
    }
)

tickets.ticket(Ticket("Write a report.", schema=schema))

Enforce schemas for all tickets with a certain label. Registering schemas centrally spares agents from passing complex schema structures during ticket creation (see ManageTicketsTool) and handovers (see FinishTool).

from agentwerk import SchemaStore

schemas = SchemaStore()
schemas.label(
    "report",
    {
        "type": "object",
        "properties": {"title": {"type": "string"}},
        "required": ["title"],
    },
)

tickets.schemas(schemas)

Policies

Policies allow you to define execution limits:

(
    tickets.max_turns(40)
    .max_time(300.0)
    .max_input_tokens(200_000)
    .max_output_tokens(50_000)
)

Tools

Tools allow agents to perform their work.

from agentwerk import Agent, BashTool, GrepTool, ReadFileTool

agent = (
    Agent()
    .tool(ReadFileTool())
    .tool(GrepTool())
    .tool(BashTool("git", "git *"))
)

FinishTool() and ManageKnowledgeTool(store) are special tools, registered automatically on every agent. They are used for interacting with the TicketQueue.

Custom tools

You can define custom tools for specific needs:

from agentwerk import tool


@tool(
    read_only=True,
    schema={
        "type": "object",
        "properties": {"name": {"type": "string"}},
        "required": ["name"],
    },
)
def greet(name: str) -> str:
    """Say hello."""
    return f"Hello, {name}!"

Events

Events give you insights to the lifecycle and activities of your agents' work.

def log(event):
    if event.kind == "ticket_finished":
        print(f"[{event.agent_id}] done {event.ticket_key}")


tickets.on_event(log)

Hooks

Hooks allow you to react to events:

tickets.create_ticket_on_failure(
    lambda event, ticket: Ticket(ticket.task, labels=["retry"])
)

def capture(event, ticket):
    if event.kind == "ticket_finished":
        model = tickets.model_for_agent(event.agent_id)
        Trajectory.from_ticket(event.agent_id, model, ticket).save("datasets")


tickets.on_ticket(capture)

Stats

Statistics give you deep insights into behavior of your agents: working time, tickets, failure rates, bottlenecks etc.

stats = tickets.stats()
print(stats.event_count("request_finished"), stats.input_tokens())

for name, stat in stats.tool_stats().items():
    print(name, stat.calls)

Knowledge

Knowledge allows agents to share insights or learnings. Knowledge pages are created in the Open Knowledge Format (OKF).

from agentwerk import Agent, Knowledge

store = Knowledge.load("./notes")
alice = Agent().knowledge(store)
bob = Agent().knowledge(store)

from agentwerk import Page

store.pages().save(
    Page(
        "build-command",
        "How the project is built.",
        "Run `make` to compile.",
        tags=["build"],
    )
)

page = store.pages().load("build-command")
store.pages().remove("build-command")

Sessions

A TicketQueue writes every ticket, reply, statistic, and lifecycle event to its working directory (default ./.agentwerk). You can continue a session from that directory.

tickets = TicketQueue.load(".agentwerk")
tickets.agent(my_agent)
tickets.start()

.agentwerk/
├── stats.json                            execution statistics
├── tickets.jsonl                         lifecycle events (one per line)
├── results.jsonl                         finished results (one per line)
├── tickets/
│   └── TICKET-1/
│       ├── ticket.json                   the ticket without its messages (key, status, labels, timestamps, result)
│       ├── replies.jsonl                 every message exchanged with the model, one per line
│       └── outputs/<tool_use_id>.txt     full tool outputs spilled out of the messages
└── knowledge/
    ├── pages/<slug>.md                   knowledge pages
    └── index.md                          knowledge index

Development

See DEVELOPMENT.md.