The effector

async def refund(ctx: ActivationContext) -> Complete | Suspend:
    """Request a refund, then report what the effector did.

    Two activations, one logical unit of work. `ctx.act` stages the intent and
    returns its deterministic id; `Suspend` persists the continuation and arms
    the fail-closed timeout. Nothing has been written to the outside world when
    this function returns — the intent is a request, not an effect.
    """
    if not ctx.is_resume:
        ctx.act("payments.refund", '{"amount": 4200}', ttl_ms=INTENT_TTL_MS)
        return Suspend(snapshot=b"awaiting-refund", adapter="example", timeout_ms=30_000)

    # On resume the same activation continues: same key, same seq, and the
    # snapshot it persisted is available as ctx.snapshot.
    assert ctx.resume_result is not None
    return Complete(output=b"refunded:" + ctx.resume_result.payload)

The effector is the external service that closes the effects loop:

Every edge in the loop is dashed because none of it is an edge in the Beam graph: the cycle closes through the message bus, outside the pipeline. The phase order under the effector is the crash argument — each boundary is explained in Behavior reference.

It consumes ToolIntents, refuses expired ones, dedups on intent_id, executes side-effecting tools from your ToolRegistry, and publishes one ToolResult per intent keyed by entity_key. It runs outside the pipeline: it imports no Beam and no beam_agents.core.

uv pip install 'beam-agents[effector]'

beam-agents-effector \
  --registry myapp.agent:TOOLS \
  --intents-from kafka://broker:9092/intents \
  --results-to   kafka://broker:9092/results \
  --approvals-to kafka://broker:9092/approvals \
  --dedup        redis://redis:6379 \
  --consumer-group effector

Every flag also reads an environment variable (EFFECTOR_REGISTRY, EFFECTOR_INTENTS_FROM, …). --registry is an import path (module:attribute) naming the same ToolRegistry your agent declares its tools from — the effector executes your tools, so it has to load them.

Anything security-relevant — intent-signature verification (--verify-intents, --signing-keys, --dead-letters-to), broker SASL/TLS, and the rule that secrets never travel in tool arguments — lives in Security. Credentials belong in environment variables rather than flags: argv is visible in process listings.

Deployment preconditions

The effector cannot enforce these; a deployment that skips them loses guarantees silently.

Bigtable table provisioning, for reference:

cbt createtable effector-dedup families="d:maxage=25h"

Budgets

What is guaranteed, and what is not

Guaranteed: at most one dispatch per intent_id, one agreed terminal result per intent_id, per-key execution order, and no intent lost (offsets commit only after the result is published).

Not guaranteed by the effector alone: exactly-once effects. Two windows remain, and both are inherent rather than fixable inside the effector:

1. Lease expiry. If a worker is alive but partitioned from the dedup store for longer than --lease-ms, another worker can re-claim and re-run its intent. The budget rule above makes this require a genuine partition, not just a slow tool.
2. Crash mid-tool. A worker killed after invoking a tool cannot know whether the effect landed. Its claim is left to expire (never handed back), and the redelivery re-executes.

The exactly-once contract

The honest statement is two-sided. The runtime guarantees:

• Deterministic intent IDs. intent_id is uuid5-derived from entity_key + seq + step_index; a replayed bundle, a duplicated sink write, and a lease-expiry re-execution all carry the byte-identical id.
• At most one completed execution per intent_id. The dedup store admits one terminal result; every other delivery is collapsed or handed the stored result.

What that adds up to depends on the tool:

• A tool that keys its downstream effect on the intent id — a Stripe Idempotency-Key, a Redis SETNX, a keyed upsert, an INSERT … ON CONFLICT DO NOTHING — gets exactly-once effects: any crash-window re-invocation replays the same key and the downstream deduplicates it.
• A tool that does not is at-least-once across crash recovery: zero lost effects, duplicates only within the two windows above, strict exactly-once when no worker dies mid-tool.

To opt in, declare a keyword-only intent: IntentInfo parameter — the effector injects the executing intent's identity (intent_id, entity_key, seq, step_index, attempt) at invocation time. The parameter never appears in the tool's provider-facing schema and cannot be supplied through args_json (a spoofed intent key is rejected before the callable runs):

from beam_agents.tools import IntentInfo, tool


@tool(side_effect=True)
async def charge(customer_id: str, amount_cents: int, *, intent: IntentInfo) -> str:
    # Replays and crash-window re-executions re-mint the same intent_id,
    # so the payment provider performs the charge exactly once.
    return await payments.charge(
        customer_id, amount_cents, idempotency_key=intent.intent_id
    )

Declaring the parameter on a side_effect=False tool is a ToolDefinitionError at decoration time: read-only tools run inline in the pipeline, where no intent exists.

Behavior reference

Phase order per intent — verify → refuse-expired → claim → execute → complete → publish → commit. Each edge is a crash argument: an unauthenticated message drives nothing at all, not even an EXPIRED result (see Security); expiry is decided before the store is touched (an outage cannot make a deadline fail open); the result is durable before it is published (a crash republishes rather than re-executes); the offset advances last (a crash redelivers rather than loses).

The verify step is inert until you turn it on: --verify-intents defaults to off, which is byte-for-byte the pre-signing behavior.

Statuses. REJECTED means the callable never ran (unknown tool, a side_effect=False tool reaching the outbox, argument validation failure). ERROR means it ran and the effect is unknown (raised, timed out, or returned something unencodable). EXPIRED means past expires_at_ms — including expires_at_ms == 0, which reads as expired, never as unbounded. OK carries the return value as canonical JSON.

Approvals. An intent with kind = APPROVAL is published verbatim to --approvals-to and never executed; no ToolResult is published, because the answer comes back as an AgentEnvelope.Approval on the approvals topic. It is marked terminal so a redelivery cannot notify twice. Expired approvals are refused like any other intent. Fronting that channel is your job, not the runtime's — see the Slack approval surface example for a worked one (Block Kit buttons, verdict → envelope keyed by entity_key, fail-closed TTL at the surface).

A busy claim blocks, it does not skip. If another worker holds a live lease, the effector waits (bounded by the lease) rather than committing past an intent it never executed. Skipping would be the one failure mode that silently drops an effect.

Shutdown

SIGINT/SIGTERM stops consumption and cancels in-flight partitions. Claims whose tool had not yet been invoked are released immediately, so a restarting replica does not wait out their leases; a claim whose tool was invoked is left to expire, since the effect may already have happened. Nothing uncommitted is lost — it redelivers.

Testing

• Offline (make test-unit): the whole loop runs against in-memory source, sink, and dedup fakes. No docker, no network, no client libraries installed.
• Semantics (make test-semantics-offline): the effectively-once gate replays an intent stream with kills injected at every phase boundary.
• Integration (make compose-up && make test-integration): Redpanda + Redis end-to-end, the Pub/Sub emulator for ordered delivery, and the Bigtable emulator for the conditional-claim semantics. The DedupStore conformance suite runs against all three backends, so a store that passes is substitutable for the others.