Quickstart

pip install "pyrula[agents]"
pyrula init --agent
pyrula dev app.py --open

The packages require a source build until their first public release; see Install.

Generated agent

pyrula init --agent writes app.py: a support agent with one tool, lookup_order. It runs offline against a scripted fake model, so it needs no API key or broker.

To use a real model, change the llm = FakeLLM(...) line:

from pyrula.agents.llm import AnthropicLLM
llm: Any = AnthropicLLM(model="claude-sonnet-4-6")   # reads ANTHROPIC_API_KEY

from pyrula.agents.llm import OpenAILLM
llm: Any = OpenAILLM(model="gpt-6-luna")             # reads OPENAI_API_KEY

For other providers install "pyrula[agents,litellm]" and use LiteLLM with any LiteLLM model string. That extra does not install on Python 3.15 yet.

Run the agent

pyrula dev app.py --open starts a local server, submits one turn to support, and opens the dev UI.

=== Pyrula Dev Server ===
Target: app.py:support
Using SQLite store .pyrula/dev.db - runs survive a restart; --fresh starts empty
Cloud: off (no PYRULA_CLOUD_API_KEY) - running locally only
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.
Saved the last run as .pyrula/last/support.case. It holds the run's inputs, every event and every recorded model and tool result, in full; .pyrula/ is local state, keep it out of version control.

The server keeps running; submit more turns from /ui or over HTTP at /agents. If the agent’s parameters have no defaults, the startup turn fails with one printed line and the server keeps serving.

See it in Cloud

Copy your environment’s key from Setup, export it and start dev again. It prints Cloud: connected to ... and, once events are accepted, Cloud: run <run-id> -> <link>.

export PYRULA_CLOUD_API_KEY="<key>"
pyrula dev app.py

See the Cloud quickstart for full output and troubleshooting.

Minimal agent

An @agent is an async function that runs on the workflow engine, with the same replay and recovery rules. This one uses a fake LLM and in-memory store.

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


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


@agent(timeout=30)
async def greet(ctx: AgentContext, name: str) -> str:
    await ctx.emit("started", {"name": name}, id="started")
    answer = await 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 test harness. runner.run(...) returns the run:* and step:* events the engine wrote (Runs as Event Streams). A sleeping run stays asleep: runner.run(...) before the wake time returns the same run:sleeping, so test with a short ctx.sleep and wait it out.

Durable side effects

On replay a completed @step returns its recorded result instead of running again:

async def prove_it() -> None:
    runner = AgentRunner(agent=greet, store=InMemoryStore(), llm=FakeLLM([FakeResponse(text="hi")]))
    runner.inject_crash_at("named", nth=1, when="pending")  # crash before lookup starts
    await runner.run(name="ada")                           # first attempt stops at the boundary
    await runner.run(name="ada")                           # resume runs lookup once

asyncio.run(prove_it())

See Determinism for the replay requirements.

Next

• Defining agents, tools, calling an LLM
• Serving over HTTP
• Bring your agent: make a Pydantic AI agent durable, or use LangChain tools and LangGraph.
• Workers and the CLI. pyrula worker refuses to start without --llm-ref when an agent declares @agent(llm=...), as the scaffold does.