Metadata-Version: 2.4
Name: dome-sdk
Version: 0.1.0
Summary: Dome Platform Python SDK — AI agent governance
Author-email: Dome Systems <eng@domesystems.ai>
License: Proprietary
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: cedarpy>=4.0
Requires-Dist: httpx>=0.27
Requires-Dist: pyjwt[crypto]>=2.0
Provides-Extra: dev
Requires-Dist: mypy>=1.13; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.8; extra == 'dev'
Description-Content-Type: text/markdown

# Dome Python SDK

The official Python SDK for the [Dome Platform](https://domesystems.ai) — AI agent governance with local Cedar evaluation.

## Install

```bash
pip install dome
```

## Quick Start

```python
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](https://github.com/k9securityio/cedar-py). 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`](packages/dome-langchain/) | `pip install dome-langchain` | LangChain |

### LangChain Example

```python
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.

```python
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`.

```python
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:

```python
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:

```python
# Plain mode (development / trusted environments)
client.check(
    tool="db:query",
    act_as=dome.ActAs(sub="user-123", email="alice@corp.com"),
    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:

```python
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](https://docs.domesystems.ai)

## License

Proprietary. See LICENSE for details.
