Skip to content

Quickstart

pip install pyrula-agents. 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.

Next