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

- Task: t-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: {task_id}, {date}, {dir}, {platform}, {os_version}, {turns_remaining}, {input_tokens_remaining}, {output_tokens_remaining}, and {time_remaining}.

Interactive

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

def show(werk, task, result):
    print(f"{task.get_id()}: {result}")


agent = Agent.from_env().interactive()
id = agent.add_task("Where does the configuration get loaded?")

werk = agent.start()
werk.on_result(show)
await werk.finish_all_tasks()

werk.add_reply(id, "And which environment variables override it?")
await werk.finish_all_tasks()

werk.set_task_finished(id, "answered")

An interactive agent never finishes its own task, because that would end the conversation. Every answer pauses the task instead: it stays in_progress with its agent, and each await werk.finish_all_tasks() returns on the answer it waited for. add_reply(id, content) supplies the next message, and set_task_finished(id, result) ends the conversation with the result reported to the hook. The answers in between arrive as events.

See more: Agent.

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

Set a model's context window or reasoning level when the defaults do not fit. Claude, GPT, Mistral, and Qwen families have built-in settings.

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

Terms

Fields

Rules

Write the field, followed by what it must match:

• Exact value: label = scan
• Contains text: result ~ timeout
• Time range: failed > -1h
• Has no value: agent IS EMPTY

Missing values do not match !=. For example, label != scan leaves out tasks with no label. To include them, use label IS EMPTY OR label != scan.

Quote values containing spaces: label = "needs review". Put lists in parentheses: label IN (scan, report).

Use parentheses when mixing AND and OR to make the order clear. NOT applies to the condition or parenthesized group after it. Query words such as AND ignore case; labels and IDs do not.

Relative times such as -30m, -2h, -7d, and -1w are measured when the query runs.

ORDER BY field sorts from lowest to highest. Add DESC for the reverse. Tasks with no value for that field come last. Without ORDER BY, tasks remain in creation order.

Examples

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

Cancellation applies only to the current execution: it does not change status or remain attached to the task. start() clears cancellation so unfinished tasks can resume. Use task.is_cancelled() for a task you hold, or cancelled = true and pending = true to select by execution state.

Task members:

See Task.

1. Agent handover API

Set the follow-up task on the first agent. The model can then finish with only its result:

analyst = (
    Agent.from_env()
    .label("analysis")
    .role("You are a product analyst.")
    .handover(Task("Write the board report from {parent_result}.", label="report"))
)

writer = (
    Agent.from_env()
    .label("report")
    .role("You write concise board reports.")
)

werk = Werk()
werk.add_agent(analyst).add_agent(writer)
werk.add_task(Task("Rank all products by value.", label="analysis"))

Finishing the analysis creates the report task and links it to its parent. The task may use {parent_id}, {parent_result}, and {parent_result_path}. Calling handover again replaces it. Bound object results keep this handover host-owned; the legacy envelope for scalar or unbound results may override its task or schema, but not its label.

2. Result hook

Use hooks to create new tasks when certain results arrive:

def hand_to_report(werk, done, result):
    if done.get_label() == "research":
        werk.add_task(Task(result, label="report"))


werk.on_result(hand_to_report)

3. Knowledge

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

store = Knowledge.load(".agentwerk/knowledge")

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

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

4. Task tool

Give the writer TaskTool(), and it reads what any finished task produced, by ID:

writer = Agent.from_env().label("report").tool(TaskTool())

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

5. Read result file

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

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

writer.add_task("Read .agentwerk/tasks/t-1/result.json, then write the board report.")

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

For small models, use shallow, focused schemas with few required fields, clear names, and short lists of allowed values. Split large results into labeled tasks with separate schemas, then combine them in a later task. Deep nesting, long property lists, and large anyOf or oneOf branches use more context and trigger retries.

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

analyst = Agent.from_env().handover(
    Task("Write the report from {parent_result}", label="report", schema=report_schema)
)

See Schema.

set_policy(policy) replaces the whole configuration, and get_policy() reads it back. A violated limit emits Event.POLICY_VIOLATED. 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 continues its task.

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

def watch(werk, event):
    if event.get_name() == Event.COMPACTION_FINISHED:
        print(f"[{event.get_task_id()}] compacted {event.get_data()['trigger']}")


werk.on_event(watch)

Each compaction event carries the trigger: proactive before a context-window error or reactive after one. A failure also carries a stable kind and human-readable message.

.agentwerk/
├── events.jsonl                          every event (one per line)
├── tasks/
│   └── t-1/
│       ├── task.json                   the task without its messages (id, 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 registered automatically on every agent. Agents use them to finish queued tasks or work with shared knowledge. An interactive agent gets no FinishTool() by default, since finishing its task would end the conversation. FinishTool() is the compatibility wrapper around EventTool()'s task_finished event; EventTool() remains opt-in.

When a task carries an object schema, its displayed fields are the finish call: pass them directly. Its configured handover runs automatically. Scalar and unbound tasks retain the explicit result envelope, which also supports an inline handover:

{
  "result": "...",
  "handover": {
    "label": "...",
    "task": "..."
  }
}

EventTool

Give an agent EventTool() to let it publish application events:

from agentwerk import EventTool

agent = Agent().tool(EventTool())

The model supplies a name and optional JSON data:

{
  "name": "...",
  "data": {}
}

Agentwerk records the current task and agent on every event. Werk handlers can react as events arrive, and queries can retrieve them later:

werk.on_event(lambda _, event: print(event.get_name()))
events = werk.find_events("event = event_name")

Event names are open-ended; lowercase snake case is conventional. Publishing an event does not change the task's status. The exception is the built-in task_finished event, which has the same result and handover behavior as FinishTool():

{
  "name": "task_finished",
  "data": { "result": "..." }
}

After publishing a non-terminal event, EventTool normally returns Event <name> published to the model. A directive with the event's name replaces this message. Its template can reference {data} for the complete JSON payload or any top-level field by name. The published event and tool result both record the directive key.

CommandTool

The CommandTool lets you specify exactly which commands and flags are allowed or 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")

FetchTool

The FetchTool fetches a URL and returns its text with the user agent agentwerk/<version>. impersonate() uses the headers and HTTP/2 settings of a browser.

web = FetchTool().impersonate()

Custom Tools

Use concurrent=True when a custom tool has no side effects and may run in parallel with other calls.

Agentwerk uses type annotations to tell the model which arguments it can pass. Arguments without default values are required. It understands lists, dictionaries, tuples, Literal, Optional, and unions. Use schema= when annotations are not enough.

from agentwerk import tool


@tool(concurrent=True, timeout=5)
def greet(name: str) -> str:
    """Say hello."""
    return f"Hello, {name}!"

See Tool.

Publish events

Publish custom events through the Werk. Add agent or task context when relevant:

from agentwerk import Event

werk.emit_event(
    Event("document_indexed")
    .data({"documents": 42})
    .task_id("t-1")
    .agent_id("indexer-1")
)

werk.emit_event(Event("index_refreshed"))

The first event's name, data, and context are stored as:

{"name":"document_indexed","data":{"documents":42},"task_id":"t-1","agent_id":"indexer-1"}

emit_event adds the timestamp and a known task's label. Lowercase snake case is conventional, but other names are accepted; quote names containing spaces or punctuation in AQL. Built-in events have named constructors, such as Event.task_finished() and Event.request_started("model"); their names are also available as constants. Publishing one runs its hooks and updates its statistics. It is saved according to the event's normal rules, but it does not change task or execution state.

Werk.emit_event() never changes a task's status. To let a model finish its current task through an event, register EventTool() on its agent. When the model emits task_finished, the tool validates and stores data.result, optionally creates a handover task, and marks the current task finished. All other names only publish an event.

Events are saved to .agentwerk/events.jsonl. text_chunk_received events are not saved. Read events through the Werk with these methods:

An event handler usually checks get_name() and then reads its payload with get_data(). Use get_task_id(), get_agent_id(), and get_label() to trace where it came from, and get_created_at() to read its timestamp. Application events can attach a model-facing instruction with directive(value) and read it back with get_directive().

You can query events with AQL or a condition you supply.

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

Event queries follow the same rules as task queries. Match event, agent, task, and label exactly; use ~ to search payload; and use < or > with created. Some events have no agent, task, or label. Find them with IS EMPTY. Without ORDER BY, events remain oldest first.

You can also search with one word. Agentwerk reads it as:

1. A task ID such as t-3 means task = t-3.
2. A built-in event name such as tool_call_failed means event = tool_call_failed.
3. Any other value such as scan means label = scan.

For your own event names, write event = document_indexed. If a label has the same name as a built-in event, write label = knowledge_read.

An invalid query string raises ValueError. Query(query) checks a query without running it and raises the same error.

Save replies of every finished task as a training example:

def capture(werk, event, task):
    if event.get_name() == Event.TASK_FINISHED:
        model = werk.get_model_for_agent(event.get_agent_id())
        Trajectory.from_task(event.get_agent_id(), model, task).save("datasets")


werk.on_task(capture)

Async handlers

on_result pauses the agent until the hook finishes. Use on_result_async for slower work such as storing results in a database, posting them to an HTTP API, or uploading them to object storage. It takes an async def that runs on the same event loop waiting for results.

async def store(werk, task, result):
    await database.insert(task.get_id(), result)


werk.on_result_async(store)

See Werk.

Each page is written to ./notes/pages/<slug>.md, and every page gets one line in ./notes/index.md. That list is injected into the prompt of every agent sharing the store, so each of them knows which pages it can read.

The prompt includes up to 12 000 characters of the knowledge index by default. If the index exceeds the configured limit, the agent reads the remainder from index.md. Pages are always saved in full.

Create entries in code:

from agentwerk import Page

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

page = store.get_pages().get_page("build-command")
store.get_pages().remove("build-command")

See Knowledge.