Use {{ context }} in a prompt to include the current task and execution limits:

- 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

Each context field is also available separately: {{ task_id }}, {{ date }}, {{ dir }}, {{ platform }}, {{ os_version }}, {{ turns_remaining }}, {{ input_tokens_remaining }}, {{ output_tokens_remaining }}, and {{ time_remaining }}.

Interactive

An interactive agent keeps a task open across replies. It has no completion tool by default.

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

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

werk.set_task_finished(id, "answered")

Replies pause the task in in_progress, and completion methods return when it pauses. Use add_reply(id, content) to resume and set_task_finished(id, result) to end the conversation. Intermediate replies arrive as events. on_result receives the final result.

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

null, empty arrays, and unmatched selectors render to an empty string.

task is the original JSON input. Unset label, schema, assignee, and directive properties are omitted. Task results, errors, replies, and cancellation state are not serialized.

For example, given this research result:

{
  "company": {"name": "Canvas Computing"},
  "findings": [{"summary": "one"}, {"summary": "two"}]
}

This template:

{{ result: research | company.name }}

Renders as:

Canvas Computing

Select each task input or event's data:

{{ tasks: research | [*].task }}
{{ events: event.name = tool_call_failed | [*].data }}

research is shorthand for task.label = research.

Missing fields, incompatible types, and out-of-range indexes produce null. Wildcards, slices, and flattening omit null values when more path steps follow. Filters, comparisons, logical expressions, literals, multi-selects, functions, and additional pipes are not supported.

{ name } stays unchanged. To output the literal text {{ name }}, write {{{{ name }}}}.

agentwerk corrects common result-formatting mistakes, such as a quoted number or a nested object encoded as JSON text. Schema-bound results must be objects. Remaining schema violations trigger a retry, subject to max_schema_retries. Without a schema, a task may return any JSON value.

Use shallow, focused schemas for small models. Split complex work into tasks with separate schemas.

See Schema.

Built-in keys override recovery text. Keys without overrides retain their defaults. Templates accept runtime values such as {{ detail }}, {{ attempt }}, and {{ path }}. Placeholders without a value remain unchanged.

See prompts/directives for the built-in text.

See Werk.

Terms

Fields

Queries using both namespaces match events with their referenced tasks. Events without an existing task do not match. Joined matches default to event-log order. ORDER BY accepts task or event fields.

Result finders return raw results where task.result is present. They select finished tasks unless the query specifies another status.

Completion methods and cancel_tasks also accept AQL. Event and joined queries snapshot matching task IDs when the operation starts. Task-only cancellation also applies to later matching tasks.

Rules

Missing values do not match !=. Include unlabeled tasks with task.label IS EMPTY OR task.label != scan.

Qualify fields in full expressions. Use parentheses when mixing AND and OR. NOT applies to the next condition or group. Query keywords ignore case. Labels and IDs do not.

Times accept UTC dates such as 2026-08-30, epoch milliseconds, or offsets such as -30m, -2h, -7d, and -1w. Offsets are resolved when a query is compiled. Reusing a compiled query keeps its original cutoff.

Missing sort values come last in either direction. Tasks and results selected through events follow matching event order, with each task returned once. Events selected through tasks follow task order, then log order within each task.

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

Examples

# Find tasks referenced by failed tool-call events.
werk.find_tasks("event.name = tool_call_failed")

# Find events attached to tasks labeled "scan".
werk.find_events("scan")

# Find results produced by tasks with finished events.
werk.find_results("event.name = task_finished")

werk.find_results("report AND task.result ~ risk")
werk.find_tasks("task.errors ~ tool_call_failed")
werk.find_tasks("task.status = todo AND task.assignee IS EMPTY")
werk.find_tasks("task.failed > -1h ORDER BY task.failed DESC")
werk.find_tasks(lambda t: len(t.get_replies()) > 4)       # a callable, for what no field carries

Cancellation affects only the current execution, not persisted task status. Starting a new run with start() clears cancellation. Inspect it with task.is_cancelled(), or query execution state with task.cancelled = true and task.pending = true.

Task members:

See Task.

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)

Runtime condition

Use an AQL query to create follow-up tasks or add agents based on conditions:

from agentwerk import Condition

werk.add_condition(
    Condition("task.label = research AND task.status = finished")
    .id("report-after-research")
    .add_agent(Agent.from_env().label("report"))
    .add_task(
        Task(
            "Write {{ result: task.label = research AND task.status = finished }}",
            label="report",
        )
    )
)

KnowledgeTool

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

TaskTool

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

ReadFileTool

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.

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                        recorded events, one per line
├── tasks/
│   └── t-1/
│       ├── task.json                   task metadata and input
│       ├── result.json                 task result
│       ├── replies.jsonl               model messages, one per line
│       └── outputs/<tool_use_id>.txt   full tool outputs
└── knowledge/
    ├── pages/<slug>.md                 knowledge pages
    └── index.md                        knowledge index

FinishTool

Agents use FinishTool to end their task and share their outcomes:

{
  "answer": "The configuration is loaded in src/config.rs.",
  "confidence": 0.9
}

To return a result, the agent must call FinishTool. If the task has a result schema, the tool validates the object against it. For a non-interactive task without a schema, the agent can instead finish by responding with plain text.

Interactive agents are the exception: they have no FinishTool unless you add one explicitly with .tool(FinishTool()).

Timeouts

Override a tool's limit with timeout(seconds). Zero disables it.

from agentwerk import FetchTool

quick_fetch = FetchTool().timeout(15)
patient_fetch = FetchTool().timeout(0)

When a Python tool times out, the agent stops waiting, but its worker thread may continue in the background.

EventTool

Give an agent EventTool to let it publish custom events:

from agentwerk import EventTool

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

The model supplies a name and optional JSON data:

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

Events carry the current task and agent context. See Events for hooks and queries. Names are unrestricted. Lowercase snake case is conventional.

Only task_finished completes the current task. Its data is the result dictionary:

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

Use Directives to customize the acknowledgement sent to the model.

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.

Event names

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

Werk.emit_event() does not change task status. Use EventTool for model-driven completion through task_finished.

Read events

Events are saved to .agentwerk/events.jsonl, except text_chunk_received.

The default logger runs when no event hook is installed.

Query events

Query events with AQL or a predicate. See Queries for shared syntax.

werk.find_events("event.name = tool_call_failed")
werk.find_events("event.name = request_finished AND event.agent_id = research-1")
werk.find_events("event.task_id = t-3 ORDER BY event.created DESC")
werk.find_events("event.data ~ timeout AND event.created > -1h")

Use IS EMPTY to find events without agent, task, or label context. event.data searches only the payload, not the name. Events default to oldest-first order.

See Event and Werk.

Hooks

on_result runs synchronously on the agent. Keep it brief. Use on_result_async for work that needs to await.

Async hooks run while a completion method is waiting, and finish before it returns. start() alone does not run them. Do not call finish, finish_task, or finish_tasks inside an async hook: it can deadlock. Python hooks use async def and run on the caller's event loop.

Pages use the Open Knowledge Format (OKF) and are stored at ./notes/pages/<slug>.md. Each has an entry in ./notes/index.md, which is included in the prompts of agents sharing the store.

By default, prompts include up to 12,000 characters of the index. Agents can read the rest 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.