One class to configure, one call to run

import asyncio
from effgen import Agent, AgentConfig, tool


@tool
def shipping_cost(weight_kg: float, express: bool = False) -> str:
    """Quote a shipping cost in euros for a parcel."""
    price = 4.50 + 1.75 * weight_kg + (9.0 if express else 0.0)
    return f"EUR {price:.2f}"


result = asyncio.run(shipping_cost.execute(weight_kg=2.5, express=True))
print("success:", result.success)
print("output: ", result.output)
print("error:  ", result.error)
print("took:    %.4fs" % result.execution_time)

with Agent(AgentConfig(model="openai:gpt-5-nano", tools=[shipping_cost])) as agent:
    r = agent.run("How much to send a 2.5 kg parcel by express?")
print(r.output)
print("calls:", r.tool_calls.names)

success: True
output:  EUR 17.88
error:   None
took:    0.0005s
EUR 17.88 for express shipping of a 2.5 kg parcel.

If you'd like, I can check the standard (non-express) rate or other options as well.
calls: ['shipping_cost']

from effgen import Tool


def seat_count(rows: int, seats_per_row: int) -> int:
    """Count the seats in a rectangular block of a theatre."""
    return rows * seats_per_row


seats = Tool.from_function(seat_count, category="computation")
print(seats.name, "·", seats.description)
print(seats.metadata.category.value)
for parameter in seats.metadata.parameters:
    print(f"  {parameter.name}: {parameter.type.value} "
          f"(required={parameter.required}) — {parameter.description}")

seat_count · Count the seats in a rectangular block of a theatre.
computation
  rows: integer (required=True) — The rows parameter.
  seats_per_row: integer (required=True) — The seats_per_row parameter.

from effgen import (Agent, AgentConfig, MultiAgentOrchestrator,
                    OrchestrationPattern, load_model)

model = load_model("openai:gpt-5-nano")
writer = Agent(AgentConfig(name="writer", model=model,
                           system_prompt="Write one plain sentence."))
editor = Agent(AgentConfig(name="editor", model=model,
                           system_prompt="Shorten what you are given. Reply with the sentence only."))

orchestrator = MultiAgentOrchestrator()
team = orchestrator.create_team(
    "copy", [writer, editor], pattern=OrchestrationPattern.SEQUENTIAL,
)
result = orchestrator.assign_task("Describe what a checkpoint is.", team)

print(result.output)
print("pattern:", result.pattern.value, "· agents:", len(result.agent_responses))
for stage in result.agent_responses:
    print(f"  {stage['agent_name']}: {stage['output'][:60]}"
          f" ({stage['tokens_used']} tokens, ${stage['cost_usd']:.6f})")
writer.close()
editor.close()

A checkpoint is a saved state that can be restored later.
pattern: sequential · agents: 2
  writer: A checkpoint is a saved state or snapshot of a process, syst (436 tokens, $0.000168)
  editor: A checkpoint is a saved state that can be restored later. (514 tokens, $0.000190)

import tempfile

from effgen import (Agent, AgentConfig, AgentMiddleware, FileCheckpointStore,
                    WorkflowDAG, WorkflowNode, load_model)


class FailOnce(AgentMiddleware):
    """Stand in for the process being killed part-way through."""

    def __init__(self) -> None:
        self.armed = True

    def before_run(self, ctx):
        if self.armed:
            self.armed = False
            raise RuntimeError("the box went away")
        return None


model = load_model("openai:gpt-5-nano")
short = "Reply with one short sentence."
research = Agent(AgentConfig(name="research", model=model, system_prompt=short))
draft = Agent(AgentConfig(name="draft", model=model, system_prompt=short))
breaker = FailOnce()
review = Agent(AgentConfig(name="review", model=model, system_prompt=short,
                           middleware=[breaker]))

store = FileCheckpointStore(tempfile.mkdtemp())


def build() -> WorkflowDAG:
    dag = WorkflowDAG("report")
    dag.add_node(WorkflowNode(id="research", agent=research))
    dag.add_node(WorkflowNode(id="draft", agent=draft))
    dag.add_node(WorkflowNode(id="review", agent=review))
    dag.connect("research", "draft")
    dag.connect("draft", "review")
    return dag


def show(label, result):
    print(f"{label}:", [(n["id"], n["status"], n["execution_time"])
                        for n in result.node_results], "success:", result.success)


show("first run ", build().run("Summarise what a DAG is.", checkpoint=store, run_id="q3"))
print("checkpoint:", "completed", sorted(store.load("q3").completed),
      "failed", sorted(store.load("q3").failed))
show("resumed   ", build().run("Summarise what a DAG is.", checkpoint=store, run_id="q3"))

for agent in (research, draft, review):
    agent.close()

first run : [('research', 'completed', 3.672), ('draft', 'completed', 3.817), ('review', 'failed', 0.002)] success: False
checkpoint: completed ['draft', 'research'] failed ['review']
resumed   : [('research', 'completed', 0.0), ('draft', 'completed', 0.0), ('review', 'completed', 4.392)] success: True