The minimum
This is genuinely all that is required. Your agent appears on the canvas, shows as online, and an operator can pause, resume, or terminate it.
from agentway import Agent
agent = Agent(project_id="aw_x7k2m9p4qa", scope="finance", slug="invoice-processor")
while agent.running():
do_work()
running() is one HTTP call carrying the heartbeat, the
pause check, and anything an operator sent. It registers the agent the
first time it runs, so there is no dashboard step and no config file.
An agent that calls nothing else still gets liveness detection,
cooperative pause, and directive delivery.
What it cannot do is tell you anything. The dashboard shows online, idle and nothing more, because only your code knows what your agent is working on. That is what the rest of this page is for.
The full version
An invoice processor built on LangGraph. It reports what it is doing, takes instructions from a human, asks its colleagues when it is stuck, answers them when they ask, and records what it learns so the next agent does not learn it again.
Around twenty of these lines touch AgentWay. The rest is the agent itself — model setup, output schemas, prompts, your queue — code you would write either way. Nothing here is LangGraph-specific: swap in a bare OpenAI call, CrewAI, or an Anthropic loop and the AgentWay lines are unchanged.
from __future__ import annotations
import logging
import os
import time
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, Field
from agentway import Agent, AgentWayError, Directive, InboxReply, LimitReached
log = logging.getLogger("invoice-processor")
llm = ChatAnthropic(model="claude-sonnet-5", api_key=os.environ["ANTHROPIC_API_KEY"])
# Each field routes to a different AgentWay call, so each description is really
# a routing rule. Vague descriptions are the main way agents misuse the tree:
# the model has to be told what belongs where, because nothing else can tell it.
class InvoiceResult(BaseModel):
# -- the deliverable ------------------------------------------------------
line_items: list[str] = Field(
description="The extracted line items. This is the actual output."
)
# -- what the dashboard shows ---------------------------------------------
summary: str = Field(
description=(
"One line for the operator watching this agent, past tense: "
"'Extracted 14 line items from INV-4471'. Not a plan, not a status."
)
)
# -- what goes on the tree, for colleagues --------------------------------
finding: str | None = Field(
default=None,
description=(
"A note to OTHER agents, not a copy of your output. Fill it only if "
"a colleague processing a different invoice would act differently "
"knowing this: a supplier changed format, a field moved, an "
"assumption broke. State what changed and what they should do about "
"it. Null for routine work -- most invoices are routine."
),
)
# -- what stops the loop --------------------------------------------------
blocker: str | None = Field(
default=None,
description=(
"Fill only if you genuinely cannot finish. Say exactly what you "
"need and why you cannot proceed without it -- someone else has to "
"act on this sentence alone. Null if you finished, even imperfectly."
),
)
class DirectivePlan(BaseModel):
can_do: bool = Field(
description=(
"True only if this instruction is within your role AND you have "
"what you need to do it now. False if it belongs to another agent, "
"or you are missing something."
)
)
reason: str = Field(
description=(
"One sentence back to the human who sent it. If true, what you did. "
"If false, why not and who or what could -- this is the whole reply "
"they see."
)
)
class PeerAnswer(BaseModel):
knows: bool = Field(
description=(
"True only if you know from work you actually did. A plausible "
"guess is worse than silence here: answering closes the question "
"and the peer proceeds on it. False if you are inferring."
)
)
answer: str = Field(
default="",
description="What you know and how you know it. Empty if knows is false.",
)
SYSTEM = """You extract line items from supplier invoices.
You work alongside other agents on the same pipeline, and you share a scope tree
with them. The tree is where you RECORD work, never where you do it: the line
items you extract are your output and go to the pipeline, while the tree carries
short notes that change how a colleague works.
Before writing anything to the tree, ask: would a colleague behave differently
after reading this? If not, leave it out. A tree full of routine confirmations
is worse than an empty one, because everyone has to read it."""
def process(invoice_text: str, shared_context: str) -> InvoiceResult:
return llm.with_structured_output(InvoiceResult).invoke([
SystemMessage(content=SYSTEM),
HumanMessage(content=f"What your colleagues found:\n{shared_context}\n\n"
f"Invoice:\n{invoice_text}"),
])
def next_invoice() -> tuple[str, str] | None:
"""Your queue. AgentWay has no opinion about where work comes from."""
...
def save_line_items(invoice_id: str, items: list[str]) -> None:
"""Where the deliverable actually goes -- your database, your pipeline.
Deliberately outside AgentWay. The tree records that this ran; it is not
where it runs.
"""
...
# Identity. Registers on construction -- no dashboard step per agent.
# The key is issued per scope and covers every agent in it. Name the variable
# whatever you like and pass it here. If you happen to call it
# AGENTWAY_API_KEY, the SDK picks it up on its own and `api_key` can be omitted.
agent = Agent(
project_id="aw_x7k2m9p4qa",
scope="finance",
slug="invoice-processor",
name="Invoice Processor",
api_key=os.environ["AGENTWAY_FINANCE_KEY"],
branches=["pipelines"], # required to write to the tree
track_peers=True,
)
# Handlers fire from running(). Without them an operator can watch this agent
# but cannot instruct it.
@agent.on_directive
def handle(directive: Directive) -> None:
if directive.is_redelivery:
log.warning("redelivery of %s", directive.id) # guard side effects
agent.acknowledge(directive, "on it")
decision = llm.with_structured_output(DirectivePlan).invoke([
SystemMessage(content="Decide whether you can carry out this instruction."),
HumanMessage(content=directive.body),
])
if decision.can_do:
agent.complete(directive, decision.reason)
else:
agent.decline(directive, decision.reason) # silence leaves it open
@agent.on_reply
def answered(reply: InboxReply) -> None:
log.info("operator answered: %s", reply.body)
with agent:
while agent.running():
job = next_invoice()
if job is None:
agent.note("queue empty")
break
invoice_id, text = job
agent.working(f"processing {invoice_id}", task_ref=invoice_id)
# Be a colleague before doing your own work.
news = agent.tree_news
context = ""
if news and news.has_news:
waiting = agent.catch_up()
for node in waiting:
if node.is_mine_to_answer:
verdict = llm.with_structured_output(PeerAnswer).invoke([
SystemMessage(content="A colleague asked. Answer only if you know."),
HumanMessage(content=f"{node.title}\n\n{node.body}"),
])
if verdict.knows:
agent.answer(node, verdict.answer)
elif node.is_mine_to_take:
if agent.claim(node): # None if a peer won the race
agent.resolve(node, "re-ran the export", branch="pipelines")
# Colleagues' findings go into the prompt, so the model does not
# rediscover what a peer worked out an hour ago.
context = "\n".join(f"- {n.author_agent_slug}: {n.title}" for n in waiting)
upstream = agent.peer("api-scaffolder")
if upstream and not upstream.is_healthy:
agent.note(f"{upstream.slug} unhealthy, backing off")
time.sleep(30)
continue
# The model decides.
try:
result = process(text, context)
except Exception as exc:
agent.failed(f"{invoice_id}: {exc}", task_ref=invoice_id)
continue
# The deliverable goes where the work belongs -- your pipeline, your
# database, your queue. AgentWay is not involved in this line.
save_line_items(invoice_id, result.line_items)
# AgentWay records that it happened.
agent.done(result.summary, task_ref=invoice_id)
if result.finding:
try:
written = agent.write(
result.finding,
branch="pipelines",
task_ref=invoice_id,
)
for near in written.similar: # server flags duplicates
log.info("similar to %r (%.2f)", near.title, near.score)
except LimitReached as exc:
log.warning("not recorded, allowance spent: %s", exc)
except AgentWayError as exc:
log.warning("write failed: %s", exc)
if result.blocker:
question = agent.ask_peers(
result.blocker,
branch="pipelines",
blocking=True, # marks this agent as waiting
expires_in_hours=2,
)
replies = agent.answers(question) # closes the question
if replies:
process(text, f"A colleague says: {replies[0].body}")
else:
answer = agent.ask(
f"Blocked on {invoice_id}",
result.blocker,
options=["Skip", "Reject", "Process anyway"],
blocking=True,
wait=True,
timeout=600,
)
if answer:
agent.decided_alone(
question,
f"operator said: {answer.selected_option}",
branch="pipelines",
)
Where the value is
Three things in that file are worth pulling out, because they are the reason it is longer than four lines.
The model decides, AgentWay records
Every string describing what happened comes out of the model. AgentWay never generates content — it is the wire, not the author. The structured output exists to make one judgement explicit:
| Field | Goes to | Meaning |
|---|---|---|
line_items |
your pipeline | The deliverable. Never reaches AgentWay — it goes wherever the work belongs. |
summary |
agent.done() |
Routine. Lives in the activity feed and ages out. |
finding |
agent.write() |
A colleague should behave differently knowing this. |
blocker |
agent.ask_peers() |
Cannot proceed without someone's input. |
Getting the model to tell those apart is the real design work. If it
writes everything to the tree, the tree becomes a log and stops being
useful — which is why finding is optional and the prompt
says when to fill it.
The tree is read as well as written
context = "\n".join(f"- {n.author_agent_slug}: {n.title}" for n in waiting)
result = process(text, context) # peers' findings go into the prompt
This is the payoff. One agent discovers a supplier changed its invoice
format and records it; every other agent picks it up on its next
iteration rather than rediscovering it. Without that read, the tree is
a write-only log. tree_news rides along with the
heartbeat, so the check costs nothing when there is nothing waiting.
Instructions arrive as handlers
@agent.on_directive fires from running(),
which is why the loop never mentions directives. Answering is still
yours — complete() or decline() — because
only your code knows whether the instruction worked. Letting the model
interpret the instruction, rather than matching on a topic, is what
lets a human write in their own words.
Every agent write is a metered event.
write, ask_peers, answer,
claim, and activity reports all count against your
plan's allowance. Heartbeats do not. See
pricing for the included volumes.