Skip to content
apomesh

First SDK dispatch

Here you write your own dispatch from scratch: a small Python script that starts a session, streams its live events, and reads the terminal outcome — the same API the shipped sample uses.

Prerequisites

Section titled “Prerequisites”

The minimal dispatch

Section titled “The minimal dispatch”

Save this as my_dispatch.py in the repo root. It builds a planner-root agent, dispatches it, prints each event as it arrives, and stops on the root’s terminal event:

import asyncio


from apomesh import dispatch, is_terminal_for_root
from apomesh.types import (
    AccessMode,
    AgentRef,
    AgentState,
    Budget,
    CapabilityTier,
    Currency,
    Goal,
    LifecycleMode,
    Locality,
    MemoryScope,
    MonetaryAmount,
    Placement,
    PlacementKind,
    SessionKind,
    SupervisorStrategy,
    ToolConcreteId,
)


ENDPOINT = "127.0.0.1:50051"
TENANT_ID = "dev"




def build_agent() -> AgentState:
    return AgentState(
        id=AgentRef(tenant_id=TENANT_ID, agent_uuid="my-planner-root"),
        capability_tier=CapabilityTier.PLANNER,
        lifecycle_mode=LifecycleMode.STATELESS,
        goal=Goal(
            description="Fix the failing test in tests/test_calculator.py by "
            "correcting the bug in src/calculator.py, then run pytest to verify."
        ),
        budget=Budget(
            tokens=200_000,
            cost=MonetaryAmount(micros=2_000_000, currency=Currency.USD),  # $2.00
        ),
        memory_scope=MemoryScope(
            episodic=AccessMode.READ_WRITE,
            semantic=AccessMode.READ,
            procedural=AccessMode.READ_WRITE,
            locality=Locality.CLOUD_ALLOWED,
        ),
        tools=[
            ToolConcreteId(concrete_id=name)
            for name in ("file.read", "file.write", "shell.exec", "pytest")
        ],
        placement=Placement(kind=PlacementKind.CLOUD),
        supervisor_strategy=SupervisorStrategy.SELECTIVE_REPLAN,
    )




async def main() -> None:
    async with await dispatch(
        build_agent(), ENDPOINT, SessionKind.GENERIC, tenant_id=TENANT_ID
    ) as session:
        print(f"started session {session.session_id}")
        async for envelope in session.events():
            print(envelope)
            if is_terminal_for_root(envelope, session.root_agent_ref):
                break
        outcome = await session.wait_for_completion()
        print(f"terminal: exit_condition={outcome.exit_condition}")




if __name__ == "__main__":
    asyncio.run(main())

Run it

Section titled “Run it”
Terminal window
uv run --project sdks/python python my_dispatch.py

What the code does

Section titled “What the code does”
  • dispatch(root_agent, endpoint, kind, ...) starts a session against the daemon and returns a Session — an async context manager that closes the transport when the block exits.
  • session.events() yields live event envelopes as the session runs. The live tail never closes on its own, so the loop breaks on the root’s terminal event via is_terminal_for_root(envelope, session.root_agent_ref) — otherwise it would follow forever.
  • session.wait_for_completion() returns the terminal SessionOutcome, whose exit_condition tells you how the session ended (GOAL_MET, FAILED, CANCELLED, ESCALATED).

Where to go next

Section titled “Where to go next”
  • The Python SDK guide covers the full API surface — the typed error vocabulary, cost queries, TLS/mTLS transport, and the higher-level helpers.
  • The Deep Research guide shows the apomesh.deep_research.run helper, which composes dispatch with a research planner and parses a typed report.
  • The Concepts section explains why the substrate is shaped the way it is — hierarchy, sessions and events, the strategy plane, and more.