dome-sdk 0.1.0

Dome Platform Python SDK — AI agent governance

Dome Python SDK

The official Python SDK for the Dome Platform — AI agent governance with local Cedar evaluation.

Install

pip install dome

Quick Start

import dome

client = dome.DomeClient(dome.DomeConfig(
    base_url="https://api.dome.example.com",
    token="dome_at_...",
))
client.start()

client.check(
    tool="database:query",
    on_allow=lambda result: print(f"Allowed: {result.reason}"),
    on_deny=lambda req, reason: print(f"Denied: {reason}"),
)

client.close()

How It Works

The SDK syncs Cedar rule bundles from the Dome Platform and evaluates authorization decisions locally using cedarpy. This means:

• Fast — no network round-trip per check, sub-millisecond evaluation
• Reliable — background sync keeps rules fresh
• Consistent — identical Cedar evaluation as the Go SDK

Framework Adapters

Framework adapters are separate packages that wrap the core SDK for specific AI frameworks:

Package	Install	Framework
dome-langchain	pip install dome-langchain	LangChain

LangChain Example

import dome
from dome_langchain import govern_tools

client = dome.DomeClient(dome.DomeConfig(base_url="https://...", token="dome_at_..."))
client.start()

# Wrap all tools at once — authorization is automatic
governed = govern_tools(client, [search_tool, db_tool])

Configuration

All configuration is explicit — passed via DomeConfig.

Argument	Required	Default	Description
base_url	yes	—	Dome API URL
token	yes	—	Agent token (dome_ key or base64 envelope)
rule_sync_interval	no	30s	Seconds between rule bundle syncs
audit_batch_size	no	100	Max on-device decision events buffered before a flush
audit_flush_interval	no	5s	Seconds between buffered-decision flushes
logger	no	None	Custom logger instance

Audit

The SDK both writes and reads audit. On every check(), the SDK reports the on-device policy decision to the control plane via the Audit IngestEvents RPC (asynchronous, batched, drop-not-block — a failed audit flush never blocks the request path). The agent supplies only the decision facts (action, tool, result, reason, rule id, latency); the server derives scope, actor, producer (dome-device), and identity from the authenticated caller — client-supplied identity is never trusted, so an agent can only write audit within its own workspace. These self-reported device.decision events are evidence of an agent's own decisions; gateway/sidecar topologies remain the source of independently-observed traffic.

Hosted Audit v1 also exposes read APIs for querying, fetching, and streaming the typed audit envelope.

from dome.audit import AuditPayloadFilter, AuditQuery
from dome.client import ControlPlaneClient
from dome.token import TokenManager

token_mgr = TokenManager("https://api.dome.example.com", "dome_at_...")
token_mgr.exchange()
control = ControlPlaneClient("https://api.dome.example.com", token_mgr)

page = control.query_audit_events(
    AuditQuery(
        event_types=("mcp.tool_call.completed",),
        results=("EVENT_RESULT_SUCCEEDED",),
        actor_kind="ACTOR_KIND_AGENT",
        producer_service="gateway",
        request_surface="INITIATOR_SURFACE_GATEWAY_MCP",
        operation_id="018fc629-4a74-7cc0-b41c-869f7ef7f6c4",
        start_time="2026-05-30T00:00:00Z",
        end_time="2026-05-31T00:00:00Z",
        payload_filters=(
            AuditPayloadFilter(
                event_type="mcp.tool_call.completed",
                field="toolName",
                values=("read_file",),
            ),
            AuditPayloadFilter(
                event_type="mcp.tool_call.completed",
                field="statusCode",
                operator="exists",
            ),
        ),
        page_size=50,
    )
)

for event in page.events:
    print(event.accepted_at, event.type, event.result, event.operation_id)
    print(event.scope.workspace_id, event.actor.kind, event.payload)

event = control.get_audit_event(page.events[0].id)

for event in control.stream_audit_events(
    AuditQuery(
        event_types=("access.denied",),
        page_token=page.next_page_token,
        payload_filters=(
            AuditPayloadFilter(field="requiredPermission", operator="exists"),
        ),
    )
):
    print(event.id)

AuditQuery uses the protobuf JSON enum names for enum filters. QueryEvents supports cursor pagination (page_token / next_page_token), time ranges, actor filters, producer/surface filters, operation linkage filters, resource filters, and registry-declared payload equality/exists filters. StreamEvents uses the same event/actor/producer/surface/operation/payload/cursor filters as the v1 stream RPC; query-only filters such as time ranges and primary-resource filters raise locally instead of being silently dropped.

Activity Correlation (Sessions)

Audit events normally chain only within one request (by trace_id). To chain the events from many Dome calls into one activity chain — a session, an LLM turn, a run — open a Session. Every Dome control-plane RPC issued while the session is active carries a stable X-Dome-Activity-Id header, and the resulting Dome audit events share that activity_id.

with client.session() as s:
    print("activity_id:", s.activity_id)  # a fresh, high-entropy UUID
    # Every Dome RPC in this block carries X-Dome-Activity-Id: <s.activity_id>.
    page = s.query_audit_events(event_types=("access.denied",))

# Outside the session, calls carry no activity_id — the SDK never mints one
# implicitly. To supply your own correlation id, pass it explicitly:
with client.session(activity_id="nightly-batch-2026-06-03") as s:
    ...

The id is minted as a UUIDv4 — a random, opaque token with no PII. (Dome mints it because a Dome-minted id is the only kind we can assert is PII-free.) Without a with block, use client.start_session() and call session.end() yourself, ideally in a finally. Sessions nest; exiting an inner session restores the outer one's id.

The header is first-party only — it rides on the SDK's calls to the Dome control plane and is never attached to third-party requests (e.g. OIDC/JWKS discovery). Server-side, a caller-supplied id is recorded as CALLER_ASSERTED. Note that streaming results are lazy: iterate s.stream_audit_events(...) inside the with block so the request fires while the session is still active.

Authorization Checks

check() requires on_allow and on_deny callbacks — both outcomes must be handled:

client.check(
    tool="db:query",
    on_allow=lambda result: execute_query(),
    on_deny=lambda req, reason: log_denial(reason),
)

Act-As (End-User Delegation)

Pass the end-user identity to enable user-aware authorization rules:

# Plain mode (development / trusted environments)
client.check(
    tool="db:query",
    act_as=dome.ActAs(sub="user-123", email="[e-mail]"),
    on_allow=..., on_deny=...,
)

# OIDC mode (verified — pass the user's real token)
client.check(
    tool="db:query",
    act_as=user_oidc_token,  # raw JWT string from the IdP
    on_allow=..., on_deny=...,
)

SDK errors (not policy denials) still raise exceptions:

try:
    client.check(tool="db:query", on_allow=..., on_deny=...)
except dome.NoBundleLoadedError:
    # No rule bundle available (fail-closed)
    pass
except dome.NotInitializedError:
    # SDK not started yet
    pass
except dome.ShutdownError:
    # SDK already closed
    pass

Fail-Closed

The SDK denies all requests by default when:

• No rule bundle has been loaded → NoBundleLoadedError
• No rules match the request → calls on_deny
• The SDK has not been started → NotInitializedError

Documentation

• Dome Platform Docs

License

Proprietary. See LICENSE for details.