You can use the {context} variable to inject contextual information:

- 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

Every value is a variable of its own: {ticket}, {date}, {dir}, {platform}, {os_version}, {turns_remaining}, {input_tokens_remaining}, {output_tokens_remaining}, and {time_remaining}.

Interactive

An interactive agent holds one ticket open across many turns, so a conversation spans a whole session.

def show(work, ticket, result):
    print(f"{ticket.key}: {result}")


agent = Agent.from_env().interactive().build()
key = agent.ticket("Where does the configuration get loaded?")

chat = agent.start()
chat.on_result(show)
await chat.finish_all()

chat.reply(key, "And which environment variables override it?")
await chat.finish_all()

chat.set_finished(key, "answered")

An interactive agent never finishes its own ticket, because that would end the conversation. Every answer pauses the ticket instead: it stays InProgress with its agent, and each await chat.finish_all() returns on the answer it waited for. reply(key, content) drives the next turn, and set_finished(key, result) ends the conversation, which is the result the hook reports. The answers in between arrive as events.

See more: AgentBuilder.

You can also read the model or provider individually: .provider(Provider.from_env()) or .model(Model.from_env()).

You can configure models to set a custom context window size or the applied reasoning. Claude, GPT, Mistral, and Qwen families are pre-configured.

Configure a custom model:

from agentwerk import Agent, Model

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

See Provider and Model.

See TicketQueue.

Terms

Fields

Rules

Operators

• key, label, status, agent, and parent take =, !=, IN, and NOT IN, which compare exactly.
• task, result, and errors take ~ and !~, which ignore case.
• created, started, finished, and failed take >, >=, <, and <=.
• IS EMPTY and IS NOT EMPTY read every field but key, status, task, and created, so finished IS EMPTY selects the tickets still open.

Times

• A compared moment is a YYYY-MM-DD date at midnight UTC, an offset back from now spelled -30m, -2h, -7d, or -1w, or milliseconds since the epoch.
• An offset is resolved when the query compiles.

Order

• ORDER BY names one field and closes the query. Every field sorts, key by its number and status along the lifecycle.
• Without ORDER BY tickets arrive in creation order.

Examples

tickets.find_results("report AND result ~ risk")           # reports that mention risk
tickets.find_tickets("errors ~ tool_call_failed")          # saw a tool call fail
tickets.find_tickets("status = todo AND agent IS EMPTY")   # waiting, never claimed
tickets.find_tickets("failed > -1h ORDER BY failed DESC")  # the last hour's failures
tickets.find_tickets(lambda t: len(t.replies) > 4)         # a callable, for what no field carries

Ticket members:

See Ticket.

1. Create tickets

Name the receiving label in the role or in the task, and the agent hands over as it finishes:

analyst = (
    Agent.from_env()
    .label("analysis")
    .role("Rank the products by value, then hand the ranking over to `report`.")
    .build()
)

writer = (
    Agent.from_env()
    .label("report")
    .role("Write the board report from the ranking you were handed.")
    .build()
)

The child ticket is filed under report and names the analysis ticket as its parent. Its body is the result that was handed over, unless the agent passes a task of its own, which may carry {parent_key}, {parent_result}, and {parent_result_path}. Either way the body ends with the parent's key and the path of its result file.

2. Read tickets

Give the writer TicketsTool(), and it reads what any finished ticket produced, by key:

writer = Agent.from_env().label("report").tool(TicketsTool()).build()

writer.ticket("Read the result of TICKET-1, then write the board report.")

3. Read result file

Give the writer ReadFileTool() instead, and it opens the result file named at the end of its ticket:

writer = Agent.from_env().label("report").tool(ReadFileTool()).build()

writer.ticket("Read .agentwerk/tickets/TICKET-1/result.json, then write the board report.")

Results live in the session directory, one result.json per ticket.

4. Share knowledge

Hand both agents one store, and either can write a page the other reads:

store = Knowledge.load(".agentwerk")

analyst = Agent.from_env().label("analysis").knowledge(store).build()
writer = Agent.from_env().label("report").knowledge(store).build()

analyst.ticket("Rank the products by value, then save the ranking to your knowledge.")

5. Register hooks

Use hooks to create new tickets when certain results arrived:

def hand_to_report(work, done, result):
    if done.has_label("research"):
        work.ticket(Ticket(result, label="report"))


tickets.on_result(hand_to_report)

A SchemaStore enforces schemas for all tickets with a certain label. Registering schemas centrally spares agents from passing complex schema structures during ticket creation (see TicketsTool) and handovers (see FinishTool):

from agentwerk import SchemaStore

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

tickets.schemas(schemas)

See Schema and SchemaStore.

policy(policy) replaces the whole configuration, and get_policy() reads it back. A violated limit emits a policy_violated event, see EventKind. compaction_threshold is the exception, see Compaction.

compaction_threshold is a fraction of the model's context window, 0.85 by default. Reaching it summarizes the older messages and the agent carries on.

Compaction also runs after the LLM provider reports the window exceeded. compaction_started, compaction_progress, compaction_finished, and compaction_failed report each step, see Events.

def watch(work, event):
    if event.kind == "compaction_finished":
        print(f"[{event.ticket_key}] compacted {event.data['reason']}")


tickets.on_event(watch)

Each of the compaction events carries the reason it ran: proactive ahead of the failure, reactive after it. Replies that still exceed the window after a reactive compaction fail the ticket.

The function returns a directive template. So you can access template variables, like {detail}, {attempt}, and {path}.

See prompts/directives for the built-in text.

.agentwerk/
├── events.jsonl                          every event (one per line)
├── tickets/
│   └── TICKET-1/
│       ├── ticket.json                   the ticket without its messages (key, status, label, timestamps)
│       ├── result.json                   the result the agent produced
│       ├── 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

FinishTool and KnowledgeTool

FinishTool() and KnowledgeTool(store) are special tools, registered automatically on every agent. They are used for interacting with the TicketQueue or knowledge base. An interactive agent gets no FinishTool() by default, since finishing its ticket would end the conversation.

CommandTool

The CommandTool allows you to granularly define what commands are allowed and what commands are denied.

git = (
    CommandTool("git")
    .allow("git status")
    .allow("git log *")
    .deny("git push*")
    .deny_flag("--force")
)

With an allow_flag set, a command carrying any other flag is refused:

cargo = CommandTool("cargo").allow("cargo test*").allow_flag("--all-features")

FetchUrlTool

The FetchUrlTool fetches a URL and returns its text, requesting it with the user agent agentwerk/<version>. impersonate() swaps in the headers and HTTP/2 settings a browser sends.

web = FetchUrlTool().impersonate()

Custom Tools

You can define custom tools for specific needs with the following parameters:

Describe the tool, then hand it the code it runs:

from agentwerk import tool


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

See Tool.

Every event is written to the session log. You read events from the ticket queue, or from the session directory in .agentwerk/events.jsonl:

You can query events with AQL syntax or callables.

tickets.find_events("tool_call_failed")
tickets.find_events("event = request_finished AND agent = research-1")
tickets.find_events("ticket = TICKET-3 ORDER BY created DESC")
tickets.find_events("payload ~ timeout AND created > -1h")

• An event query takes the same operators, AND / OR / NOT, and ORDER BY a ticket query does.
• IS EMPTY and IS NOT EMPTY read agent, ticket, and label.
• A lone word is the short form of event = <word> when it names an event, and of label = <word> when it does not. A lone TICKET-N is the short form of ticket = TICKET-N.
• A string that does not compile raises ValueError, as does Query(query), which compiles one without running it.

See EventKind and TicketQueue.

Save replies of every finished ticket as a training example:

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


tickets.on_ticket(capture)

Async handlers

on_result is blocking and prevents an agent continuing its work till the hook is finished. If you perform time-consuming operations use on_result_async instead: storing results in a database, posting them to an HTTP API, or uploading them to object storage. It takes an async def and runs it on the event loop you await finish on.

async def store(work, ticket, result):
    await database.insert(ticket.key, result)


tickets.on_result_async(store)

See TicketQueue.

The prompt carries the index up to index_char_limit, 12 000 characters by default. Past it the prompt lists the pages that fit and names index.md for the agent to read the rest. No page is refused for the length of the index, and page bodies are never shortened.

Programmatically create entries:

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")

See Knowledge.