Quickstart

pip install pyrula-agents, then scaffold a runnable agent:

pyrula init --agent
pyrula dev --open

What it wrote

pyrula init --agent writes app.py: a support agent with one tool, lookup_order. It runs offline as written, against a scripted fake model, so there’s no API key and no broker to set up first. Export ANTHROPIC_API_KEY and rerun pyrula dev to swap in a real Anthropic model; that branch lives in the generated file itself, so which model you’re talking to is never hidden in the CLI.

What `pyrula dev` does

pyrula dev --open starts a local server, submits one turn against support with its default message, runs it, and opens the dev UI so you can see the turn’s event stream. The terminal output looks like this:

=== Pyrula Dev Server ===
Target: app.py:support
Using in-memory store - history clears on exit; --valkey-url to persist
UI: http://127.0.0.1:7978/ui

Press Ctrl+C to stop

  support  submitted  ae43518948a0429cb2b0f62a341c4a8a
  tool  lookup_order
  completed - Order A-1042 has shipped. Total $42.00.

support calls lookup_order, the fake model reads the result back, and the turn completes with the reply as its result. The server keeps running after that: the /ui page lists the finished turn, and you can submit more from there or over HTTP against /agents. The startup turn is a demonstration, not the whole session.

If your own agent’s parameters have no defaults, the startup turn can fail (a missing required argument, for instance). That doesn’t take the server down: pyrula dev prints one line saying the turn didn’t complete and why, then keeps serving so you can submit a turn with the arguments it needs.

Write it yourself

Prefer to see the pieces without the scaffold? An agent is an async function decorated with @agent. It takes a context (ctx) and runs on the workflow engine, so the same replay and recovery rules apply.

This example uses a fake LLM and an in-memory store, so it runs with no API key and no broker.

import asyncio
from pyrula.agents import AgentContext, agent, data_step
from pyrula.agents.testing import AgentRunner, FakeLLM, FakeResponse, InMemoryStore


@agent(timeout=30)
async def greet(ctx: AgentContext, name: str) -> str:
    await ctx.emit("started", {"name": name}, id="started")

    async def lookup() -> str:
        return "42"

    # data_step records the result so a replay won't run lookup again
    answer = await data_step(ctx, "lookup", lookup)
    return f"hello, {name} ({answer})"


async def main() -> None:
    runner = AgentRunner(agent=greet, store=InMemoryStore(), llm=FakeLLM([FakeResponse(text="hi")]))
    events = await runner.run(name="ada")
    done = [e for e in events if e.kind == "run:complete"]
    print(done[0].payload["result"])   # hello, ada (42)


asyncio.run(main())

AgentRunner is the in-process harness for tests. The events it returns are the same run:* and step:* events the engine writes to the stream (see Runs as Event Streams).

Side effects go through the context

data_step and ctx.emit are durable. On a replay, a step that already finished replays its recorded result instead of running again. The harness can prove it:

runner = AgentRunner(agent=greet, store=InMemoryStore(), llm=FakeLLM([FakeResponse(text="hi")]))
runner.inject_crash_at("data", nth=1, when="done")   # crash after lookup, before step:done
await runner.run(name="ada")                          # first attempt dies mid-step
await runner.run(name="ada")                          # resumes, lookup runs once total

That’s the whole point of building on the engine. You write straight-line async code and get crash-resume for free, as long as side effects go through ctx. See Determinism.

Quickstart

What it wrote

What pyrula dev does

Write it yourself

Side effects go through the context

Next

What `pyrula dev` does