Python SDK
Reference.

Installation

Requires Python 3.10 or higher.

pip install orgkernel

identity.issue()

Issue an AgentIdentity from a CSR. The service generates a new Ed25519 keypair; the private key is returned once and never stored server-side.

from orgkernel.schemas import AgentIdentityCSR

csr = AgentIdentityCSR(
    agent_name="invoice-processor",
    org_id="acme-corp",
    requested_ou="finance_team",
    public_key="<agent-ed25519-public-key-base64url>",
    purpose="automated-invoice-processing",
)

result = await identity_svc.issue_from_csr(csr)
print(result.identity.agent_id)      # → "aid_7f3k9..."
print(result.agent_private_key_pem) # Store this — never sent again

Returns: AgentIdentityIssueResult { identity, certificate, agent_private_key_pem, ca_fingerprint }

identity.list()

List all AgentIdentities for an organization (no pagination).

identities = await identity_svc.list_by_org(org_id="acme-corp")
for identity in identities:
    print(f"{identity.agent_id}: {identity.agent_name} ({identity.status})")

identity.verify()

Static verification of an AgentIdentity: checks status (ACTIVE / SUSPENDED / REVOKED / EXPIRED) and expiry. For cryptographic proof of key possession, use challenge.request() + challenge.verify().

from orgkernel.schemas import AgentIdentityVerify

result = await identity_svc.verify(AgentIdentityVerify(agent_id="aid_7f3k9..."))
print(result.is_active)  # → True / False
print(result.message)    # human-readable

identity.challenge.request()

Generate a cryptographic challenge for an agent. The agent must sign the nonce with its private key and respond via challenge.verify().

challenge = await identity_svc.request_challenge(
    agent_id="aid_7f3k9...",
    issued_by="platform-gateway",
    ttl_seconds=300,
)
nonce = challenge.nonce

identity.challenge.verify()

Verify an agent's signed challenge. Confirms the agent holds the private key corresponding to its identity — cryptographically, not just by status.

from orgkernel.schemas import ChallengeResponse

result = await identity_svc.verify_challenge(
    ChallengeResponse(
        challenge_id=challenge.challenge_id,
        agent_id="aid_7f3k9...",
        certificate_id="aid_7f3k9...",
        public_key="<agent-public-key-base64url>",
        nonce=nonce,
        signature="<agent-ed25519-signature>",
    )
)
assert result.overall_valid is True

identity.suspend()

Suspend an AgentIdentity. Recoverable via identity.reactivate().

identity = await identity_svc.suspend(agent_id="aid_7f3k9...")
print(identity.status)  # → SUSPENDED

identity.reactivate()

Reactivate a suspended AgentIdentity. REVOKED identities cannot be reactivated.

identity = await identity_svc.reactivate(agent_id="aid_7f3k9...")
print(identity.status)  # → ACTIVE

identity.revoke()

Permanently revoke an AgentIdentity. This is irreversible.

from orgkernel.schemas import AgentIdentityRevoke

identity = await identity_svc.revoke(
    agent_id="aid_7f3k9...",
    data=AgentIdentityRevoke(
        revoked_by="aid_admin01",
        reason="agent compromised — key rotation required",
    ),
)
print(identity.status)  # → REVOKED

token.mint()

Mint a scoped, time-bounded ExecutionToken. The token is signed by the Org CA (prevents Token Grafting). Call this at mission start.

from datetime import datetime, timedelta, timezone
from orgkernel.schemas import ExecutionTokenCreate

token = await token_svc.mint(
    ExecutionTokenCreate(
        agent_id="aid_7f3k9...",
        mission_id="msn_invoice01",
        execution_scope=["read_invoice", "write_payment_draft"],
        expires_at=datetime.now(timezone.utc) + timedelta(hours=4),
    )
)
print(token.token_id)  # → "tok_abc123..."

token.scope_check()

Validate a proposed tool call against an ExecutionToken's scope. Call this from your Tool Gateway before every external tool invocation. (Implemented as check_scope on the service.)

from orgkernel.schemas import ScopeCheckRequest

result = await token_svc.check_scope(
    ScopeCheckRequest(
        token_id="tok_abc123...",
        tool_name="read_invoice",
        params={"invoice_id": "INV-4521"},
    )
)
assert result.passed is True

token.use()

Mark an ExecutionToken as consumed. Typically called when a mission completes or a single-shot token is spent. (Implemented as mark_used on the service.)

token = await token_svc.mark_used(token_id="tok_abc123...")
print(token.used)  # → True

token.invalidate()

Invalidate an ExecutionToken early with a reason. Use when a mission aborts or a security incident occurs.

token = await token_svc.invalidate(
    token_id="tok_abc123...",
    reason="mission cancelled by operator",
)
print(token.invalidation_reason)  # → "mission cancelled by operator"

token.list()

List ExecutionTokens for a mission or an agent. Supports pagination via page_by_mission() / page_by_agent().

# All tokens for a mission (no pagination)
tokens = await token_svc.list_by_mission(mission_id="msn_invoice01")

# Paginated
result = await token_svc.page_by_mission(
    mission_id="msn_invoice01",
    page_no=1,
    page_size=20,
)
for t in result["items"]:
    print(f"{t.token_id}: used={t.used}, valid={t.is_valid}")

audit.initialize()

Initialize an AuditChain for a mission. Writes the genesis IDENTITY entry. Call once at mission start.

from orgkernel.schemas import AuditLayer

chain_id = await audit_svc.initialize(
    db,
    mission_id="msn_invoice01",
    agent_id="aid_7f3k9...",
)

audit.append()

Append an entry to an AuditChain. Call after each significant action. The chain must not be closed.

entry = await audit_svc.append(
    db,
    chain_id=chain_id,
    layer=AuditLayer.EXECUTION,
    event="EXECUTION_tool_call",
    agent_id="aid_7f3k9...",
    mission_id="msn_invoice01",
    data={"tool": "read_invoice", "invoice_id": "INV-4521", "result": "success"},
    token_id="tok_abc123...",
)
print(entry.sequence)     # monotonically increasing
print(entry.entry_hash)  # SHA-256 of this entry

audit.close()

Close an AuditChain. No further entries may be appended. Call when the mission ends.

chain = await audit_svc.close(db, chain_id=chain_id)
print(chain.head_hash)
print(chain.closed_at)

audit.get_chain()

Retrieve the full AuditChain, including all entries with their hashes.

chain = await audit_svc.get_by_chain_id(db, chain_id)
# or by mission:
chain = await audit_svc.get_by_mission(db, mission_id="msn_invoice01")

for entry in chain.entries:
    print(f"  [{entry.layer}] {entry.event}: {entry.data}")

audit.verify()

Verify the integrity of an AuditChain: sequence continuity, hash chain validity, and entry self-consistency. (Implemented as verify_integrity on the service.)

valid = await audit_svc.verify_integrity(db, chain_id)
assert valid is True