`agentkit.adapters`¶

Concrete Port implementations: LLM providers (Claude, OpenAI, DeepSeek, OpenRouter), vector / store / checkpoint back-ends, observer + replay tooling, and the OTel bridge.

Most adapters are behind an opt-in extra (http, postgres, redis, observability) so the zero-dep core stays clean.

agentkit.adapters itself re-exports nothing — each adapter lives in its own subpackage. Reference them directly:

LLMPort adapters — CallableLLM (inject any provider).

Model fallover is no longer an adapter — it's the fallback middleware. These adapters just wrap a single provider seam. CallableLLM adapts an injected fn/chat_fn (e.g. the OpenRouter ModelSDKTools). The raw↔LLMResult translation helpers live in _mapping. The public surface is agentkit.adapters.llm.

For the offline, deterministic FakeLLM/Turn test double, import from agentkit.testing — real adapters live in adapters/, test doubles in testing/.

CallableLLM ¶

CallableLLM(
    fn: Callable[..., Any],
    *,
    to_result: Callable[[Any], LLMResult] | None = None,
    chat_fn: Callable[..., Any] | None = None,
    cost_fn: Callable[[Any, int, int], float] | None = None,
)

Adapt fn(*, system, user, model, …) (single call) and optionally chat_fn(*, messages, model, tools, …) (tool-aware) into an LLMPort. Sync fns run off the loop via asyncio.to_thread; coroutines are awaited. Provider tool-calls are normalized to ToolCall.

StorePort adapters — ONE durable-KV seam backing checkpoints, idempotency, audit, and cache.

InMemoryStore is the offline reference + contract (single-flight get_or_set, never-negative on a raised producer, append-only logs). FileStore is the zero-dependency durable adapter (survives a restart). RedisStore/PostgresStore (behind extras) are the same shape over real infra. One KV seam serves checkpoints, idempotency, the audit log, and the result cache.

One backend per module (single responsibility); the public surface is agentkit.adapters.store.

FileStore ¶

FileStore(base_dir: str)

Durable StorePort backed by JSON files under base_dir — survives a process restart, so a human-gate suspend or a crashed run resumes from disk. Zero-dependency (stdlib json/pathlib) and async-first: the blocking file I/O is bridged via asyncio.to_thread so it never stalls the loop.

The reference durable adapter; a Postgres/Redis StorePort is the same shape behind an extra. Values must be JSON-serializable (which the loop/workflow checkpoints are). Single-flight is in-process (an asyncio.Lock per key); cross-process single-flight needs the real DB's transaction — documented. ttl is accepted but not enforced (no expiry sweeper); use a TTL-native backend for that.

InMemoryStore ¶

InMemoryStore(*, clock: Callable[[], float] = monotonic)

get_or_set `async` ¶

get_or_set(
    key: str,
    fn: Callable[[], Awaitable[Any]],
    *,
    ttl: int | None = None,
) -> Any

Single-flight: a hit returns the stored value; a miss runs fn once and stores it. A raised fn propagates and is NOT stored, so a transient error retries clean (failures are never cached). ttl is applied on the stored result, matching the other backends.

PostgresStore ¶

PostgresStore(dsn: str | None = None, *, pool: Any = None)

Durable StorePort over Postgres (extra: arc-agentkit[postgres], via asyncpg). One KV table + one append-log table, JSON-text values. Call await init() once to create the tables. Single-flight is in-process; durability/atomicity is Postgres's. TTL is not enforced (no sweeper — use RedisStore for TTL). Inject a pool for tests; else an asyncpg pool is created from dsn.

RedisStore ¶

RedisStore(
    url: str | None = None,
    *,
    client: Any = None,
    namespace: str = "agentkit",
)

Durable StorePort over Redis (extra: arc-agentkit[redis]). KV via GET/SET(EX), append-logs via RPUSH/LRANGE, values JSON-encoded. Single-flight is in-process (an asyncio.Lock per key); TTL is honored natively (SETEX). Inject a client (e.g. a fake) for tests; else built from url.

VectorPort adapters.

InMemoryVector ¶

InMemoryVector()

VectorPort — scope-isolated, cosine over TF vectors, metadata-filtered, scored (tests/lean).

CheckpointPort adapters — InMemoryCheckpointStore is the offline reference + contract every durable backend matches; PostgresCheckpointStore is the production durable backing (extra: arc-agentkit[postgres]). Public surface is agentkit.adapters.checkpoint.

InMemoryCheckpointStore ¶

InMemoryCheckpointStore()

A CheckpointPort over an in-process dict. One list of Checkpoints per run_id.

PostgresCheckpointStore ¶

PostgresCheckpointStore(pool: Pool)

A CheckpointPort over Postgres via asyncpg. Pool is application-owned (matches the explicit-lifecycle convention PostgresStore follows — see store/postgres.py). Call await ensure_schema() once at startup; the per-call methods then go through pool.acquire() like every other asyncpg adapter in this package.

ensure_schema `async` ¶

ensure_schema() -> None

Idempotent DDL — CREATE TABLE IF NOT EXISTS + CREATE INDEX IF NOT EXISTS. Safe to call multiple times. Apps call this once at startup before any save/load (mirrors PostgresStore.init()).

Concrete observer adapters for the observation channel.

CollectingObserver / QueueObserver — terminal sinks (list capture; bounded async-consumable queue).
PolicyObserver / RollupObserver — emission-cadence wrappers around an inner observer.
Hooks — lifecycle subscription (on(stage, handler)), sugar over the stream.

In-process transports. A cross-process transport (e.g. Redis pub/sub → SSE) is a future adapter on the same ObserverPort. The public surface is agentkit.adapters.observer.

PolicyObserver ¶

PolicyObserver(
    inner: Any,
    *,
    allow: frozenset[str] | set[str] | None = None,
)

The emission-cadence knob: forwards only selected kinds to an inner observer.

result/error/interrupt always pass (never silence a result or a human gate); other kinds pass only if allow is None (everything) or contains the kind. Compose it around any ObserverPort (e.g. a QueueObserver); the consumer still streams from the inner observer. Emit-only by design — an ObserverPort needs only emit.

everything `classmethod` ¶

everything(inner: Any) -> PolicyObserver

Forward every observation (the streaming/step cadence).

summaries `classmethod` ¶

summaries(inner: Any) -> PolicyObserver

Forward rolled-up summary/progress (+ always-forwarded), drop finer events.

result_only `classmethod` ¶

result_only(inner: Any) -> PolicyObserver

Send only on a complete result/error (+ interrupt) — the quiet-pipeline cadence.

RollupObserver ¶

RollupObserver(
    inner: Any,
    *,
    every: int = 8,
    summarize: Callable[
        [Sequence[Observation]], str | Awaitable[str]
    ]
    | None = None,
    kind: ObservationKind = "summary",
)

The 'rolled-up summary' cadence: buffer non-critical observations and emit one summary every every of them — flushing the buffer first whenever a critical kind (result/error/interrupt) passes through, and again on close().

summarize(buffer) -> str builds the roll-up text; it may be sync or async (async is the hook for an LLM judge or a Compactor-backed summariser). Critical observations are always forwarded immediately (after the flush), so a result is never delayed behind buffered progress.

close `async` ¶

close() -> None

Flush any buffered tail (and close the inner observer if it supports it).

Hooks ¶

Hooks(inner: Any | None = None)

Lifecycle subscription: a thin ObserverPort that dispatches each observation to handlers registered with on(stage, handler), where stage is the observation kind (e.g. run_start, summary, interrupt, error, result) or "*" for every observation.

Lifecycle is not a new mechanism — it's the observation stream, named. Handlers are sync or async and never break the run: an exception in a handler is swallowed. Optionally forwards to an inner observer, so Hooks chains in front of a QueueObserver.

CollectingObserver ¶

CollectingObserver()

Records every observation into .items (no backpressure). For tests and small in-proc consumers.

QueueObserver ¶

QueueObserver(maxsize: int = 256)

Non-blocking, bounded async observer with the never-drop-results rule.

emit never blocks and never raises into the run. Non-critical observations (progress/summary) are bounded to maxsize; when over, the oldest non-critical is dropped (coalesced). result/error are never dropped and never count against the bound. stream() async-iterates in insertion order until close().

OpenTelemetry bridge. Distinct from adapters.observer (in-process observer-pattern impls). Requires the observability extra.

otel_exporter_otlp_http ¶

otel_exporter_otlp_http(
    *, endpoint: str | None = None
) -> None

One-call setup of the OTel SDK with an OTLP-HTTP exporter.

Builds a TracerProvider with a BatchSpanProcessor wrapping an OTLP-HTTP exporter pointed at endpoint (or the OTEL_EXPORTER_OTLP_ENDPOINT env var). Call once at process startup BEFORE otel_tracer() is consumed — sets the global tracer provider.

The convenience version: operators who want a "give me OTel with sensible defaults" path call this. Operators with their own setup (custom processor, multiple exporters, etc.) skip this and configure opentelemetry.trace.set_tracer_provider themselves.

otel_meter ¶

otel_meter(
    meter_provider: Any = None,
    *,
    instrument_name: str = "agentkit",
) -> Any

Construct a MetricsPort backed by the OpenTelemetry SDK.

Production: pass nothing — pulls the global meter (configured via env vars OTEL_EXPORTER_OTLP_ENDPOINT / OTEL_EXPORTER_OTLP_METRICS_ENDPOINT). Tests: pass an explicit meter_provider so each test gets isolated state.

Raises ImportError when opentelemetry-sdk isn't installed (i.e., the [observability] extra wasn't installed); the error message names the extra so the operator knows what to do.

otel_metrics_exporter_otlp_http ¶

otel_metrics_exporter_otlp_http(
    *, endpoint: str | None = None, interval_ms: int = 60000
) -> None

One-call OTel metrics SDK setup with an OTLP-HTTP exporter.

Sister to otel_exporter_otlp_http for traces. Call once at startup BEFORE otel_meter() is consumed — sets the global meter provider with a PeriodicExportingMetricReader pushing to the OTLP-HTTP endpoint every interval_ms (default 60s, the OTel default).

The convenience version: operators who want a "give me OTel metrics with sensible defaults" path call this. Operators with their own setup (custom readers, multiple exporters, etc.) skip this and configure opentelemetry.metrics.set_meter_provider themselves.

otel_sampler ¶

otel_sampler(ratio: float = 1.0) -> Any

Convenience: build a SamplerPort from agentkit's pure-Python TraceIdRatioSampler. ratio=1.0 = always-on; ratio=0.1 = 10%.

The sampler is a kernel-level seam (no OTel dependency) so this factory just wraps the in-tree impl — it lives next to the OTel metrics/tracer factories so operators have ONE module to import from when wiring observability. For richer policies (parent-based, rule-based), construct your own SamplerPort impl and inject it directly into Services(sampler=...).

otel_tracer ¶

otel_tracer(tracer_provider: Any = None) -> OtelTracePort

Construct a TracePort backed by the OpenTelemetry SDK.

Production: pass nothing — pulls the global tracer (configured via env vars OTEL_SERVICE_NAME / OTEL_EXPORTER_OTLP_ENDPOINT / OTEL_EXPORTER_OTLP_HEADERS / ... per the standard OTel SDK convention).

Tests: pass an explicit tracer_provider (e.g. a TracerProvider wired to InMemorySpanExporter) so each test gets its own isolated provider. OTel forbids overriding the global set_tracer_provider once it's been set, so sharing a global across tests gives stale-provider failures.

Raises ImportError when opentelemetry-sdk isn't installed (i.e., the [observability] extra wasn't installed); the error message names the extra so the operator knows what to do.

ReplayStore adapters.

FileReplayStore ¶

FileReplayStore(root: Path | str)

ReplayStore that writes each record to <root>/<span_id>.json.

Constructed with an explicit root directory (created on first write if missing). Use FileReplayStore.from_env() to read the RIO_REPLAY_DIR env var.

Best-effort: a failed write logs a warning and returns None (the contract of ReplayStore.put says writes must not raise into the run). A failed read returns None (same as a genuine miss).

from_env `classmethod` ¶

from_env() -> FileReplayStore | None

Read RIO_REPLAY_DIR and construct; return None when the env var isn't set so callers can fall back to NoopReplayStore.

default `classmethod` ¶

default() -> FileReplayStore

Standard location: $XDG_DATA_HOME/rio/replays or ~/.rio/replays.

agentkit.adapters¶