Python SDK

The apomesh Python SDK wraps the daemon’s gRPC API in Pydantic-typed substrate models. You dispatch a session with one call, iterate the live event stream with async for, and read the terminal outcome and cost rollup — without touching gRPC plumbing. This page walks from install to a complete run, then shows the deep-research helper.

Prerequisites

• Python 3.11+
• A running apomesh daemon to talk to. Bring up the local stack, or run the orchestrator directly from the workspace root: cargo run -p apomesh-orchestrator. The daemon’s gRPC listener is 50051.

Install

The SDK is not published to a package index yet — install it from source with uv:

cd sdks/python
uv sync --all-extras

Connect

Every client targets a daemon endpoint (host:port). On loopback the channel is cleartext — the development default. For anything beyond localhost, pass a grpc.ChannelCredentials via the credentials= parameter; the SDK reuses gRPC’s credential vocabulary rather than inventing its own.

import grpc
from apomesh import SessionClient

# Dev: insecure loopback (cleartext).
async with SessionClient("127.0.0.1:50051") as client:
    ...

# TLS with system root CAs (public cert).
async with SessionClient(
    "daemon.example.com:443",
    credentials=grpc.ssl_channel_credentials(),
) as client:
    ...

SessionClient is the low-level transport. For driver code, prefer the dispatch helper below — it constructs the client for you.

Dispatch a session and subscribe to events

apomesh.dispatch starts a session against a daemon and returns an ergonomic Session handle. You build the planner-root AgentState, then iterate the live event tail until the session’s root reaches a terminal lifecycle.

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_planner() -> 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="Summarize the top three headlines today."),
        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,
            locality=Locality.CLOUD_ALLOWED,
        ),
        tools=[ToolConcreteId(concrete_id="web.search")],
        placement=Placement(kind=PlacementKind.CLOUD),
        supervisor_strategy=SupervisorStrategy.ESCALATE,
    )


async def main() -> None:
    async with await dispatch(
        build_planner(), ENDPOINT, SessionKind.GENERIC, tenant_id=TENANT_ID
    ) as session:
        print(f"session {session.session_id}")

        # The live tail follows like `tail -f`, and it ends on its own when the
        # session reaches its terminal event. Breaking early is still useful when
        # you only need to watch for one thing.
        async for envelope in session.events():
            if is_terminal_for_root(envelope, session.root_agent_ref):
                break

        outcome = await session.wait_for_completion()
        print(f"exit: {outcome.exit_condition.kind} reason={outcome.reason}")

        cost = await session.cost_breakdown()
        print(f"total cost: {cost.total_cost_usd}")


asyncio.run(main())

Caution

With no strategy binding, dispatch must be given a concrete kind — omitting it (the UNSPECIFIED default) is rejected by the daemon with INVALID_ARGUMENT. Name a kind, or bind a manifest_id / published_agent / strategy_config.

session.wait_for_completion() returns a SessionOutcome — its exit_condition (an ExitCondition whose .kind is one of GOAL_MET, CANCELLED, FAILED, ESCALATED, UNSPECIFIED), the strategy’s terminal_output bytes, and a human reason. session.cost_breakdown() queries the daemon’s canonical cost rollup on demand; session.cancel(reason) requests graceful subtree cancellation.

Resuming a tail

The tail is lossless and resumable: every event carries the log position that identifies it. session.events() keeps yielding bare envelopes — the position is bookkeeping most callers never touch — and exposes the cursor as session.last_position. Persist it, and pass it back after a disconnect to receive exactly the remainder, with no gap and no duplicate.

Callers that want the position inline can iterate the lower-level client instead, which yields (position, envelope) pairs and accepts a from_position:

from apomesh.clients.daemon import SessionClient

async with SessionClient(ENDPOINT) as client:
    cursor = b""
    async for position, envelope in client.tail_events(session_id, from_position=cursor):
        handle(envelope)
        cursor = position  # durable consumers checkpoint this alongside their own state

The token is opaque — treat it as bytes rather than parsing it.

Handle errors by category

The SDK raises typed errors in three orthogonal categories — catch at the level that matches your recovery:

from apomesh import (
    ApomeshError,            # abstract parent of all three
    ApomeshOperationalError, # daemon unreachable / mis-configured / transient
    ApomeshVersionError,     # SDK and daemon disagree on the wire vocabulary
    ApomeshSemanticError,    # resource in a state that doesn't admit the call
)

Raw gRPC status codes are translated at the transport boundary — drivers never see grpc.aio.AioRpcError directly (it is preserved as __cause__).

Drive the deep-research vertical

apomesh.deep_research.run composes dispatch with a deep-research planner and parses the synthesized report into a typed DeepResearchReport:

import asyncio

from apomesh.deep_research import DeepResearchConfig, SynthesisDepth, run


async def main() -> None:
    report = await run(
        "What are the leading approaches to durable agent state?",
        "127.0.0.1:50051",
        budget_dollars="10.00",
        tenant_id="dev",
        config=DeepResearchConfig(synthesis_depth=SynthesisDepth.COMPREHENSIVE),
    )
    print(report.model_dump_json(indent=2))


asyncio.run(main())

See Deep Research for the vertical’s shipped shapes, launch options, and the full set of DeepResearchConfig tuning knobs.

Full API surface

This page covers the driver-facing path. The complete public surface — every client (RegistryClient, StrategyConfigClient, McpAdminClient, TokenAdminClient, and more), the typed model set, and the conversion helpers — is the SDK source itself: sdks/python/apomesh and its README. The wire protocol is in greenfield posture and may change across releases until the substrate reaches “stable within a major.”