Reference

Complete reference for the dome-sdk package on PyPI (imported as dome). All configuration is explicit — the SDK does not read environment variables.

Client construction

The SDK ships two clients with identical surfaces: dome.Client (sync) and dome.AsyncClient (async). Both accept keyword arguments only.

import dome

client = dome.Client(
    token="{{AGENT_TOKEN}}",
    gateway_url="https://gateway.domesystems.ai/gateways/{{GATEWAY_ID}}",
    control_plane_url="https://api.domesystems.ai",
    act_as_method="none",
)
client.connect()

gateway_url names the Gateway this client uses (/gateways/{uuid}). A client maps to one Gateway; there is no implicit global endpoint. connect() validates the URL and verifies that gateway_id, when supplied, names the same Gateway.

connect() prepares credentials and transport. It performs a token exchange when needed for control-plane discovery, but never blocks on local Cedar sync.

Gateways

A Gateway is a workspace-scoped grouping of MCP tools and LLM models with its own access grants and protocol endpoints. The SDK accepts only a complete scoped URL:

Agent-key creation and token exchange return a GatewayEndpoints object containing all four call-ready values. Use gateway_url as returned; do not derive it from a bare data-plane host. If the client has control_plane_url, pass gateway_id to select a Gateway during exchange. Omission succeeds only when the agent can access exactly one Gateway; the workspace default marker does not resolve ambiguity.

The tools-list cache is keyed by the complete Gateway URL, agent, workspace, and act-as identity, so clients pointed at different Gateways never share entries. Find IDs with dome gateways list, the dashboard, or the admin surface.

Client surface

AsyncClient exposes the same members; methods marked synchronous below have async counterparts on AsyncClient.

Gateway tools

client.gateway.tools is the MCP tool client.

tools/list can do real work: upstream discovery, per-user authorization, audit emission, and credential-link generation. The SDK never does an implicit list-before-call.

catalog = client.gateway.tools.list(refresh=True, act_as=user)
for tool in catalog.tools:
    print(tool.name, tool.description)

for advisory in catalog.auth_required:
    print("credential required:", advisory.connection_name, advisory.provision_url)

The cache is in-memory, bounded, and keyed by gateway root URL, agent fingerprint, workspace, act-as method, and an act-as header hash. Cache hits do not call the gateway and do not emit gateway audit. Per-user credential advisories on tools/list are returned on the fresh response and are not cached.

Calling a tool

result = client.gateway.tools.call(
    "github/list_issues",
    {"repo": "dome"},
    act_as=dome.PlainActAs(email="alice@corp.com"),
)
if result.is_error:
    print("upstream tool error:", result.structured_content or result.content)
else:
    print(result.structured_content or result.content)

call() distinguishes JSON-RPC errors from MCP tool errors. JSON-RPC errors raise typed DomeGatewayError subclasses. If the upstream tool returns isError=true, the SDK raises DomeToolExecutionError by default. Pass raise_on_tool_error=False to receive ToolCallResult(is_error=True) instead.

Reading tool results

ToolCallResult exposes convenience accessors so you do not have to walk the MCP content array by hand.

result = client.gateway.tools.call(
    "github/list_issues",
    {"repo": "dome"},
    act_as=user,
).require_success()

issues = result.json()
summary = result.first_text("no summary returned")

Gateway LLM

client.gateway.llm posts OpenAI- and Anthropic-shaped requests through the Dome LLM ingress and decodes structured Dome errors.

response = client.gateway.llm.chat(
    model="prod-gpt",
    messages=[{"role": "user", "content": "Summarize the open incidents"}],
    act_as=dome.PlainActAs(email="alice@corp.com"),
)

For a typed handle around one model, use client.gateway.model(...):

model = client.gateway.model("prod-claude", provider="anthropic", act_as=user)
message = model.messages.create(
    messages=[{"role": "user", "content": "Draft a status update"}],
    max_tokens=512,
)

Provider-native clients

For teams that want provider-native APIs, the gateway hands back a configured openai/anthropic client pointed at the Dome ingress:

openai_client = client.gateway.openai_client(act_as=user)
completion = openai_client.chat.completions.create(
    model="prod-gpt",
    messages=[{"role": "user", "content": "Hello"}],
)

Provider factories return provider-shaped clients and do not promise typed Dome exceptions unless Dome owns the transport for that path. Use client.gateway.llm.* or client.gateway.model(...) when you want SDK-owned typed error decoding.

Gateway readiness

client.gateway.wait_ready() polls unauthenticated GET /ready and is an explicit diagnostic helper — not a startup requirement.

client.gateway.wait_ready(timeout=30)

For freshly created agent keys, use dome.wait_for_agent_key(...) or dome.bootstrap.ensure_agent(..., wait_gateway=True) in setup scripts to wait for the gateway's synced API-key snapshot.

Act-as

The gateway act-as trust model is driven by Dome configuration, not by the client choosing a header shape. Pass the SDK value that matches the gateway method:

# Plain caller-asserted identity
user = dome.PlainActAs(email="alice@corp.com", roles=("admin",))

# Verified OIDC delegation
user = dome.OIDCActAs(jwt="eyJ...")

# HMAC-signed identity
user = dome.HMACActAs(secret=hmac_secret, email="alice@corp.com")

For bound, the SDK fails closed if caller code tries to send an act-as header.

Activity correlation

Wrap a run in an activity context to stamp every gateway call and audit event with the same opaque activity ID.

with client.activity(metadata={"case": "incident-123"}) as activity:
    client.gateway.tools.call("github/list_issues", {"repo": "dome"})
    page = client.audit.query(event_types=("tool.call",))
    print(activity.activity_id)

Direct calls outside an activity carry no activity ID — the SDK never mints one implicitly. To thread an ID across processes, mint an opaque ID and put human-readable labels in metadata:

activity_id = dome.new_activity_id()

with client.activity(activity_id, metadata={"job": "nightly-batch"}) as activity:
    ...

AuditEventV1.correlation.activity_id and activity_trust surface the stamp on each event.

dome.new_activity_id() -> str

Mint a fresh UUIDv4 activity ID. Use when persisting an ID for later correlation. Equivalent to the ID client.activity() mints when none is passed.

Audit reads

client.audit queries the hosted Audit v1 read APIs through the configured control plane. The SDK does not write hosted audit events — gateway and control-plane services do that.

from dome import AuditPayloadFilter, AuditQuery

page = client.audit.query(
    AuditQuery(
        event_types=("model.call",),
        results=("RESULT_ALLOWED",),
        producer_service="gateway",
        request_surface="INITIATOR_SURFACE_GATEWAY_LLM",
        start_time="2026-06-23T00:00:00Z",
        end_time="2026-06-24T00:00:00Z",
        payload_filters=(
            AuditPayloadFilter(field="model", values=("prod-gpt",)),
        ),
        page_size=50,
    )
)

for event in page.events:
    print(event.type, event.correlation.activity_id, event.payload)

AuditQuery enum fields use the proto JSON enum names (for example RESULT_ALLOWED, ACTOR_KIND_AGENT, INITIATOR_SURFACE_GATEWAY_MCP). stream rejects query-only filters (agent_ids, primary_resource_id, primary_resource_kind, trace_id, activity_id, workspace_id, start_time, end_time, page_size) at call time instead of silently dropping them.

Waiting on activity events

Gateway audit lands asynchronously. Use wait_for_activity in tests and post-run verification to block until the events you correlated under an activity ID are visible.

with client.activity() as run:
    client.gateway.tools.call("github/list_issues", {"repo": "dome"}, act_as=user)

events = client.audit.wait_for_activity(
    run.activity_id,
    type_prefix="tool.",
    timeout=10.0,
)
assert any(event.type == "tool.call" for event in events)

find_gateway_events is the non-blocking form — it issues one query and returns whatever is already visible. Both helpers default the workspace to the client's authenticated workspace; pass workspace_id= to scope a platform-key reader explicitly.

Audit envelope

AuditEventV1 is the normalized read-side envelope returned by every audit read method.

Local policy checks

In-process Cedar evaluation is still available when an agent needs a fast, self-enforced decision. Local checks are opt-in — they require start_policy_sync() and a control_plane_url.

client = dome.Client(
    token="{{AGENT_TOKEN}}",
    control_plane_url="https://api.domesystems.ai",
)
client.start_policy_sync()

decision = client.evaluate(tool="database/query", action="mcp:call")
if decision.allowed:
    run_query()

For the two most common MCP decisions, use the typed shortcuts. Both wrap evaluate() with resource_type="mcp_tool" and the right action verb.

decision = client.evaluate_tool_call("database/query", act_as=user)
if not decision.allowed:
    raise PermissionError(decision.reason)

visible = [
    tool for tool in catalog.tools
    if client.evaluate_tool_discovery(tool.name, act_as=user).allowed
]

The callback form mirrors the Go SDK:

client.check(
    tool="database/query",
    action="mcp:call",
    on_allow=lambda result: run_query(),
    on_deny=lambda req, reason: log.warning("denied: %s", reason),
)

Admin client

DomeAdminClient is the workspace-scoped provisioning surface. Use it from orchestrator code or setup scripts to register agents, issue keys, deploy Cedar bundles, and configure gateway MCP and LLM connections. It authenticates with a workspace platform key (dome_pk_...) — not an agent token — and every call hits the control plane.

Client and DomeAdminClient are separate on purpose: Client is the runtime agent surface (authorization, audit reads, gateway calls); DomeAdminClient is the admin surface that creates the agents Client runs as.

from dome import DomeAdminClient, AgentActAsConfig

with DomeAdminClient(
    base_url="https://api.domesystems.ai",
    platform_key="{{PLATFORM_KEY}}",
    workspace_id="{{WORKSPACE_ID}}",
) as admin:
    agent = admin.register_agent(
        name="incident-bot",
        act_as_config=AgentActAsConfig(method="oidc", required=True),
        allowed_tools=["github/list_issues"],
        allowed_gateway_ids=["{{GATEWAY_ID}}"],
    )
    key = admin.ensure_agent_key(
        agent_id=agent.id,
        name="default",
        gateway_id="{{GATEWAY_ID}}",
    )
    print(key.token, key.gateway_url)

Constructor

Agent registry

AgentKeyMaterial.token is the only place the bearer is ever returned. Persist it before discarding the response.

Workspace and gateway setup

Gateway admin

Create a Gateway, add tool sources, grant admission, and hand the resulting /gateways/{id} URL to agents.

The admin client covers Gateway CRUD, state/default selection, resource membership, per-agent access, and the all-agents grant. Gateway cost quotas live in the CLI, dashboard, and MCP tools.

from dome import DomeAdminClient

with DomeAdminClient(
    base_url="https://api.domesystems.ai",
    platform_key="{{PLATFORM_KEY}}",
    workspace_id="{{WORKSPACE_ID}}",
) as admin:
    gateway = admin.create_gateway(
        name="incident-response",
        description="Tools + models for the incident-response agents",
    )

    # Expose every tool this MCP connection publishes through the Gateway.
    admin.add_gateway_tool_source(
        gateway_id=gateway.id,
        connection_id="{{MCP_CONNECTION_ID}}",
    )

    # Admit every agent in the workspace.
    admin.set_gateway_all_agents_grant(
        gateway_id=gateway.id,
        granted=True,
    )

    # The control plane returns complete call-ready endpoints.
    print(gateway.gateway_url)
    print(gateway.endpoints.mcp_url)

Deleting a Gateway (delete_gateway) removes its memberships but leaves the underlying connections and pools unchanged. Callers still pointed at the deleted Gateway's URL fail closed; repoint them first.

Admin errors

Every admin RPC raises AdminAPIError on non-2xx responses. The exception carries path, status_code, and the full response body. body is truncated in the exception message so a verbose HTML error page does not produce an unreadable str().

from dome import AdminAPIError

try:
    admin.register_agent(name="incident-bot")
except AdminAPIError as exc:
    if exc.status_code == 409:
        print("agent already exists")
    else:
        raise

Bootstrap helpers

Setup scripts can provision a development agent and issue a fresh gateway key with one call.

agent = await dome.bootstrap.ensure_agent(
    name="incident-bot-dev",
    control_plane_url="https://api.domesystems.ai",
    platform_key="dome_pk_...",
    workspace_id="{{WORKSPACE_ID}}",
    allowed_tools=["github/list_issues"],
    gateway_id="{{GATEWAY_ID}}",
    wait_gateway=True,
)

print(agent.gateway_url)

Pass an existing DomeAdminClient as admin_client=, or pass control_plane_url, platform_key, and workspace_id and let the helper construct one. The returned BootstrapAgent exposes agent_id, token, gateway_id, and the complete gateway_url, plus the underlying AgentRecord and AgentKeyMaterial.

ensure_agent is idempotent. If an agent with the exact name exists, it is reused; if create_key=True, the named key is created or rotated and fresh key material is returned.

Pass wait_gateway=True to block until the Gateway has synced the new key. Select it with gateway_id=; gateway_url= is only an optional complete-URL override for the readiness probe and must name the same Gateway.

Errors

All error classes inherit from dome.DomeError.

Control-plane / lifecycle

Gateway data-plane

All gateway errors inherit from DomeGatewayError and carry status_code plus the raw response when available. request_id and activity_id are populated when the gateway response exposes them. gateway_id is stamped from the client's configured endpoint, so every gateway error names the Gateway the failing call was routed through.

try:
    client.gateway.tools.call("github/list_issues", {"repo": "dome"}, act_as=user)
except dome.DomeCredentialRequired as exc:
    print("connect", exc.connection_name, "at", exc.provision_url)
except dome.DomeAuthorizationDenied as exc:
    print("denied:", exc.reason_code, exc.determining_policy)

Async client

dome.AsyncClient has the same constructor and exposes the same surfaces with await-able methods.

import asyncio
import dome

async def main() -> None:
    async with dome.AsyncClient(
        token="{{AGENT_TOKEN}}",
        gateway_url="https://gateway.domesystems.ai/gateways/{{GATEWAY_ID}}",
        act_as_method="none",
    ) as client:
        result = await client.gateway.tools.call(
            "github/list_issues",
            {"repo": "dome"},
            act_as=dome.PlainActAs(email="alice@corp.com"),
        )
        print(result.structured_content or result.content)

asyncio.run(main())

client.audit.stream(...) returns an async iterator on AsyncClient. client.activity(...) returns an AsyncActivity and supports async with.

Framework adapters

Adapters are separate PyPI packages that wrap Client for a specific framework. Refer to Adapters for the catalog and integration details.

Next steps

Wrap the client for a framework, or wire credentials and Act-As from Develop:

• Adapters for framework wrappers
• Develop for credentials, endpoints, and Act-As envelopes
• Delegated agents when access depends on the person
• Examples for a governed agent pattern