Plugin SDK — graph.sdk

The consumption SDK (ADR 0043): a stable, importable surface onto agent capability, so a plugin can run a subagent, search knowledge, schedule work, or record a metric without reaching into core internals. The distinction matters — internals are free to change underneath you between releases; this module and registry.host are the two surfaces that aren't.

from graph import sdk

answer = await sdk.complete('Summarize the changelog')

Import it lazily (inside the function that uses it) if your plugin must also run host-free in its own test suite. Generated from graph/sdk.py; the sections below are the module's own.

Contents

Plugin services (the plugin↔plugin call seam, ADR 0116) — service()

Agent + model access (the plugin↔agent channel, ADR 0043) — complete(), config(), gateway_client(), run_subagent(), subagent_types(), turn_stop_reason()

Model in-flight priority (the plugin↔limiter channel, ADR 0115 D6) — llm_lanes(), llm_priority()

Knowledge graph (the plugin↔knowledge channel, ADR 0043 — "shared knowledge") — knowledge_add(), knowledge_purge(), knowledge_search()

Goal-driven recurring loop (the OODA pattern) — clear_watch(), create_watch(), list_watches(), run_in_session(), start_goal_loop(), stop_goal_loop(), update_watch()

Background jobs (ADR 0050 spawn + ADR 0070 results pipeline) — background_status(), spawn_background()

Reactive rules (ADR 0039 events → one-shot turns) — react_on()

Plugin-owned recurring jobs (#1642) — cancel_plugin_jobs(), cancel_scheduled(), plugin_job_prefix(), schedule_recurring()

Plugin state (own the file, never core's) — plugin_store()

Plugin metric timeseries (#1632) — metric_history(), metric_last(), record_metric()

Run tracing (one Langfuse trace per multi-step plugin run) — trace_attributes(), trace_run()

Delegation ledger (who handed what work to whom) — record_delegation()

Managed Python runtime (ADR 0094 — the provisioned child interpreter) — managed_python_exe()

Plugin services (the plugin↔plugin call seam, ADR 0116)

sdk.service

sdk.service(name: str) -> Callable | None

The callable another plugin offers as name ("<plugin_id>.<name>"), or None.

The way one plugin CALLS another without importing it: the provider registers it with registry.register_service and you resolve it here, at call time — never at register() time, since plugins load in no guaranteed order and a reload can swap the provider in or out. None means no loaded plugin provides it (disabled, not installed, or a core older than the provider needs): degrade with an actionable message rather than fail. Guard the lookup itself with getattr(sdk, "service", None) on cores older than this seam. The callable's signature and return shape are the PROVIDER's documented API — read its docstring (help(sdk.service("artifact.show"))). The reference provider is the artifact plugin's artifact.show::

show = sdk.service("artifact.show")
if show is None:
    return "The Artifact plugin is off — enable it to see charts."
r = show(kind="vega-lite", code=spec_json, title="Sales by weekday")
return r["message"] + r["ref"]   # the artifact-ref chip tail goes LAST

Agent + model access (the plugin↔agent channel, ADR 0043)

sdk.config

sdk.config() -> Any

The live runtime LangGraphConfig.

sdk.gateway_client

sdk.gateway_client(*, timeout: float | None = None) -> Any

An httpx.AsyncClient pre-configured for the model gateway (#1931): base_url = the configured api_base, bearer auth, the allowlisted User-Agent (the gateway's WAF 403s default SDK UAs), and a sane timeout.

For OpenAI-compatible endpoints the chat model doesn't cover — /images/generations, /images/edits, /audio/* (core's own transcription rides the same client). Request relative paths and use it per call::

async with sdk.gateway_client(timeout=300) as client:
    resp = await client.post("/images/generations", json={...})
    resp.raise_for_status()

Call gateway endpoints through this — never a provider backend directly: the api_base host is auto-trusted by the egress guard + OpenShell network policy (ADR 0008); other hosts are not (a private backend IP is denied outright). timeout=None keeps the factory default.

sdk.subagent_types

sdk.subagent_types() -> set[str]

Ids of the configured subagents — for validating/listing recipe steps.

sdk.run_subagent

await sdk.run_subagent(subagent_type: str, prompt: str, *, description: str, extra_tools: Any = None, truncate: int | None = None, turn_model: str | None = None) -> str

Run a subagent to completion and return its text output.

Pulls the config + knowledge store + scheduler from runtime state, so a plugin tool only supplies the subagent + prompt. This is the capability the workflows plugin's engine injects as its per-step run_step.

extra_tools defaults to the host's plugin + MCP tools (the same set the lead graph and the console fan-out expose) — a subagent whose allowlist names a plugin tool (the review-finder's github_pr_diff, a finance backtester) must see it here too, or every SDK-driven workflow step silently degrades to "No tools available". Pass an explicit list (even []) to override.

turn_model is the calling turn's model override; None (the default) reads it from the in-flight turn (graph.subagent_model.current_turn_model, bound by the chat drivers), so a workflow step follows the model the operator picked for the turn — under the one precedence: the subagent's pin > the override > routing.aux_model > the main model (#3944, #3955). Outside a turn there is none and the step resolves pin > aux > main, as before.

sdk.turn_stop_reason

sdk.turn_stop_reason() -> str

Why the turn this code runs under was stopped from above, or "".

Read it on a CancelledError to tell the two apart (#3940): "" is a plain cancel (an operator's A2A CancelTask, a shutdown); otherwise the stall guard ended the turn because it stopped making progress, and this is its message — a failure, not a cancel. See graph.turn_liveness.

sdk.complete

await sdk.complete(prompt: str, *, system: str | None = None, model_name: str | None = None) -> str

Run a single bare LLM completion and return the text — no tools, no agent loop, no persona, no memory. The clean primitive for a plugin that just needs the model to answer a prompt (e.g. an interactive artifact calling back to the agent, a one-shot classifier/summarizer). Distinct from run_subagent, which runs a full tool-using subagent. Uses the live config's model through the gateway; pass model_name to target a different model on the same gateway, system for a system instruction.

Model in-flight priority (the plugin↔limiter channel, ADR 0115 D6)

sdk.llm_priority

sdk.llm_priority(cls: str) -> _LlmPriorityScope

Run a block of model calls under the ADR 0115 D6 priority class cls on the gateway in-flight limiter, restoring the previous class on exit.

When a lane is saturated the limiter serves waiters by class — interactive (a chat turn an operator is watching) first, then default (A2A / background / scheduled turns), then bulk (wide fan-outs, sweeps, review-panel finders) — with a slot reserved for interactive callers. A plugin running a burst of non-urgent model work marks it bulk so it queues behind an operator's chat instead of racing it::

from graph import sdk

async with sdk.llm_priority("bulk"):
    await asyncio.gather(*(run_finder(f) for f in finders))

Usable as either with or async with — pick whichever the call site is. The class is carried in a contextvars.ContextVar, so any subagent / asyncio task spawned inside the block inherits it, and nested scopes restore the enclosing class on exit. When the limiter is off (model.max_inflight: 0, the default) the class is still set but costs nothing — acquisition is a pass-through.

Args:

• cls — one of "interactive" / "default" / "bulk".

Raises:

• ValueError — if cls is not one of the three classes — the message names the valid ones.

sdk.llm_lanes

sdk.llm_lanes() -> dict

The ADR 0115 D8 gateway in-flight-limiter snapshot for this process, read straight out of memory (no HTTP, no DB).

The same payload the operator console reads over GET /api/telemetry/llm-lanes, but a plugin running inside the agent process (pr-reviewer's review panels, say) reads it here without a round-trip. Use it to tell a lane that is deliberately queuing from a gateway that has gone slow: high queued / wait_p90_s_5m or saturated: true with slots full is local back-pressure this process is applying on purpose; retries or timeouts while slots are NOT saturated point at the gateway itself.

Returns:

{"enabled": bool, "generated_at": iso, "lanes": [...]}. enabled is false (and lanes empty) when the limiter is off (model.max_inflight: 0, the default). Each lane entry carries lane, limit, reserve, inflight, queued, queued_by_priority (interactive/default/bulk), oldest_wait_s, wait_p50_s_5m / wait_p90_s_5m, queue_timeouts_5m and saturated (queued > 0 held continuously for 60 s or more).

Knowledge graph (the plugin↔knowledge channel, ADR 0043 — "shared knowledge")

sdk.knowledge_search

await sdk.knowledge_search(query: str, *, k: int = 5, domain: str | None = None, epoch: str | None = None) -> list[dict]

Search the agent's knowledge graph (hybrid FTS5 + embeddings); return the top-k matching chunks (each a dict with preview/content, domain, score …), or [] when no store is configured. domain scopes to one bucket (e.g. "loop-lessons"); epoch (#1634) scopes to chunks tagged with exactly that era via knowledge_add(..., epoch=...) — out-of-era and untagged chunks don't match (both search modes filter). None = unfiltered.

sdk.knowledge_add

await sdk.knowledge_add(content: str, *, domain: str = 'general', heading: str | None = None, epoch: str | None = None, memory_kind: str | None = None, delivery_policy: str | None = None, review_state: str | None = None, expires_at: str | None = None) -> int | None

Add one chunk to the agent's knowledge graph; return its id, or None when no store is configured / it was a no-op. domain is the bucket, heading an optional title — e.g. knowledge_add(lesson, domain="loop-lessons", heading=cls). epoch (#1634) tags the chunk with the era it was learned in — an opaque string, typically a reset date (epoch="2026-06-29"). On the next wipe the plugin just searches with the NEW epoch: old lessons stay for post-mortems but stop matching. memory_kind / delivery_policy (ADR 0108 D4) are the typed-memory columns — what the chunk IS ("fact", "note", "reference", …) and WHEN it enters the prompt ("always" / "retrieved" / "on_demand"). review_state / expires_at (ADR 0108 D7) are the write lifecycle: whether an operator has confirmed the row ("confirmed" / "pending" / "rejected") and an optional ISO-8601 UTC timestamp after which it lapses. Omitted = the store's own stamping: kind inferred from the domain, retrieved, "pending" (a plugin write is not operator intent), no expiry.

sdk.knowledge_purge

await sdk.knowledge_purge(domain: str, *, before: str | None = None) -> int

HARD-delete every chunk in domain — optionally only those created strictly before before (an ISO-8601 timestamp) — and return how many were removed.

The knowledge-lifecycle primitive (#1634): retire a bucket of now-wrong lessons (knowledge_purge("st-routes")) or expire just the stale tail (knowledge_purge("st-routes", before="2026-06-01")). Deletes consistently from every index (main rows, FTS, vectors); on a layered store only the PRIVATE tier is purged — the shared commons is curated, never bulk-deleted. Keep-for-audit retirement is the epoch tag instead (see knowledge_add). Returns 0 when no store is configured (or the backend has no purge_domain), when domain is empty, or when before is unparseable — it refuses rather than risk deleting the wrong rows.

Goal-driven recurring loop (the OODA pattern)

sdk.run_in_session

sdk.run_in_session(session_id: str, prompt: str, *, delay_seconds: float = 0.0, job_id: str | None = None) -> dict

Enqueue prompt as a one-shot agent turn in session_id — non-blocking.

This is the primitive behind "when a goal fires, prompt the agent." Call it from a goal on_achieved / on_failed hook (registry.register_goal_hook(...)) — or any plugin event handler — with a prompt built from the terminal GoalState (its condition / last_reason / last_evidence), and the agent runs a follow-up turn (with that session's memory and full tool set) reacting to what just happened::

async def on_achieved(goal):
    sdk.run_in_session(
        goal.session_id,
        f"The goal '{goal.condition}' just completed. Evidence: {goal.last_evidence}. "
        f"Write up a summary and open the follow-up PR.",
    )
registry.register_goal_hook(on_achieved=on_achieved)

Mechanics: it schedules a one-shot job (an ISO fire time, not a cron) into the session's context via the scheduler, so the turn runs on the normal fire path (the same loopback A2A call cron ticks use) and the caller returns immediately. It NEVER runs the turn inline, so it is safe to call from a goal hook / monitor tick without blocking it.

Args:

• session_id — the A2A contextId to run the turn in (e.g. goal.session_id).
• prompt — the message the agent processes as a turn.
• delay_seconds — fire at now + this delay (default 0 → the next poll tick, ~1s).
• job_id — a stable id so a re-call REPLACES the pending one-shot (idempotent).

Returns {"ok", "job_id", "fires_at", "message"}; ok=False with a readable message when the scheduler is unavailable or the inputs are bad.

sdk.create_watch

sdk.create_watch(*, condition: str, verifier: str, verifier_args: dict | None = None, watch_id: str | None = None, interval_s: float | None = None, deadline: float | None = None, stall_after: int | None = None, run_prompt: str = '', run_session: str = '', trigger: str = 'met', repeat: bool = False) -> dict

Register a WATCH from a plugin (ADR 0067): poll condition — ground-truthed by the plugin verifier named verifier ("<plugin-id>:<name>") — on a cadence, and when it trips run run_prompt as a follow-up turn in run_session (via run_in_session) and fire on_met hooks. Plugin-verifier only (like a set_goal-tool goal); hold as MANY as trigger + repeat decide WHEN it fires and whether firing ends it: met/False is the default one-shot tripwire; met/True fires on every rising EDGE of the predicate (never per-tick while it stays true); change/True is a standing monitor — it fires whenever the verifier's evidence MOVES, whatever the predicate says, and calls on_changed rather than on_met. A repeating watch only ends on its deadline or an explicit clear, so give it one if it shouldn't run forever.

you like (unlike a monitor goal, which is one-per-session). Returns {"ok", "watch_id", "message"} — ok=False with a readable message if the subsystem is off or the verifier is rejected.

sdk.list_watches

sdk.list_watches(prefix: str = '') -> list[dict]

List the registered watches — each {"id", "condition", "status", "verifier"} — optionally filtered to ids starting with prefix. This is the read half create_watch was missing (#1638): a plugin that arms a watch suite under stable ids (st-credits, st-contract …) lists its own with list_watches("st-") to verify the suite or render it on a dashboard, and — paired with clear_watch — reconciles on upgrade: clear the ids no longer in its spec set, then create/replace the rest (stable-id replace alone only heals specs that still exist; a renamed/dropped spec would keep polling forever). Returns [] when the watch system is unavailable.

sdk.clear_watch

sdk.clear_watch(watch_id: str) -> bool

Remove the watch watch_id (it stops being polled; its state is deleted). Returns True if it existed, False when it didn't — or when the watch system is unavailable. The remove half of the list_watches reconcile pattern.

sdk.update_watch

await sdk.update_watch(watch_id: str, **fields) -> dict

Edit a live watch in place — {"ok", "watch_id", "message"}.

Accepts any of condition / interval_s / deadline / stall_after / run_prompt / run_session; an omitted field is untouched, and passing None CLEARS one (deadline=None drops the expiry). Async, unlike its sibling create_watch, because the edit takes the controller's per-watch lock so it can't interleave with a tick that is mid-evaluation on the same watch.

A plugin may only edit a watch whose verifier is plugin — the same trust boundary create_watch enforces (ADR 0067 D4) — and cannot change the verifier at all. Terminal watches are immutable; create a new one.

This completes the reconcile loop that list_watches / clear_watch began (#1638): a plugin whose spec changed a threshold no longer has to clear and recreate, which reset the watch's accumulated stall state every time it shipped a tweak.

sdk.start_goal_loop

sdk.start_goal_loop(*, goal: str, verifier: str, every: str, prompt: str, plugin_id: str, loop_id: str, session_id: str = '', verifier_args: dict | None = None, done_prompt: str = '', timezone: str | None = None, interval_s: float | None = None, deadline: float | None = None, stall_after: int | None = None) -> dict

Wire a goal-driven recurring loop in ONE call (the OODA / self-improving pattern): arm a WATCH on goal — ground-truthed by the plugin verifier verifier — and schedule the recurring prompt tick that drives it. Sugar over create_watch + schedule_recurring under a shared derived id (#2060); goals themselves stay drive-only (ADR 0067) — the "wait for the metric" half IS the watch.

The tick keeps firing until torn down: call stop_goal_loop from a watch on_met hook for a self-retiring cadence (or from a stop tool). Register the verifier — and that hook — in register() first; pass session_id from your tool's InjectedState so the tick (and the done_prompt reaction) run in the caller's session::

from graph.sdk import start_goal_loop, stop_goal_loop

start_goal_loop(goal="reach 1M credits", verifier="fleet:credits",
                verifier_args={"min": 1_000_000}, every="30m",
                prompt="Run the OODA tick and report.",
                plugin_id=registry.plugin_id, loop_id="credits-1m",
                session_id=sid)

PREFIX = f"{PLUGIN_ID}:goal-loop:"
async def on_met(watch):                       # self-retiring cadence
    if watch.id.startswith(PREFIX):
        stop_goal_loop(plugin_id=PLUGIN_ID, loop_id=watch.id.removeprefix(PREFIX))
registry.register_watch_hook(on_met=on_met)

Idempotent by (plugin_id, loop_id): a re-call REPLACES both the pending tick and the watch, so a plugin re-arms in register() without colliding. If the tick can't be scheduled the watch is rolled back — never half a loop.

Args:

• goal — the condition the watch verifies (e.g. "reach 1,000,000 credits").
• verifier — the registered plugin verifier name, "<plugin-id>:<name>".
• every — the tick cadence — a 5-field cron ("0 */6 * * *") or a duration shorthand "15m" / "2h" / "1d".
• prompt — the recurring tick prompt (e.g. "Run the manage-the-fleet OODA tick …").
• plugin_id — the owning plugin's id (registry.plugin_id) — the tick job rides the plugin namespace, so disable/uninstall sweeps it. ':' is rejected.
• loop_id — a stable plugin-local id for this loop (e.g. "credits-1m").
• session_id — the session the tick fires into; empty → the durable Activity thread.
• verifier_args — declarative args for the verifier (e.g. {"min": 1000000}).
• done_prompt — optional follow-up turn to run (in session_id) when the watch trips — the wrap-up/celebration half. Requires a non-empty session_id.
• timezone — IANA tz the cron is evaluated in (None = UTC). interval_s, deadline, stall_after: watch knobs, passed through to create_watch (poll floor / epoch-seconds deadline / stall hook).

Returns {"ok", "goal", "loop_id", "watch_id", "job_id", "schedule", "message"}; ok=False with a readable message when a subsystem is off or the inputs are bad.

sdk.stop_goal_loop

sdk.stop_goal_loop(*, plugin_id: str, loop_id: str) -> dict

Tear down the goal loop (plugin_id, loop_id): cancel its recurring tick and clear its watch — both re-derived, so the caller only holds the two ids it started the loop with. Idempotent: stopping an absent loop is ok=True with both flags False. Call it from a watch on_met hook (the self-retiring pattern in start_goal_loop's example), a stop tool, or when winding down.

Background jobs (ADR 0050 spawn + ADR 0070 results pipeline)

sdk.spawn_background

await sdk.spawn_background(prompt: str, *, subagent_type: str, origin_session: str, label: str | None = None) -> dict

Spawn a detached background subagent job (ADR 0050) and return immediately.

The job runs as its own detached A2A turn under the subagent_type role; when it finishes, the ADR 0070 pipeline delivers the report — a push-resume nudge into origin_session, the notified-gated <task-notification> drain, knowledge-store indexing, and the console report card. Poll in between with background_status (e.g. to render campaign progress on a plugin dashboard).

Args:

• prompt — detailed instructions for the background worker.
• subagent_type — which subagent role runs the job — one of the registered roster (subagent_types), plugin-contributed subagents included.
• origin_session — the chat session the report drains back into (and gets the completion nudge). Required — a job with no origin has nowhere to report.
• label — short human description for the job card / report heading. Defaults to the first line of prompt (clipped).

Returns {"ok", "task_id", "message"} — task_id is the bg-… job id (the handle for background_status, cancel, and the by-id API route); ok=False with a readable message when the background subsystem is off or the inputs are bad.

sdk.background_status

sdk.background_status(task_id: str) -> dict

Look up a background job by its bg-… id — the status-query companion to spawn_background, so a plugin can render campaign progress on its own surface instead of being blind between launch and the ADR 0070 completion nudge.

Returns {"ok", "task_id", "status", "subagent_type", "description", "created_at", "completed_at", "message"} plus — only once the job is terminal (completed/failed/canceled) — "report" with the full result text. An unknown id (or the subsystem being off) returns ok=False with status="unknown" and a readable message. Reads the durable jobs store directly (cheap local SQLite).

Reactive rules (ADR 0039 events → one-shot turns)

sdk.react_on

sdk.react_on(topic: str, *, prompt: Callable[[dict], str | None], job_id: str, session: str = 'system:activity', debounce_s: float = 0.0) -> Callable[[], None]

When a bus event matching topic fires, enqueue a follow-up agent turn.

prompt is called with the full event payload ({"event", "data", "seq"}) at delivery time; return the turn's prompt text, or None/empty to skip that event (cheap filtering). The turn is enqueued via run_in_session with job_id, so a rule re-fires idempotently (a pending turn is REPLACED, never duplicated)::

unsub = sdk.react_on(
    "spacetraders.opportunity",
    prompt=lambda ev: f"A {ev['data']['margin']}% route appeared. Evaluate it.",
    job_id="spacetraders-opportunity",
    debounce_s=30,
)

Args:

• topic — bus topic pattern (* = one segment, # = tail — any namespace; subscribing is read-only, like registry.on).
• prompt — (payload) -> str | None — the prompt builder / filter.
• job_id — stable id for the enqueued turn (run_in_session's idempotent-replace key). Required — it's what keeps a chatty rule from stacking turns.
• session — the session the turn runs in; defaults to the durable Activity thread (ACTIVITY_CONTEXT).
• debounce_s — > 0 coalesces a burst into ONE turn — trailing-edge: the timer re-arms on every qualifying event, fires debounce_s after the LAST one, and that last event's prompt text wins. Skipped events (prompt returned None/empty) neither fire nor extend the window. Note a sustained stream arriving faster than debounce_s keeps deferring the turn (classic debounce). Thread-safe — the bus may deliver from worker threads.

Returns an unsubscribe callable (mirroring registry.on's seam): it stops delivery and cancels any pending debounce timer. When no host bus is wired (tests, headless), logs a warning and returns a no-op unsubscribe.

Plugin-owned recurring jobs (#1642)

sdk.plugin_job_prefix

sdk.plugin_job_prefix(plugin_id: str) -> str

The id prefix every scheduler job owned by plugin_id carries.

sdk.schedule_recurring

sdk.schedule_recurring(prompt: str, cron: str, *, plugin_id: str, job_id: str, session: str = '', timezone: str | None = None) -> dict

Schedule prompt as a RECURRING agent turn on a cron cadence, owned by plugin_id.

Thin over STATE.scheduler.add_job with the id namespaced plugin:<plugin_id>:<job_id> — that ownership tag is what lets the host cancel the plugin's jobs on disable/uninstall (and cancel_scheduled / cancel_plugin_jobs find them). Idempotent by id: re-calling with the same job_id REPLACES the pending job, so a plugin re-arms its cadence in register() (or when a cadence knob changes) without colliding.

There is no ambient plugin identity in the SDK (functions are plain imports, not registry-bound), so plugin_id is an explicit required kwarg — pass registry.plugin_id. Note: a disable cancels the plugin's jobs; re-enabling relies on the plugin re-arming in register(), which runs after the scheduler is wired on both boot and hot-reload.

Args:

• prompt — the message the agent processes each fire.
• cron — a 5-field cron expression (e.g. "0 9 * * *"). One-shot turns belong to run_in_session, so an ISO datetime is rejected here.
• plugin_id — the owning plugin's id (registry.plugin_id).
• job_id — a stable plugin-local id for this cadence (e.g. "strategist-tick").
• session — the A2A contextId to fire into; empty → the durable Activity thread (the default for scheduled work).
• timezone — IANA name the cron is evaluated in (None = UTC).

Returns {"ok", "job_id", "next_fire", "message"} — job_id is the full namespaced id; ok=False with a readable message when the scheduler is unavailable or the inputs are bad.

sdk.cancel_scheduled

sdk.cancel_scheduled(job_id: str, *, plugin_id: str) -> bool

Cancel the plugin-owned recurring job job_id (the same plugin-local id passed to schedule_recurring — namespacing is applied here, never by the caller). Returns True if a job was removed; False when there was none — or when the scheduler is unavailable.

sdk.cancel_plugin_jobs

sdk.cancel_plugin_jobs(plugin_id: str) -> int

Cancel EVERY scheduler job owned by plugin_id (ids plugin:<plugin_id>:*). Returns how many were cancelled (0 when the scheduler is unavailable).

This is the lifecycle hygiene hook (#1642): the loader sweeps a disabled plugin's jobs on (re)load and the installer sweeps on uninstall, so an orphan cadence can't keep firing prompts about a plugin that's gone. Only jobs under this instance's agent_name are visible (list_jobs filters), so the ADR 0004 scoping holds. Also useful to a plugin that wants to tear down its whole cadence at once.

Plugin state (own the file, never core's)

sdk.plugin_store

sdk.plugin_store(subdir: str = '', *, plugin_id: str) -> Path

This plugin's own writable directory, scoped to the running instance.

Use it for state the plugin owns — a SQLite file, a cache, exports, generated assets. The directory is created if absent, and it lives under the instance root (ADR 0004 / ADR 0065), so the dev sandbox and every fleet member get their own copy without the plugin doing anything: the isolation that used to be each plugin's job to remember.

Do not open core's databases (checkpoints.db, knowledge.db, the telemetry or tasks stores). They carry no compatibility promise for outside readers and core migrates them freely; reach core data through this SDK instead (knowledge_search, metric_history, …).

Prefer an existing seam where one fits — small numeric series belong in record_metric, retrievable facts in knowledge_add, and an index you can rebuild at load belongs in memory. Reach for a file when you have durable, structured, plugin-shaped state.

Args:

• subdir — optional path under the plugin's directory ("exports", "cache/thumbs"). Must be relative and must not escape upward; a traversing or absolute value is rejected.
• plugin_id — the owning plugin's id (registry.plugin_id) — explicit, like record_metric, since the SDK has no ambient plugin identity.

Returns the created directory as a Path.

Raises:

• ValueError — on an empty/unsafe plugin_id or an escaping subdir — a bad path is raised rather than silently redirected, because writing an operator's data somewhere unexpected is worse than failing loudly.

from graph import sdk

db = sdk.plugin_store(plugin_id=registry.plugin_id) / "state.db"

Plugin metric timeseries (#1632)

sdk.record_metric

sdk.record_metric(name: str, value: float, *, ts: float | None = None, plugin_id: str) -> dict

Append one sample to the plugin metric timeseries name (#1632).

Small named numeric series — treasury, net worth, fleet size — are what history-dependent watch verifiers (ADR 0067 drawdown-vs-high-water, flatline detection) and dashboard sparklines need, and a verifier can only see live state (sdk.telemetry() is point-in-time). Series are namespaced <plugin_id>:<name>, SQLite-backed in the instance dir (observability/metrics_store.py), and retention-capped per series (90 days / 10k points, trimmed on write) — record freely from an engine tick.

Args:

• name — the plugin-local series name (e.g. "credits"). Namespacing is applied here, never by the caller.
• value — the sample — any real number (NaN/inf are rejected; they poison drawdown math downstream).
• ts — Unix epoch seconds for the sample; None → now. Useful for backfill and deterministic tests.
• plugin_id — the owning plugin's id (registry.plugin_id) — explicit, like schedule_recurring; ':' is rejected.

Returns {"ok", "series", "message"} — ok=False with a readable message when the store is unavailable (non-server context) or the inputs are bad; a write failure never raises into an engine loop.

sdk.metric_history

sdk.metric_history(name: str, *, since: float | None = None, limit: int = 500, plugin_id: str) -> list[tuple[float, float]]

The newest limit samples of the plugin metric series name — at/after since (Unix epoch seconds) when given — returned oldest→newest as (ts, value) tuples: chronological order, ready for verifier math (high_water = max(v for _, v in points)) or a sparkline. Returns [] when the store is unavailable, name/plugin_id are invalid, or the series has no samples. Same namespacing + plugin_id contract as record_metric.

sdk.metric_last

sdk.metric_last(name: str, *, plugin_id: str) -> tuple[float, float] | None

The most recent (ts, value) of the plugin metric series name, or None when there is no sample (never recorded / fully aged out / store unavailable). The cheap read for "what did I last see?" checks — e.g. a verifier comparing the live reading against the last recorded one. Same namespacing + plugin_id contract as record_metric.

Run tracing (one Langfuse trace per multi-step plugin run)

sdk.trace_attributes

sdk.trace_attributes(metadata: dict | None = None, *, tags: list[str] | None = None)

Stamp metadata on every observation traced inside the block, tags on its trace.

Every subagent:<type> span a run_subagent call opens in the block — however deep — and every tool call and generation under it carries metadata in its own metadata. The keys stay on the OBSERVATIONS, not the trace: a trace holds a whole run, and trace metadata has one value per key, so a per-step key there would read as whichever step started last. tags are a set and do go on the trace. Scopes nest and merge; the innermost key wins. Use it around one step of a multi-step run, inside trace_run::

with sdk.trace_attributes({"step_id": sid}, tags=[f"step:{sid}"]):
    out = await sdk.run_subagent(subagent, prompt, description=...)

Pass ids and labels — never raw content: nothing here is redacted for you. A no-op (beyond the contextvar) when tracing is disabled; the block always runs.

sdk.trace_run

sdk.trace_run(name: str, *, run_id: str = '', input: Any = None, metadata: dict | None = None, inputs: dict | None = None, tags: list[str] | None = None)

Group a plugin's multi-step run into ONE Langfuse observation.

Every run_subagent call opens a subagent:<type> span in the CURRENT trace. One trace per execution SEGMENT: a run that parks (a human gate) and later resumes is two traces, grouped in Langfuse's Sessions view by the shared run_id. The OTel context isn't persisted across a pause of unbounded length.

Called from a turn (the run_workflow tool), that nests; called with no turn around it (a REST route, a detached Studio run, a scheduled fire), each step became its own sessionless ROOT trace — a 9-step review panel was 9 unrelated traces, and a busy reviewer buried every agent's turns under them.

Inside a traced turn this opens a nested span, in the turn's trace and session; otherwise it opens the run's own root trace, whose session is the caller's session when there is one and run_id when there isn't. Input/output are redacted and capped, and dropped entirely in an incognito turn. A block that raises marks the observation ERROR and re-raises unchanged. A no-op when tracing is disabled — the block always runs::

async with sdk.trace_run(f"workflow:{name}", run_id=run_id, input=inputs) as run:
    result = await execute(...)
    run.output(result["output"], failed=bool(result["failed"]))

Join keys (#3565). Every observation in the run — each step's subagent: span included — carries run_id, metadata, the caller's session as parent_session_id, and the run's scalar inputs (default: input when it is a dict) as redacted, capped input_<name> keys; the trace is tagged with name, run:<run_id> and tags. Incognito keeps the ids and drops the inputs. Add per-step keys with trace_attributes.

Delegation ledger (who handed what work to whom)

sdk.record_delegation

sdk.record_delegation(*, to_kind: str, to_name: str, to_instance: str = '', what: str = '', session_id: str = '', parent_task_id: str = '', task_id: str = '', outcome: str = 'ok', error: str = '', duration_ms: int = 0, cost_usd: float | None = None, origin: str = '') -> int | None

Record one delegation edge in the instance ledger — who handed what work to whom.

The seam a plugin uses to put its own dispatches on the org's record. A plugin that hands work to an agent or a coder outside the built-in funnels (a bespoke scheduler, a board loop of its own) is otherwise invisible: turn telemetry has no actor column and no edge, so its work would show up as spend with no explanation of who asked for it.

cost_usd=None means UNKNOWN, not free — pass a number only where one is actually measured. A confident zero makes an unmeasured delegate look free and silently understates every rollup built on the column.

Best-effort and never raises: a ledger failure must not break a dispatch.

Returns the new edge_id, or None when no store is wired or the write failed.

Managed Python runtime (ADR 0094 — the provisioned child interpreter)

sdk.managed_python_exe

sdk.managed_python_exe() -> Path | None

The managed Python runtime's interpreter, or None when it isn't provisioned.

This is the interpreter execute_code spawns on the packaged desktop app, where sys.executable is the frozen server binary and cannot run a script. It is a pinned CPython the host downloads on demand into the box-shared data dir (runtime.python_install), shared by every instance on the machine.

Returns the interpreter's Path only when a working install is present — the executable actually exists — so a half-extracted, wiped or never-provisioned runtime reads as None rather than as a path that fails to spawn. On None, a source run can fall back to sys.executable (what execute_code itself does); a frozen one should tell the operator to provision the managed Python runtime rather than guess at a system Python of arbitrary version.

Cheap (one stat), never spawns, never raises.

import sys
from graph import sdk

exe = sdk.managed_python_exe()
if exe is None and not getattr(sys, "frozen", False):
    exe = Path(sys.executable)  # source run: this process's own interpreter