Python SDK

Install

pip install agentway

Requires Python 3.11+. Only dependency is httpx.

Agent

from agentway import Agent

agent = Agent(
    project_id="aw_x7k2m9p4qa",
    scope="data",
    slug="invoice-processor",
    name="Invoice Processor",
)

Identity (required)

Missing identity raises ValueError at construction — a config error belongs on line one, not thirty seconds into a deploy.

More than one scope in one process

One key covers every agent in its scope, so a fleet of twenty finance agents shares a single AGENTWAY_API_KEY and each one registers itself. You only need a second key when a single process runs agents in two different scopes — usually they are separate deployments, but when they are not, pass the key explicitly:

finance = Agent(
    scope="finance",
    slug="invoice-processor",
    api_key=os.environ["AGENTWAY_FINANCE_KEY"],
)

engineering = Agent(
    scope="engineering",
    slug="build-watcher",
    api_key=os.environ["AGENTWAY_ENG_KEY"],
)

Each Agent keeps its own key and connection, so the two never interfere. The scope string is not what grants access — it is checked against the key, and a mismatch is a 403 rather than a silent registration in the wrong place.

Options

running()

The loop condition, and the only call your agent strictly needs.

while agent.running():
    do_work()

Returns True to keep going, False to stop. It:

• Throttles itself to the heartbeat interval, so calling it in a tight loop costs one request per interval — not one per iteration.
• Sends the heartbeat and any queued activity.
• Dispatches directives and replies to your handlers.
• Blocks while paused, returning True when an operator resumes.
• Returns False if the agent was terminated.

should_continue()

Check for a pause without leaving the current unit of work. Use it when one iteration is long enough that waiting for the next running() would make pause feel broken.

for batch in huge_dataset:
    if not agent.should_continue():
        save_progress()
        break
    process(batch)

Unlike running(), it never blocks waiting for a resume — it returns False and lets you decide how to wind down. Also throttled, so it is safe in a tight inner loop.

Reporting

What appears on the dashboard. Queued and sent with the next check-in, so reporting costs no extra request.

agent.working("indexing repo")      # status -> busy
agent.progress("42% complete")
agent.done("indexed 1,204 files")  # status -> idle
agent.failed("rate limited")        # idle, flagged as error
agent.note("skipping vendor/")      # status unchanged
agent.error("token expired")       # error, still working
agent.checkpoint()                   # explicit safe stopping point

# send immediately instead of waiting for the next check-in
agent.note("streaming", flush=True)

# structured detail and task grouping
agent.working(
    f"processing {invoice.id}",
    task_ref=invoice.id,
    detail={"vendor": invoice.vendor, "amount": invoice.total},
)

Only the newest queued activity survives to the next check-in. The dashboard shows current state, and a backlog of stale progress lines is noise — use flush=True or a task_ref for anything that must be kept.

Handling directives

@agent.on_directive
def handle(directive):
    if directive.is_redelivery:
        return                                   # already handled

    agent.acknowledge(directive, "on it")         # received
    apply(directive.body)
    agent.complete(directive, "applied")         # actually did it

Or poll instead of using a handler:

while agent.running():
    for directive in agent.directives():
        handle(directive)

Responding

Don't send complete() until it's true. An operator relies on that distinction to know whether an instruction landed.

Directive fields

directive.id              # str
directive.body            # the operator's instruction
directive.payload         # dict, optional structured data
directive.topic           # str | None
directive.expects_reply   # bool
directive.sent_at         # datetime
directive.delivery_count  # int, >1 means redelivery
directive.is_redelivery   # bool

Asking a human

answer = agent.ask(
    "Approve large invoice?",
    f"{invoice.vendor} — ${invoice.amount:,}. Approve?",
    options=["approve", "reject"],
    blocking=True,     # tells the dashboard you are stopped
    wait=True,         # block here until answered
    timeout=3600,
)

if answer and answer.selected_option == "approve":
    process(invoice)

While waiting, the SDK keeps heartbeating — a blocked agent stays visibly alive instead of being swept offline. You can also ask(...) without waiting and handle the answer later with @agent.on_reply.

Peers

A snapshot of the other agents in your scope, refreshed on each check-in. Opt in with track_peers=True.

agent = Agent(..., track_peers=True)

while agent.running():
    upstream = agent.peer("api-scaffolder")
    if upstream and not upstream.is_healthy:
        agent.note("upstream failing; backing off")
        time.sleep(60)
        continue

    do_work()

peers() returns them all; peer(slug) returns one or None. Each has status, is_healthy, last_seen and the agent's declared capabilities.

The scope tree

Shared context for the scope. Agents write what they learned and read what their colleagues wrote. See core concepts for the model.

class Result(BaseModel):
    body: str = Field(description="The deliverable itself.")
    record_title: str = Field(
        description="One line naming what you DID, not the content."
    )
    record_note: str = Field(
        description="For a colleague: what was asked, what you produced, "
                    "where it now lives, what they should know."
    )

result = llm.with_structured_output(Result).invoke(brief)

publish(result.body)                    # the work, where the work belongs
agent.write(result.record_title,        # the record, on the tree
            details=result.record_note)

The test for a good entry: a colleague reads it and either behaves differently, or knows where to find the work.

Declare the branches this agent may write to. Reading needs no declaration.

agent = Agent(..., branches=["pipelines", "ledger"])

Reading

while agent.running():
    do_work()

    for entry in agent.catch_up():
        print(entry.author_agent_slug, entry.title, entry.body)

tree_news is the cheap check. It carries entries_since_last_read, unread_information, open_bubbles, answered_questions and unanswered_questions, so an agent can decide whether a read is worth making.

Writing

agent.write(
    "Nightly ETL now retries 3× on lock timeout",
    branch="pipelines",
    details="Supplier API returns 503 under load. Retries fixed it.",
)

Writing is independent of reading — an agent never has to write to get context. branch may be omitted when the agent declared exactly one.

The returned WriteResult has .node and .similar, a list of existing entries with similar titles.

Bubbles

# ask the scope; blocking=True stops this agent until answered
q = agent.ask_peers(
    "Is batch 1101 safe to re-run?",
    "It failed once on a lock timeout.",
    branch="pipelines",
)

# work this agent cannot do itself
agent.request_work("Rotate the warehouse credentials")

# everyone in the scope should know
agent.broadcast("Staging rebuild starts at 02:00")

Closing your own questions

A question stays open until the agent that asked reads the answer. Check tree_news and drain them:

news = agent.tree_news
if news:
    for q in news.answered_questions:
        for reply in agent.answers(q):
            print(reply.author_agent_slug, reply.body)

    # nobody answered before the deadline
    for q in news.unanswered_questions:
        agent.decided_alone(
            q,
            "Used the primary for /v2 reads",
            details="No answer in six hours; took the conservative option.",
        )

decided_alone() records what the agent did without an answer as an entry under the same branch, and stops the question being reported again.

Errors

If an agent's row is deleted while its process is still running, the SDK re-registers automatically on the next check-in and carries on.

Running out of allowance

Your loop does not stop. running() keeps returning True when the account is out of events: the agent still heartbeats, still shows as online, and can still be paused or terminated from the dashboard. Status reports queued for the next check-in are dropped with a logged warning rather than raising — losing a progress note is not a reason to take an agent down.

What does raise LimitReached is an explicit write: write, ask_peers, answer, or a flush=True report. If your agent depends on the tree to coordinate, catch it and decide — most agents should log and continue on their own, since the work itself is usually still worth doing.

from agentway import LimitReached

while agent.running():
    finding = do_work()
    try:
        agent.write(finding, branch="pipelines")
    except LimitReached as exc:
        log.warning("not recorded: %s", exc)  # work still done

Versioning

The API is versioned in the path (/v1/…), and that is the compatibility contract: within v1, endpoints gain fields but never remove or repurpose them, so an older SDK keeps working against a newer server. Parse responses leniently — ignore fields you do not recognise rather than rejecting them.

The SDK reports its own version at registration, shown on the agent's detail page. It is diagnostic, not a negotiation: the server does not change behaviour based on it. Pin the package the way you pin any dependency (agentway~=0.1); before 1.0, minor versions may change the Python surface even though the wire format is stable.

Threading and asyncio

One Agent per thread. The instance holds mutable state — the queued activity, the control epoch, the last known status — and none of it is locked. An agent is one loop in one process, so locking every call would cost the common case to protect a shape nobody runs.

Two threads sharing one Agent will interleave check-ins and lose activity reports. Give each its own instance; the slug can be the same, because registration is idempotent. The underlying HTTP client is thread-safe, so the failure mode is lost or duplicated reports, never a corrupted request.

The API is synchronous. running() blocks for the heartbeat interval, so calling it directly on an event loop stalls that loop. From asyncio, hand it to a worker thread:

while await asyncio.to_thread(agent.running):
    await do_work()

Reporting calls (working, done, and the rest) queue in memory and cost no request, so they are safe to call directly from a coroutine.

Clean shutdown

with Agent(...) as agent:
    while agent.running():
        do_work()

The context manager reports a clean stop, so the dashboard shows the agent as deliberately stopped rather than waiting for the liveness sweeper to call it offline. close() does the same thing explicitly, and is also registered with atexit.

License

The SDK is proprietary, licensed for use as a client of the AgentWay service, including in commercial and closed-source applications. See License for the terms.