Python SDK

pip install meerkat-meta

[mob]
id = "python-quickstart"

[profiles.worker]
model = "gpt-5.5"
runtime_mode = "turn_driven"

[profiles.worker.tools]
comms = true

import asyncio
from meerkat_meta import Meta

async def main():
    async with await (
        Meta.builder()
        .mob("config/mob.toml")
        .gateway("/path/to/rpc_gateway")
        .console_auth_required(False)
        .build()
    ) as rt:
        handle = rt.mob_handle()

        # Send a message (comms)
        await handle.ensure_member("agent-1", "worker")
        await handle.send("agent-1", "Hello")

        # Watch what happens (observation)
        async for event in handle.subscribe_agent("agent-1"):
            print(event.event_type, event.event)

asyncio.run(main())

from meerkat_meta import DetachedJobExecution, DetachedJobResult

async def run_scan(ctx):
    await ctx.progress(1, "scan started")
    # ctx.arguments is the persisted non-secret specification.
    # ctx.credentials is resolved for this committed attempt only.
    result_ref = await perform_scan(ctx.arguments, ctx.credentials)
    return DetachedJobResult(result_ref)

class AppAgentBuilder:
    async def build_agent(self, options):
        options.register_tool(
            "security_scan",
            lambda args: {"error": "detached owner required"},
            description="Run a durable security scan",
            execution=DetachedJobExecution(
                runner="example.security_scan",
                version="v1",
                restart_class="checkpoint_resumable",
                idempotency_scope="interaction_and_arguments",
                submission_timeout_ms=30_000,
                credential_scopes=("unifi.read",),
                handler=run_scan,
            ),
        )

runtime = await (
    Meta.builder()
    .persistent_state("state/meta")
    .job_credentials(resolve_profile_credentials)
    .session_service(AppAgentBuilder())
    .gateway("/path/to/rpc_gateway")
    .build()
)

job = await runtime.jobs.get(job_id)
await runtime.jobs.cancel(job_id)

from meerkat_meta import Meta, TurnOutputStatus
from meerkat_meta.identity_first_models import DispatchInput, DurableAgentSpec, ManagedPeerEdge

# Boot with roster provider
rt = await (
    Meta.builder()
    .mob_inline(MOB_TOML)                 # or .mob("mob.toml")
    .persistent_state(state_dir)
    .agent_memory()                       # optional: durable per-agent memory
    .roster(my_roster)                    # provides DurableAgentSpec list
    .topology_provider(my_topology)       # optional: managed edges
    .agent_customizer(my_customizer)      # optional: per-identity prompts
    .gateway("./rpc_gateway")
    .build()
)

# Identity-scoped agent handles
triage = rt.agent("triage:main")
alice = rt.agent("identity:alice")

# Delivery
await alice.send("Hello")                                          # Addressable only
await triage.dispatch_text("School closed", origin="connector")   # Any identity

# On a turn_driven member (runtime_mode = "turn_driven"): send, then wait for
# THAT turn and return its own output. The send carries a turn ticket, so
# another delivery to the identity (a peer message, a scheduled turn, a fork
# completion wake) cannot satisfy the wait. On an autonomous_host member (the
# default mode) the send is still delivered once, but the turn cannot be
# tracked: the call waits identity-wide with a TurnTrackingUnavailableWarning,
# and send_and_wait_outcome(...).attributed is False (the text may be another
# turn's). require_attribution=True raises instead.
answer = await alice.send_and_wait("Hello", timeout=90)

# The same, in two steps: track the turn, then wait on its ticket.
sent = await alice.send("Hello", track_turn=True)
if sent.turn_ticket is not None:
    turn = await alice.wait_for_turn(sent.turn_ticket, timeout=90)
    if turn.output_status is TurnOutputStatus.TEXT:
        answer = turn.output  # turn.output_truncated says whether it was cut
elif sent.turn_unavailable is not None:
    # e.g. "autonomous_host"; delivered is False only for "not_delivered"
    print("untracked:", sent.turn_unavailable.code, sent.turn_unavailable.delivered)
else:
    print("untracked: the gateway predates turn tickets")

# Observation (no mob_handle, no member IDs needed)
inspection = await triage.inspect()   # output_preview, completion_cursor, peers
cursor = await triage.completion_cursor()  # cursor only; never reads the member session
output = await triage.wait_for_output(timeout=90)   # any output, including an earlier turn's
await rt.wait_until_ready(["triage:main", "identity:alice"])

# Status
status = await alice.status()          # state, session_id, generation, labels

# Lifecycle
await alice.reset()                    # Destructive: new generation, new session
await rt.agent("domain:calendar").respawn()  # Non-destructive recovery
await rt.reconcile()                  # In-process roster reconciliation

DurableAgentSpec(
    identity="archivist:main",
    profile="archivist",
    runtime_mode_override="turn_driven",      # pin this identity regardless of the profile default
    initial_message="Index yesterday's mail.", # or [TextBlock(...), ImageBlock(...)]
)

class MyRoster:
    async def roster(self, context):
        mob_id = (context.mob_definition or {}).get("id")
        known = set(context.previous_identities)
        ...

DispatchInput.system("text")
DispatchInput.connector("event from slack", correlation_id="msg-1")
DispatchInput.scheduler("nightly sync")

from meerkat_meta import DeliveryScope, ScopedRecoveryState, StaleScopeError

status = await runtime.status("domain:calendar")
scope = status.delivery_scope            # None: see status.delivery_scope_unavailable
saved = scope.to_dict()                  # persist byte-exact before dispatching

try:
    result = await runtime.dispatch_text(
        "domain:calendar", "occurrence 42",
        idempotency_key="occ-42", correlation_id=correlation,
        expected_scope=DeliveryScope.from_dict(saved),
    )
    result.receipt.stage                 # "ingress_accepted": not a durable-input claim
except StaleScopeError as error:
    error.mismatch                       # which atom moved; nothing was submitted

recovery = await runtime.recover_delivery(
    "domain:calendar", DeliveryScope.from_dict(saved),
    idempotency_key="occ-42", correlation_id=correlation,
)
if recovery.state is ScopedRecoveryState.ABSENT:
    ...                                  # a point-in-time miss, never permission to retry

result = await handle.send("agent-1", "Review PR #42")
print(result.accepted, result.member_id, result.session_id)

# Stream one agent's events
async for event in handle.subscribe_agent("agent-1"):
    print(event.event_type, event.event)

# Stream mob-wide events (all agents)
async for event in handle.subscribe_mob():
    print(event.member_id, event.event)

from meerkat_meta import EventQuery, MobEventsStaleError

# Latest 256 matching events
events = await handle.query_mob_events(EventQuery(
    event_types=["flow_started", "flow_completed"],
    mob_id="home",
))
for e in events:
    print(e.cursor, e.kind, e.run_id, e.data)

# Resume from a checkpointed cursor
try:
    new_events = await handle.query_mob_events(EventQuery(
        after_seq=last_seen_cursor,
    ))
except MobEventsStaleError as exc:
    # Cursor is past the current frontier — rewind.
    new_events = await handle.query_mob_events(EventQuery(
        after_seq=exc.latest_cursor,
    ))

# All runs across all flows
runs = await handle.list_runs()

# Filter to one flow
runs = await handle.list_runs(flow_id="onboarding")

for run in runs:
    print(run.run_id, run.status, len(run.step_ledger))

# Inspect runtime state
status = await handle.status()
print(status.running, status.loaded_modules)

# Ensure an EPHEMERAL worker exists (spawn if missing). Durable members
# (per-user agents, coordinators) belong on the identity plane instead —
# see "new user first contact" below.
snapshot = await handle.ensure_member("scratch-worker-7", role="worker")
print(snapshot.agent_identity, snapshot.state)

# Find members by label
members = await handle.find_members("owner", "user-123")

# Reconcile the named modules (routing, delivery). This does not touch the
# roster; identity roster reconciliation is rt.reconcile().
result = await handle.reconcile(["routing", "delivery"])

# Call an MCP tool on a loaded module
result = await handle.call_tool("gmail", "gmail_search", {"query": "is:unread"})

# Or bind a module for repeated calls
gmail = handle.tool_caller("google-workspace")
messages = await gmail("gmail_search", query="is:unread")

# List the full roster
members = await handle.list_members()
for m in members:
    print(f"{m.agent_identity}: {m.state} ({m.role})")

# Inspect a specific member
snapshot = await handle.get_member("agent-1")

# Member lifecycle
await handle.retire_member("agent-1")
await handle.respawn_member("agent-1")

# Non-destructive cold reload (same session, same generation) and the
# in-process health read that works while the actor loop is stalled
health = await handle.member_health("agent-1")
if health.last_delivery_error and health.last_delivery_error["class"] == "reload_required":
    outcome = await handle.reload_member("agent-1")
    assert outcome.disposition in ("discarded", "not_degraded")

# Slack DM from a new user — add the identity to your roster source,
# reconcile, then deliver by identity.
my_roster.add_identity(f"personal:{user_id}", profile="personal",
                       labels={"owner": user_id})
await runtime.reconcile()
result = await runtime.send(f"personal:{user_id}", slack_message)
async for event in runtime.mob_handle().subscribe_agent(f"personal:{user_id}"):
    forward_to_slack(event)

from meerkat_meta import Meta

rt = await (
    Meta.builder()
    .mob("config/mob.toml")
    .session_service(my_builder)
    .roster(my_roster)
    .gateway("./rpc_gateway")
    .build()
)

runtime = await (
    Meta.builder()
    .mob("config/mob.toml")
    .fork_instruction_policy("reactivate_inherited")
    .build()
)

from meerkat_meta import Meta
from meerkat_meta.config import event_log, runtime_store

runtime = await (
    Meta.builder()
    .mob("config/mob.toml")
    .persistent_state("state/meta")
    # Declare an in-memory runtime store (sessions do not survive restart).
    .runtime_store(runtime_store.memory())
    # Declare a bounded queryable in-process event log — or event_log.null().
    .event_log(event_log.memory(batch_size=64))
    .gateway("/path/to/rpc_gateway")
    .build()
)

from meerkat_meta import CompositionProvenanceError

try:
    runtime = await builder.build()
except CompositionProvenanceError as err:
    if err.kind == "divergent":
        print("mob.toml differs from the stored definition in", err.fields)
    raise

runtime = await (
    Meta.builder()
    .mob("config/mob.toml")
    .gateway("./rpc_gateway")
    .console_auth_required(False)  # explicit opt-out: local console, no identity provider
    .console_config("config/console.toml")
    .build()
)

# Context manager (recommended)
async with await Meta.builder().mob("mob.toml").gateway("./rpc_gateway").build() as rt:
    handle = rt.mob_handle()
    # ...

# Explicit lifecycle
rt = await Meta.builder().mob("mob.toml").gateway("./rpc_gateway").build()
try:
    handle = rt.mob_handle()
    # ...
finally:
    await rt.shutdown()

import os
from meerkat_meta import AdoptedGatewayChild, AdoptedGatewayTransport, Meta

child = AdoptedGatewayChild(
    input=os.fdopen(stdin_write_fd, "wb", buffering=0),   # writes the child's stdin
    output=os.fdopen(stdout_read_fd, "rb", buffering=0),  # reads the child's stdout
    request_retirement=owner.request_retirement,
    wait_process=owner.wait_process,
)
transport = AdoptedGatewayTransport(child)
runtime = await Meta.builder().mob("mob.toml").gateway_transport(transport).build()

from meerkat_meta import SessionAgentBuilder, SessionBuildOptions

class MyBuilder(SessionAgentBuilder):
    async def build_agent(self, options: SessionBuildOptions) -> None:
        # options.profile_name arrives pre-filled from the spawn role.
        options.register_tool("calendar_lookup", my_calendar_handler)
        if options.profile_name == "security":
            options.register_tool(
                "unifi_scan",
                my_scan_handler,
                description="Scan the UniFi controller",
                input_schema={"type": "object", "properties": {"site": {"type": "string"}}},
            )

runtime = await (
    Meta.builder()
    .mob("config/mob.toml")
    .session_service(MyBuilder())
    .gateway("./rpc_gateway")
    .build()
)

from meerkat_meta import (
    AgentBuildContext,
    AgentBuildDraft,
    AgentCustomizerProtocol,
    DurableAgentSpec,
)

class MyCustomizer(AgentCustomizerProtocol):
    async def customize_build(
        self,
        context: AgentBuildContext,
        spec: DurableAgentSpec,
        draft: AgentBuildDraft,
    ) -> None:
        # context.identity, context.active_peers, context.managed_edges are read-only.
        if spec.profile == "security":
            draft.register_tool("unifi_scan", my_scan_handler, description="Scan the UniFi controller")
        draft.additional_instructions.append(f"Your wired peers: {', '.join(context.active_peers)}")

draft.provider_params = {
    "provider_tag": {
        "provider": "anthropic",
        "cache_control": "automatic",   # disabled | automatic | system_prefix | system_and_conversation
        "cache_ttl": "1h",              # 5m | 1h; effective only with a cache_control policy
    },
}

from meerkat_meta import (
    SourceStaleRevisionError,
    SourceVersionConflictError,
    supports_announcement_sources,
)

handle = runtime.mob_handle()
if not supports_announcement_sources(await handle.capabilities()):
    raise RuntimeError("this gateway does not serve announcement sources")

source = (await handle.announcement_source_get("mail")).source
if source is None:
    raise RuntimeError("declare the source first")
items, next_cursor = read_provider(source.cursor.token)  # your adapter
try:
    step = await handle.announcement_source_advance(
        f"{run_id}:v{source.cursor.version}",
        "mail",
        source.revision,
        source.cursor.version,
        principal,
        next_cursor,
        items=items,
        observed={"kind": "delta"},
    )
except (SourceStaleRevisionError, SourceVersionConflictError):
    pass  # another step or a configuration change moved the source: read it again

page = await handle.announcement_escalations_list(after=last_cursor)
for event in page.events or ():
    record(event.escalation_id, event.event, event.cause)  # your own
last_cursor = page.next

template = await handle.announcement_role_template_put(
    "op-reviewers", "reviewers", {"mob_id": "mob-a", "role": "reviewer"}, ["ops.alert"])
change = await handle.announcement_subscription_pause("op-pause", "reviewers", template.revision)
assert change.subscription is None and change.template.state == "paused"

from meerkat_meta.errors import (
    MetaError,           # Base exception
    TransportError,        # Transport layer failures
    RpcError,              # JSON-RPC error responses
    NotConnectedError,     # No active runtime connection
    CapabilityUnavailableError,  # Requested capability unavailable
    ContractMismatchError,       # SDK/runtime version mismatch
    StorageResolutionError,      # Fail-closed storage refusal at init (-32014)
    CompositionProvenanceError,  # Mob composition provenance refusal at init (-32019): .kind, .fields
    InitOutcomeUnknownError,     # Init written, settlement never arrived (native state may have changed)
    TransportReaderFailedError,  # The transport reader stopped; no response can arrive
    GatewayChildEndedError,      # The gateway child ended and takes no more requests (.reason, .exit_code); a TransportReaderFailedError
    IdentityUnresumableError,          # Core cannot resume the identity (-32051, final): .reason, .remedy
    IdentityHistoryUnavailableError,   # Core's history of the identity cannot be read now (-32052, retryable)
    IdentityHostAdoptedError,          # A host adopted the identity, so Meta's verbs leave it alone (-32053)
    IdentityWakeHeldError,             # The wake is held until an operator acts (-32603): .kind names the hold
    MobAnnouncementsUnavailableError,  # Mob announcements are not enabled (-32044)
    AnnouncementConflictError,         # Conflict or stale request (-32045): .kind
    SourceStaleRevisionError,          # A source's revision moved (-32045): .current_revision, .current_version, .generation, .cursor
    SourceVersionConflictError,        # A source's cursor version moved (-32045): .current_version, .cursor, .state, .revision
    AnnouncementFrontierError,         # Start frontier outside a mob's announcements (-32046): .kind
    AnnouncementRefusedError,          # Typed mob announcement refusal (-32047): .kind
    AnnouncementNotFoundError,         # Unknown subscription, delivery, source or mob (-32048): .kind
)

import base64

# Create a draft from the blank template, validate, export
draft = await handle.mobpack_create(template="blank", name="support-bot")
document = draft.row.document

validation = await handle.mobpack_validate(document)
print(validation.ok, len(validation.display_rows))

pack = await handle.mobpack_export(document)
archive = base64.b64decode(pack.content_base64)  # pack.filename, pack.media_type

from meerkat_meta import (
    # Builder + Runtime
    Meta, MetaBuilder, MetaRuntime,
    # Data models
    DiscoverySpec, PreSpawnData, SessionBuildOptions, SessionQuery,
    # Agent builder
    SessionAgentBuilder,
    # Errors
    MetaError, TransportError, RpcError,
    MobEventsStaleError, MOB_EVENTS_STALE_CURSOR_CODE,
    StorageResolutionError, STORAGE_RESOLUTION_CODE,
    CompositionProvenanceError, COMPOSITION_PROVENANCE_CODE,
    NotConnectedError, CapabilityUnavailableError, ContractMismatchError,
    # Typed results
    StatusResult, CapabilitiesResult, ReconcileResult,
    StorageSummary, StorageSlotSummary,
    SpawnResult, SpawnMemberResult,
    SendMessageResult, SubscribeResult,
    KeepAliveConfig, EventEnvelope,
    RoutingResolution, DeliveryResult,
    # Roster
    MemberSnapshot,
    MEMBER_STATE_ACTIVE, MEMBER_STATE_RETIRING, MEMBER_STATE_BROKEN,
    MEMBER_STATE_COMPLETED, MEMBER_STATE_UNKNOWN,
    # Routing & delivery
    RuntimeRouteResult, DeliveryHistoryResult,
    # Gating
    GatingEvaluateResult, GatingDecisionResult,
    GatingAuditEntry, GatingPendingEntry, GatingOrigin,
    # Memory
    MemoryQueryResult, MemoryStoreInfo, MemoryIndexResult,
    AgentMemoryRecord, AgentMemoryRecallResult, AgentMemoryForgetResult,
    # Events
    MobEvent, AgentEvent, EventStream,
    # Structural mob events + flow runs
    MobStructuralEvent, MobRun, MobRunStatus,
    StepRecord, FailureRecord, FrameRecord, LoopRecord, LoopIterationRecord,
    # Mob announcements
    Announcement, AnnounceResult, Announcer, AnnouncementRef, AnnouncementMember,
    AnnouncementSubscription, AnnouncementSubscriptionChange,
    AnnouncementRoleSelector, AnnouncementRoleTemplate,
    AnnouncementDeclarationRecord, AnnouncementDeclarationsView,
    AnnouncementSubscriptionView, AnnouncementSuspendMarker,
    AnnouncementBatch, AnnouncementBatchItem, AnnouncementBatchHandling,
    AnnouncementDeliveryObservation,
    AnnouncementSource, AnnouncementSourceIdentity, AnnouncementSourceRoute,
    AnnouncementSourceCursor, AnnouncementSourceRunner, AnnouncementSourceHealth,
    AnnouncementSourcePutResult, AnnouncementSourceStream, AnnouncementSourceView,
    SourceAdvanceResult, SourceItemOutcome, SourceStreamPhase,
    AnnouncementAuditEntry, MobAnnouncementsCapability, MobAnnouncementsStatus,
    AnnouncementMemberView, AnnouncementMemberBacklog, AnnouncementMemberHandling,
    AnnouncementEscalation, AnnouncementEscalationEvent, AnnouncementEscalationsPage,
    supports_announcement_sources, MOB_ANNOUNCEMENT_SOURCES_V1,
    MobAnnouncementsUnavailableError, AnnouncementConflictError,
    SourceStaleRevisionError, SourceVersionConflictError,
    AnnouncementFrontierError, AnnouncementRefusedError, AnnouncementNotFoundError,
    MOB_ANNOUNCEMENTS_UNAVAILABLE_CODE, ANNOUNCEMENT_CONFLICT_CODE,
    ANNOUNCEMENT_FRONTIER_CODE, ANNOUNCEMENT_REFUSED_CODE, ANNOUNCEMENT_NOT_FOUND_CODE,
    # Config modules
    auth, event_log, memory, runtime_store,
)