plugin-sdk-api.md

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

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

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()

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

sdk.config {#sdk-config}

sdk.config() -> Any

The live runtime LangGraphConfig.

sdk.gateway_client {#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}

sdk.subagent_types() -> set[str]

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

sdk.run_subagent {#sdk-run-subagent}

await sdk.run_subagent(subagent_type: str, prompt: str, *, description: str, extra_tools: Any = None, truncate: int | 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.

sdk.complete {#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.

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

sdk.knowledge_search {#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 {#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 {#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}

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}

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}

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}

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 {#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}

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}

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 {#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}

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}

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}

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}

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}

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}

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}

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}

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}

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}

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.