Distributed usage

Distributed runtime support is easiest to approach in two modes:

use the managed entrypoint when you want the normal runtime API and one primary worker behind it
use explicit hosts and workers when you need to control topology yourself

Most applications should start with distributed_runtime(...), which yields a managed DistributedRuntimeEngine. Reach for WorkerAgentRuntime and WorkerAgentRuntimeHost directly only when worker placement or topology needs to be explicit.

Start With The Managed Entry Point

Use distributed_runtime() when you want to keep the runtime mental model unchanged. You still call send_message(...), publish_message(...), register_factory(...), and the other normal runtime APIs. The difference is that a host and a primary worker are created for you behind the scenes.

from agentlane.messaging import AgentId
from agentlane.runtime import (
    BaseAgent,
    Engine,
    MessageContext,
    distributed_runtime,
    on_message,
)


class CounterAgent(BaseAgent):
    def __init__(self, engine: Engine) -> None:
        super().__init__(engine)
        self._count = 0

    @on_message
    async def handle(self, payload: str, context: MessageContext) -> object:
        _ = payload
        _ = context
        self._count += 1
        return {"count": self._count}


async with distributed_runtime() as runtime:
    runtime.register_factory("counter", CounterAgent)

    first = await runtime.send_message(
        "one",
        recipient=AgentId.from_values("counter", "session-1"),
    )
    second = await runtime.send_message(
        "two",
        recipient=AgentId.from_values("counter", "session-1"),
    )

    assert first.status.value == "delivered"
    assert second.response_payload == {"count": 2}

This is the right default when:

you want distributed execution without designing your own topology
one host plus one primary worker is enough
you want the same mental model as the in-process runtime

Move To Explicit Hosts And Workers When Placement Matters

Use explicit WorkerAgentRuntimeHost and WorkerAgentRuntime instances when you want different agent types to live on different workers.

The important design idea is that the host chooses the destination worker, but the chosen worker still runs the delivery locally with its own registry, scheduler, and agent instances.

Direct Send Across Workers

from agentlane.messaging import AgentId
from agentlane.runtime import (
    BaseAgent,
    MessageContext,
    WorkerAgentRuntime,
    WorkerAgentRuntimeHost,
    on_message,
)


class EchoAgent(BaseAgent):
    @on_message
    async def handle(self, payload: str, context: MessageContext) -> object:
        _ = context
        return {"agent_id": self.id.key.value, "echo": payload}


host = WorkerAgentRuntimeHost(address="127.0.0.1:0")
await host.start()

caller = WorkerAgentRuntime(host_address=host.address, address="127.0.0.1:0")
echo_worker = WorkerAgentRuntime(host_address=host.address, address="127.0.0.1:0")

echo_worker.register_factory("echo", EchoAgent)

await caller.start()
await echo_worker.start()

try:
    outcome = await caller.send_message(
        "ping",
        recipient=AgentId.from_values("echo", "session-1"),
    )
    assert outcome.status.value == "delivered"
finally:
    await caller.stop_when_idle()
    await echo_worker.stop_when_idle()
    await host.stop_when_idle()

The caller does not need to host the target agent type. The host resolves the destination worker and forwards the delivery there.

Publish Across Workers

Publish follows the same high-level pattern, but the host resolves all matching subscriptions before it forwards concrete deliveries to workers.

from agentlane.messaging import TopicId
from agentlane.runtime import (
    BaseAgent,
    MessageContext,
    WorkerAgentRuntime,
    WorkerAgentRuntimeHost,
    on_message,
)


class ListenerAgent(BaseAgent):
    @on_message
    async def handle(
        self,
        payload: dict[str, object],
        context: MessageContext,
    ) -> object:
        _ = context
        print(f"listener received: {payload}")
        return None


host = WorkerAgentRuntimeHost(address="127.0.0.1:0")
await host.start()

publisher = WorkerAgentRuntime(host_address=host.address, address="127.0.0.1:0")
listener = WorkerAgentRuntime(host_address=host.address, address="127.0.0.1:0")

listener.register_factory("listener", ListenerAgent)
listener.subscribe_exact(topic_type="alerts", agent_type="listener")

await publisher.start()
await listener.start()

try:
    ack = await publisher.publish_message(
        {"event": "ready"},
        topic=TopicId.from_values(type_value="alerts", route_key="session-1"),
    )
    assert ack.enqueued_recipient_count == 1
finally:
    await publisher.stop_when_idle()
    await listener.stop_when_idle()
    await host.stop_when_idle()

What To Keep In Mind

The current distributed implementation is intentionally narrow:

distributed_runtime() manages one in-process host plus one primary worker
a given AgentType is owned by one worker at a time
PublishAck still means enqueue only, not handler completion
cancellation does not yet propagate across the process boundary

Those are current implementation facts, not just usage advice.

A few distributed-only behaviors are worth knowing when wiring or debugging:

send_message(...) and publish_message(...) require a connected host and raise RuntimeError if the worker is not connected to one
unlike the in-process path, distributed publish_message(...) surfaces host or transport failures as a raised RuntimeError rather than a failed ack
registering a factory/instance or subscribing on a running worker schedules an asynchronous catalog sync (a full agent-type plus subscription snapshot) to the host; send_message/publish_message await the pending sync before routing, so freshly registered types become routable as soon as the call proceeds
WorkerAgentRuntime exposes address (the resolved bind endpoint) and worker_id (assigned after registration) for inspecting bound workers