SAGE Python SDK

Python client for the SAGE (Sovereign Agent Governed Experience) protocol -- a governed, verifiable institutional memory layer for multi-agent systems.

Requires Python 3.10+ | SAGE v11.17.9 SDK | TLS, app-v26 explicit Access Group authority, app-v24 memory integrity, app-v25 immutable envelopes and historical continuity recovery, canonical local and federated Messages with read receipts, read-only federation, scoped governance, and per-record classification supported

Installation

# From PyPI
pip install sage-agent-sdk

# From source (development)
git clone https://github.com/l33tdawg/sage.git
cd sage/sdk/python
pip install -e .

# With dev/test dependencies
pip install -e ".[dev]"

Quickstart

from sage_sdk import SageClient, AgentIdentity

# Generate a new agent identity (Ed25519 keypair)
identity = AgentIdentity.generate()

# Save for reuse across sessions
identity.to_file("my_agent.key")

# Connect to a SAGE node
client = SageClient(base_url="http://localhost:8080", identity=identity)

# Register yourself on-chain
reg = client.register_agent(name="my-agent", role="member", provider="python-sdk")
print(f"Registered: {reg.agent_id}")

# Submit a memory
result = client.propose(
    content="Flask web challenges with SQLi require prepared statements bypass",
    memory_type="fact",
    domain_tag="challenge_generation",
    confidence=0.85,
)
print(f"Memory {result.memory_id} submitted (tx: {result.tx_hash})")

# Query by vector similarity
matches = client.query(
    embedding=[0.1] * 768,  # 768-dim (nomic-embed-text)
    domain_tag="challenge_generation",
    min_confidence=0.7,
    top_k=5,
)
for mem in matches.results:
    print(f"  [{mem.status.value}] {mem.content[:80]}")

# Vote on a proposed memory
client.vote(result.memory_id, decision="accept", rationale="Verified correct")

Authentication

SAGE uses Ed25519 keypairs for agent identity. Every API request is signed with the agent's private key.

from sage_sdk import AgentIdentity

# Generate a new identity
identity = AgentIdentity.generate()

# The agent_id is the hex-encoded public key
print(identity.agent_id)  # e.g. "a1b2c3d4..."

# Persist to disk
identity.to_file("agent.key")

# Load from disk
identity = AgentIdentity.from_file("agent.key")

# Create from a known 32-byte seed (deterministic)
identity = AgentIdentity.from_seed(b"\x00" * 32)

Request signing is handled automatically by the client. Each request includes four headers:

If you sign requests by hand instead of using the SDK, always include a fresh 8-byte nonce (auth.py). The generic authentication middleware can still verify the old nonce-less signature shape during the compatibility window, but exact message, acknowledgement, receipt, and delegated-governance actions reject it. Current integrations must not rely on nonce-less signing.

Complete API Reference

SageClient exposes 84 public operations. AsyncSageClient exposes those same 84 operations as coroutines plus the async-only close() lifecycle method, for 85 public methods total.

Health & Status

# Check node health (unauthenticated)
client.health()      # GET /health
client.ready()       # GET /ready

Agent Registration & Management

Before an agent can participate in the SAGE network, it must register on-chain. Registration creates an immutable identity record tied to the agent's Ed25519 public key.

# Register on-chain (first time only — idempotent)
reg = client.register_agent(
    name="security-analyst",       # Human-readable name
    role="member",                 # self-registration never self-promotes
    boot_bio="Analyzes CVEs",      # Optional: agent description
    provider="claude-code",        # Optional: LLM provider identifier
)
# Returns: AgentRegistration(agent_id, name, role, provider, status, tx_hash)

# Update your profile
client.update_agent(name="security-analyst-v2", boot_bio="Updated bio")

# Get your caller-scoped profile and access standing
profile = client.get_profile()       # GET /v1/agent/me
print(profile.enrollment_status, profile.home_domain, profile.can_write)

# Get any registered agent's info
agent = client.get_agent("a1b2c3...")  # GET /v1/agent/{id}
# Returns: AgentInfo(agent_id, name, role, clearance, org_id, dept_id, ...)

# List active ordinary agents visible to this signed caller
agents = client.list_agents()        # GET /v1/agents → {"agents": [...], "total": N}

# Lightweight local recipient directory and bounded human-name resolution.
# Neither response is evidence that a recipient is online or has read a message.
directory = client.agent_directory()
matches = client.lookup_agents("mynah", limit=10)

# Page the caller's authoritative owned-domain inventory without loading a
# roster or scanning memories.
owned = client.owned_domains(limit=50)
sample = client.domain_access_sample()

# Agent roles, operating modes, app-v26 Access Group membership/authority, and
# grants are governed in the local CEREBRUM Access Controls screen; the SDK
# deliberately exposes no legacy per-agent permission mutation shortcut.

The desktop app listens on IPv4 loopback at 127.0.0.1:8080 by default. localhost in the example is a client-side loopback alias, not a LAN bind, and the SDK does not rewrite it. If localhost resolves only to an unbound IPv6 ::1, use http://127.0.0.1:8080. A LAN base_url can reach only routes the node deliberately exposes there; it cannot satisfy the direct-loopback peer and Host checks protecting CEREBRUM operator actions.

Memory Operations

# Submit a memory transaction and wait for its chain commit
result = client.propose(
    content="The observation text",
    memory_type="fact",           # "fact", "observation", "inference", or "task"
    domain_tag="security",
    confidence=0.9,               # 0.0 - 1.0
    embedding=[0.1, 0.2, ...],    # Optional: precomputed 768-dim vector
    knowledge_triples=[           # Optional: structured knowledge
        KnowledgeTriple(subject="SQLi", predicate="bypasses", object_="prepared_statements")
    ],
    parent_hash="abc123",         # Optional: link to parent memory
    classification=3,             # Optional: per-record clearance 0-4 (3=Secret); omitted = PUBLIC(0)
)
# Returns MemorySubmitResponse with the chain receipt. For app-v23 tasks it
# also carries task_status, projection_confirmed, idempotency_key, replay state,
# retryable, and a reconciliation message when needed.

# Query by vector similarity
results = client.query(
    embedding=[0.1] * 768,       # Required: 768-dim query vector
    domain_tag="security",       # Optional: filter by domain
    min_confidence=0.7,          # Optional: minimum confidence
    top_k=10,                    # Number of results (default: 10)
    status_filter="committed",   # Optional: filter by status
    cursor="abc123",             # Optional: pagination cursor
)
# Returns: MemoryQueryResponse(results, next_cursor, total_count)
# App-v23 examines at most 8,192 raw authorization candidates per node request;
# HTTP 422 means narrow domain/provider/tag/status filters.

# Hybrid BM25/vector recall with optional query expansions
hybrid = client.hybrid(
    query="credential rotation",
    embedding=[0.1] * 768,
    domain_tag="security",
    expansions=[
        {"query": "rotate access keys", "embedding": [0.2] * 768},
    ],                              # At most 8 submitted entries
)
# Under app-v23 the primary query, all expansions, and every text/vector store
# leaf share one 8,192-candidate live-authorization budget. HTTP 422 means reduce
# expansions or narrow filters. Governed leaf/budget failures fail the whole
# call; the server never returns HTTP 200 with partial hybrid fusion.

# Get a single memory
memory = client.get_memory("550e8400-e29b-41d4-a716-446655440000")

# List memories with filtering and pagination
memories = client.list_memories(
    limit=50,                    # 1-200 (default: 50)
    offset=0,
    domain="security",           # Optional: filter by domain
    status="committed",          # Optional: filter by status
    sort="newest",               # "newest", "oldest", or "confidence"
    agent="a1b2c3...",           # Optional: filter by agent
)
# Returns: MemoryListResponse(memories, total, limit, offset, has_more,
# total_exact, filtered). Under app-v23 total is a visible lower bound while
# has_more is true; denied/raw counts are never exposed. Visible offset is
# capped at 7,900 and one request examines at most 8,192 raw authorization
# candidates; HTTP 422 means narrow filters or page sequentially.

# Get memory timeline (time-bucketed counts)
timeline = client.timeline(
    domain="security",           # Optional
    bucket="day",                # "hour", "day", or "week"
    from_time="2026-03-01T00:00:00Z",
    to_time="2026-03-16T00:00:00Z",
)
# Returns: TimelineResponse(buckets=[{period, count, domain}], total=N)
# App-v23 counts only records that pass current live disclosure; aggregate
# existence/timing never bypasses RBAC or classification. Range is capped at
# 31 days and 8,192 raw candidates per request.

# Link related memories
client.link_memories(
    source_id="mem-1",
    target_id="mem-2",
    link_type="related",         # Default: "related"
)

# Dry-run validation (check without submitting)
result = client.pre_validate(
    content="Test content",
    domain="security",
    memory_type="fact",
    confidence=0.9,
)
# Returns: PreValidateResponse(accepted, votes=[{validator, decision, reason}], quorum)

Task Management

Task memories are a special memory type for tracking actionable work items.

# Submit a task
result = client.propose(
    content="Investigate CVE-2026-1234",
    memory_type="task",
    domain_tag=None,               # App-v23: use this agent's owned home domain
    confidence=0.9,
)

if result.projection_confirmed is False:
    # The transaction is already committed. Reconcile this exact ID; retrying
    # creation would be wrong even though the serving projection is unavailable.
    print(f"Reconcile {result.memory_id}; do not resubmit")
elif result.idempotent_replay:
    print(f"Existing task at {result.task_status}; no new task was created")
else:
    print(f"Created {result.memory_id}")

# Omission derives a permanent semantic key. To intentionally create another
# occurrence with identical content/domain, supply a fresh explicit key:
recurring = client.propose(
    content="Investigate CVE-2026-1234",
    memory_type="task",
    domain_tag="security",
    confidence=0.9,
    idempotency_key="investigate-cve-2026-1234-occurrence-2",
)

# List open tasks
tasks = client.list_tasks(
    domain="security",           # Optional
    provider="claude-code",      # Optional: filter by provider
)
# Returns: TaskListResponse(tasks=[{memory_id, content, domain_tag, task_status, ...}], total)

# Update task status
client.update_task_status(result.memory_id, "in_progress")  # in_progress/done/dropped

The task feed is exact-assignee and live-authorized: assignment never bypasses current domain/group/grant or classification access. Unassigned pickup, re-planning, and terminal reopen are local CEREBRUM operator actions.

Voting & Validation

# Vote on a proposed memory
client.vote(
    memory_id="550e8400-...",
    decision="accept",            # "accept", "reject", or "abstain"
    rationale="Verified correct",
)

# Challenge a committed memory
client.challenge(
    memory_id="550e8400-...",
    reason="Outdated information",
    evidence="See CVE-2024-XXXX",
)

# Corroborate (strengthen confidence)
client.corroborate(
    memory_id="550e8400-...",
    evidence="Independently verified via testing",
)

# Get memories pending validation
pending = client.get_pending(domain_tag="security", limit=20)

# Get current epoch info and validator scores
epoch = client.get_epoch()
# Returns: EpochInfo(epoch_num, block_height, scores=[{validator_id, current_weight, ...}])

Compatibility pipeline (agent-to-agent transport)

The pipe_* SDK methods remain for existing integrations and federated transport/receipt compatibility. New same-node SDK integrations should use the canonical message_* / messages_* methods below. New MCP clients should use the canonical sage_message_* workflow; sage_pipe* MCP names are deprecated and hidden from tool discovery. Legacy local pipe completion may journal a summary, while foreign work and canonical Messages are never silently promoted into governed memory.

# Send a message to another agent
msg = client.pipe_send(
    payload="Please analyze this CVE",
    to_agent="target-agent-id",  # Route by agent ID
    # OR: to_provider="chatgpt",  # Route by provider name
    intent="analysis",           # Optional: message intent
    ttl_minutes=1440,            # Explicit legacy expiry (maximum: 1440 = 24h)
)
# Returns: PipeSendResponse(pipe_id, status, expires_at)

# Check your inbox
inbox = client.pipe_inbox(limit=5)
for msg in inbox.items:
    print(f"From {msg.from_agent}: {msg.payload}")

# Claim a message for processing
client.pipe_claim(msg.pipe_id)

# Submit your result
result = client.pipe_result(msg.pipe_id, result="Analysis complete: CVE is critical")
# Returns: PipeResultResponse(status, journal_id) — auto-journaled to memory

# Inspect this node's local workflow row (not a delivery/read receipt;
# negotiated federated receipt-v2 evidence uses separate signed routes)
status = client.pipe_status(msg.pipe_id)

# List completed results
results = client.pipe_results(limit=5)

# Reopen retained messages without claiming or re-queueing them. History keeps
# claimed/completed rows while the normal transient pipeline retention window
# still retains them; it is not a federated delivery/read receipt.
received_history = client.pipe_inbox_history(limit=20)
sent_history = client.pipe_outbox(limit=20)

For same-node work that needs retry-safe delivery and read receipts, use the v11.17 canonical Messages service. It uses the same retained inbox rows, not a second queue:

sent = client.message_send(
    to_agent="target-agent-id",
    payload="Please review the incident notes",
    intent="review",
    idempotency_key="incident-42-review-v1",
)
# Omitted/None ttl_minutes keeps the message durable until handled. Pass an
# explicit value from 1 to 1440 only when the message should expire.

# The token makes a lost HTTP response safe: an exact retry returns the same
# ordered claimed batch rather than consuming later messages.
batch = client.messages_receive("session-2026-08-02-turn-1", limit=5)
for item in batch.items:
    pass
client.messages_mark_read_batch([item.message_id for item in batch.items])
for item in batch.items:
    client.message_reply(item.message_id, "Reviewed")

# Exact sender only; no payload or reply content is exposed here.
receipt = client.message_status(sent.message_id)
print(receipt.transport_status, receipt.read_status, receipt.workflow_status)

idempotency_key and receive_token are 1–256 bytes. Omitted/None/0 ttl_minutes is durable until handled; explicit expiry is 1–1440 minutes. Receive-token replay metadata is retained for 48 hours and bounded to 4096 tokens per agent; a purged/incomplete exact batch fails instead of claiming newer messages.

The asynchronous client exposes the same methods as coroutines. Federated sends continue to use the pipeline contact/revalidation path. When both peers negotiate federated-pipeline-receipts-v2, the exact sender can query the separate payload-free receipt projection through the REST/MCP receipt-status surface; a locally queued pipe still must never be described as remotely read. Receipt recipients pass the complete singular challenge response directly to pipe_receipt_record(), or the ready batch items to pipe_receipt_record_batch(); the SDK constructs every exact-event agent ID, nonce, signature, and canonical-request proof.

Embeddings

# Generate embeddings via SAGE's local Ollama (no cloud API calls)
embedding = client.embed("your text here")  # Returns 768-dim float list

Access Control (RBAC)

SAGE uses a hierarchical access control model. All operations are on-chain BFT transactions — immutable once committed.

Organization
  +-- Department (membership metadata and federation scope)
        +-- Domain (knowledge category — access-controlled)
              +-- Agent (with clearance level 0-4)

Clearance Levels

Setup Order

1. Register organization  -->  2. Create departments  -->  3. Register domains
4. Generate agent keypairs  -->  5. Add agents to org + depts  -->  6. Agents operate

For production domains, register the domain or grant access before handing writers their identities. If an authenticated agent submits to a genuinely unowned, non-shared domain, the chain auto-registers that domain to the first writer and grants that owner level-2 access.

Organization Management

# Register an organization (you become permanent admin)
org = admin_client.register_org("Acme Corp", description="AI security research")
org_id = org["org_id"]

# Get organization info
client.get_org(org_id)

# Add agents to the organization
admin_client.add_org_member(org_id, agent_id="a1b2c3...", clearance=2, role="member")

# List organization members
members = admin_client.list_org_members(org_id)

# Update an agent's clearance level
admin_client.set_org_clearance(org_id, agent_id="a1b2c3...", clearance=3)

# Remove an agent from the organization
admin_client.remove_org_member(org_id, agent_id="a1b2c3...")

Department Management

Departments are sub-groups within an organization. They are used for membership metadata and federation scoping (allowed_depts); they do not by themselves isolate all same-org memory visibility.

# Create departments
eng = admin_client.register_dept(org_id, name="Engineering", description="Core eng team")
eng_dept = eng["dept_id"]

security = admin_client.register_dept(org_id, name="Security", description="Security research")
sec_dept = security["dept_id"]

# Sub-departments
crypto = admin_client.register_dept(
    org_id, name="Cryptography", description="Crypto team", parent_dept=sec_dept
)

# List all departments
depts = admin_client.list_depts(org_id)

# Get department info
dept = admin_client.get_dept(org_id, sec_dept)

# Add agents to departments (used by department-scoped federation agreements)
admin_client.add_dept_member(org_id, sec_dept, agent_id="a1b2c3...", clearance=2)

# List department members
members = admin_client.list_dept_members(org_id, sec_dept)

# Remove from department
admin_client.remove_dept_member(org_id, sec_dept, agent_id="a1b2c3...")

Domain Registration & Access Control

Domains have on-chain ownership. The first writer of a genuinely unowned, non-shared domain becomes owner automatically; explicit registration is still recommended when you want a predictable owner before any agent writes.

# Register domains (you become the domain owner)
admin_client.register_domain(name="security.crypto", description="Cryptographic security")
admin_client.register_domain(name="security.web", description="Web security", parent="security")

# Get domain info
info = admin_client.get_domain("security.crypto")

# Request access to a domain
client.request_access(domain="security.crypto", justification="Need crypto data", level=2)

# Grant access (domain owner or ancestor owner only)
admin_client.grant_access(
    grantee_id="a1b2c3...",
    domain="security.crypto",
    level=2,                    # 1=read, 2=read+write, 3=modify on v11/app-v15
    expires_at=0,               # Unix timestamp, 0 = never
)

# Revoke access
admin_client.revoke_access(grantee_id="a1b2c3...", domain="security.crypto", reason="Decommissioned")

# List grants for an agent
grants = admin_client.list_grants(agent_id="a1b2c3...")

Access Rules

• Department membership scopes federation agreements when allowed_depts is set
• An agent in Org X cannot access ANY memories in Org Y unless a federation agreement exists
• An agent always has access to memories it submitted, regardless of RBAC
• Read and write access are enforced through REST preflight checks and consensus-side HasAccessMultiOrg

Cross-Organization Federation

Federation enables controlled data sharing between separate organizations.

# Org A proposes federation
fed = admin_a.propose_federation(
    target_org_id=org_b_id,
    allowed_depts=["Engineering"],  # Only Org B's Engineering dept gets access
    max_clearance=2,                # Cap at Confidential
    requires_approval=True,
)

# Org B approves
feds = admin_b.list_federations(org_b_id)
admin_b.approve_federation(feds[0]["federation_id"])

# Now: Org B's Engineering can query Org A's data up to clearance 2
# Org B's Research dept still CANNOT see Org A's data

# Revoke when partnership ends
admin_a.revoke_federation(fed["federation_id"], reason="Partnership ended")

# Get federation details
info = admin_a.get_federation(fed["federation_id"])

Federation rules:

• Both org admins must agree (propose + approve)
• allowed_depts restricts which departments in the TARGET org can access your data
• max_clearance caps the clearance level regardless of agent's actual clearance
• Revocation is immediate and on-chain

Domain Write Enforcement

SAGE enforces domain writes in the node:

• REST submit handlers check the caller's domain policy before broadcasting.
• The consensus path checks HasAccessMultiOrg before committing writes to owned domains.
• Post-v8 grants walk ancestor domains, so a grant on security can cover security.crypto where appropriate.
• Post-v8 access grants can auto-claim genuinely unowned, non-shared domains for the granter.
• v11/app-v15 adds level 3 for modify workflows; level 2 remains read+write.

Application-specific routing can still add a narrower policy on top, for example:

AGENT_DOMAIN_MAP = {
    "designer": ["design.generation", "design.patterns"],
    "evaluator": ["evaluation.calibration"],
}

def validate_submission(agent_name: str, domain_tag: str) -> bool:
    allowed = AGENT_DOMAIN_MAP.get(agent_name, [])
    return any(domain_tag.startswith(prefix) for prefix in allowed)

Domain Reassign Recovery

SAGE includes an access-control recovery primitive: a chain admin can take over a domain whose owner is unavailable or compromised. The flow is governance-gated: a domain_reassign proposal carries the new owner, optional parent, an open_to_shared flag, and (on app-v26) the exact observed current owner as its compare-and-swap binding; validators vote; once accepted, TxTypeDomainReassign consumes the proposal, transfers ownership, purges all existing grants on the domain, and optionally promotes the domain to shared.

The SDK exposes both a one-shot helper and the two underlying primitives.

from sage_sdk import SageClient, AgentIdentity

admin = AgentIdentity.from_file("chain-admin.key")
client = SageClient(base_url="http://localhost:8080", identity=admin)

# This identity must be the target validator node's configured governance
# operator. The node's live validator key remains the on-chain actor.

# One-shot: propose -> poll -> submit. Raises SageAPIError on
# owner change/reject/expire/cancel/timeout. On app-v26 the helper reads and
# binds the chain-authoritative current owner automatically; older chains keep
# the historical payload.
result = client.reassign_domain(
    domain="acme.engineering",
    new_owner_id="b" * 64,
    reason="original owner offboarded, restoring access for the team",
    open_to_shared=False,
    poll_interval_s=2.0,
    timeout_s=120.0,
)
print(result.tx_hash, "purged", result.purged_grants, "grants")

Governance responses carry a deterministic proposal_id (use it for vote/cancel) and a distinct CometBFT tx_hash. After app-v20, the SDK first fetches the authenticated /v1/governance/context and signs its validator ID and chain domain into every delegated mutation. Consensus treats the exact signed proposal as global-admin authorization while the target node's validator key remains the proposal/vote actor; one operator cannot make another reachable validator vote. Vote/cancel operators can remain validator-local and do not need the global proposal-admin key. Missing node key/operator/domain wiring returns 503 before broadcast, a different valid signer receives 403, and stale context receives 409; repeat the SDK call to fetch and sign fresh context.

If you want to drive the flow manually (e.g. you already accepted a proposal out of band), use the two primitives directly. governance_propose now accepts a payload kwarg — pass a dict and the SDK JSON-encodes + base64s it for you.

propose = client.governance_propose(
    operation="domain_reassign",
    target_id="acme.engineering",
    reason="recovery",
    payload={
        "domain": "acme.engineering",
        "new_owner_id": "b" * 64,
        "parent_domain": "",
        "open_to_shared": False,
        "expected_owner_id": "a" * 64,
    },
)
# ... validators vote, proposal hits status="executed" ...
result = client.submit_domain_reassign(
    domain="acme.engineering",
    new_owner_id="b" * 64,
    proposal_id=propose.proposal_id,
    open_to_shared=False,
    expected_owner_id="a" * 64,
)

App-v20 operators can inspect the canonical scope topology and immutable current-revision anchors without decoding Badger directly:

scopes = client.list_scopes()
for scope in scopes.scopes:
    print(scope.scope_id, scope.state, scope.revision, scope.revision_hash)

research = client.get_scope("research-quorum")
print([(m.validator_id, m.assigned_weight) for m in research.members])

Create the first canonical scope without hand-encoding binary governance payloads:

proposal = client.governance_propose_scope(
    scope={
        "scope_id": "research-quorum",
        "revision": 1,
        "state": "active",
        "controller_validator_id": validator_id,
        "domains": ["research"],
        "members": [
            {"validator_id": validator_id, "assigned_weight": 1},
        ],
    },
    reason="form the research replica quorum",
)

The server sorts domains and members canonically and supplies consensus-owned heights. For later revisions, include every member's historical joined_revision; scope and legacy binary payload cannot be combined.

Code 50 (shared domain not ownable) surfaces as HTTP 403 with shared domain not ownable in the error detail — see sage_sdk.exceptions for the documented mapping.

Async Client

For async/concurrent workloads, use AsyncSageClient — it has identical methods, all returning awaitables:

import asyncio
from sage_sdk import AsyncSageClient, AgentIdentity

async def main():
    identity = AgentIdentity.generate()
    async with AsyncSageClient(base_url="http://localhost:8080", identity=identity) as client:
        # Register
        await client.register_agent(name="async-agent", provider="python")

        # Submit a memory
        result = await client.propose(
            content="Async observation",
            memory_type="observation",
            domain_tag="testing",
            confidence=0.75,
        )

        # Concurrent queries
        results = await asyncio.gather(
            client.query(embedding=[0.1] * 768, domain_tag="security"),
            client.query(embedding=[0.2] * 768, domain_tag="testing"),
        )

        # Pipeline messaging
        msg = await client.pipe_send(payload="Hello", to_provider="chatgpt")
        inbox = await client.pipe_inbox()

asyncio.run(main())

Models

MemoryType

MemoryType.fact          # Verified factual knowledge
MemoryType.observation   # Agent-observed data
MemoryType.inference     # Derived conclusion
MemoryType.task          # Actionable work item

MemoryStatus

MemoryStatus.proposed     # Awaiting validation
MemoryStatus.validated    # Passed quorum vote
MemoryStatus.committed    # Finalized on-chain
MemoryStatus.challenged   # Under dispute
MemoryStatus.deprecated   # Superseded or invalidated

TaskStatus

TaskStatus.planned        # Not yet started
TaskStatus.in_progress    # Currently being worked on
TaskStatus.done           # Completed
TaskStatus.dropped        # Abandoned

PipelineStatus

PipelineStatus.pending    # Awaiting claim
PipelineStatus.claimed    # Being processed
PipelineStatus.completed  # Result submitted
PipelineStatus.expired    # TTL exceeded
PipelineStatus.failed     # Processing failed

Error Handling

from sage_sdk.exceptions import (
    SageError,            # Base exception
    SageAPIError,         # Any API error; includes structured RFC 7807 fields
    SageAuthError,        # 401/403; subclass of SageAPIError
    SageNotFoundError,    # 404 resource not found
    SageValidationError,  # 422 validation error
)

try:
    memory = client.get_memory("nonexistent-id")
except SageNotFoundError as e:
    print(f"Not found: {e.detail}")
except SageAuthError as e:
    # App-v23 canonical write denials preserve exact machine-readable guidance.
    print(e.reason_code, e.remedy, e.retryable)
except SageAPIError as e:
    print(f"API error {e.status_code}: {e.detail}")

SageAPIError exposes status_code, detail, error_type, reason_code, remedy, and retryable. For https://sage.dev/errors/domain-write-denied, branch on the structured reason_code and retryable=False, never by matching detail. In v11.15.0 a missing_write_grant remedy uses the owned-domain or a Root/Admin-approved Access Group whose explicit tier is Read + write or Read + write + modify; it does not imply that CEREBRUM ships a direct level-2 grant editor.

Configuration

client = SageClient(
    base_url="http://localhost:8080",  # SAGE node URL
    identity=identity,
    timeout=30.0,                      # Request timeout (default: 30s)
    ca_cert=None,                      # TLS CA cert path, False to disable, None for system default
)

# Use as context manager for automatic cleanup
with SageClient(base_url="http://localhost:8080", identity=identity) as client:
    profile = client.get_profile()

TLS Support (v6.5 Quorum Mode)

When connecting to a SAGE node running in quorum mode with encrypted node-to-node communication (TLS), use the ca_cert parameter to specify the CA certificate used by the quorum.

from sage_sdk import SageClient, AgentIdentity

identity = AgentIdentity.from_file("my_agent.key")

# Connect to a TLS-enabled SAGE node with the quorum CA certificate
client = SageClient(
    "https://sage-node:8443",
    identity,
    ca_cert="/path/to/ca.crt",
)

# All requests now use the custom CA for TLS verification
profile = client.get_profile()

The CA certificate (ca.crt) is included in agent bundles generated by quorum-init and quorum-join. Look for it in your node's data directory (e.g., ~/.sage/quorum/ca.crt).

Options

# Disable TLS verification for local development (NOT for production)
dev_client = SageClient("https://localhost:8443", identity, ca_cert=False)

The async client supports the same parameter:

async with AsyncSageClient("https://sage-node:8443", identity, ca_cert="/path/to/ca.crt") as client:
    await client.health()

Embeddings

SAGE uses 768-dimensional vectors (Ollama nomic-embed-text). Three options:

1. Direct Ollama (local agents)

import httpx
resp = httpx.post(
    "http://localhost:11434/api/embed",
    json={"model": "nomic-embed-text", "input": "your text"},
    timeout=30.0,
)
embedding = resp.json()["embeddings"][0]

2. SAGE Embed Endpoint (remote agents)

embedding = client.embed("your text here")  # Uses SAGE's Ollama

3. Hash Embedding (testing only)

import hashlib, struct

def hash_embed(text: str, dim: int = 768) -> list[float]:
    rounds = (dim * 4 + 31) // 32
    raw = b""
    current = text.encode("utf-8")
    for i in range(rounds):
        current = hashlib.sha256(current + struct.pack(">I", i)).digest()
        raw += current
    return [(struct.unpack(">I", raw[j*4:j*4+4])[0] / 2147483647.5) - 1.0 for j in range(dim)]

Complete API Reference Table

Memory

Agent

Pipeline

Canonical local Messages

Validator

Embedding

Organization

Department

Domain & Access

Federation

Governance

Health

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
python -m pytest tests/ -v

# Run async tests
python -m pytest tests/test_async_client.py -v

License

Apache 2.0 — see the project root LICENSE file.