Python SDK - kanoniv-agent-auth

Cryptographic identity and delegation for AI agents. Native Rust extension compiled via PyO3 and maturin - the same Ed25519 engine as the Rust crate, exposed through Python bindings. Cross-language compatible with the Rust and TypeScript SDKs.

Installation

pip install kanoniv-agent-auth

Version: 0.2.0 | Python: 3.10+ | License: MIT

Prebuilt wheels are available for Linux (x86_64) and macOS (x86_64, ARM). If no wheel is available, the build requires Rust 1.70+.

INFO

This is a PyO3 native extension. The cryptographic operations run in Rust - Python is the interface layer. There are no pure-Python fallbacks.

---

Identity

Agent cryptographic identity - Ed25519 keypairs and did:agent: identifiers.

AgentKeyPair

An agent's Ed25519 keypair.

class AgentKeyPair:
    @staticmethod
    def generate() -> AgentKeyPair
        """Generate a new random keypair."""

    @staticmethod
    def from_bytes(secret: bytes) -> AgentKeyPair
        """Reconstruct from 32-byte secret key."""

    def secret_bytes(self) -> bytes
        """Export the 32-byte secret key."""

    def identity(self) -> AgentIdentity
        """Derive the public identity."""

    def sign(self, payload_json: str) -> SignedMessage
        """Sign a JSON payload string. Returns a SignedMessage."""

Example:

from kanoniv_agent_auth import AgentKeyPair

kp = AgentKeyPair.generate()
identity = kp.identity()
print(identity.did)  # did:agent:a1b2c3d4...

# Persist and restore
secret = kp.secret_bytes()
kp2 = AgentKeyPair.from_bytes(secret)
assert kp.identity().did == kp2.identity().did

AgentIdentity

Public identity derived from a keypair - safe to share and store.

class AgentIdentity:
    @property
    def did(self) -> str
        """The DID string (did:agent:{hex(sha256(pubkey)[..16])})."""

    @property
    def public_key_bytes(self) -> bytes
        """The raw 32-byte public key."""

    @property
    def created_at(self) -> str | None
        """RFC 3339 creation timestamp, or None if unknown."""

    @staticmethod
    def from_bytes(bytes: bytes) -> AgentIdentity
        """Reconstruct from raw 32-byte public key bytes."""

    def did_document(self) -> str
        """Generate a W3C DID Document as JSON string."""

    def did_document_with_services(self, services: list[ServiceEndpoint]) -> str
        """Generate a W3C DID Document with service endpoints."""

ServiceEndpoint

A service endpoint for DID Documents. Imported as ServiceEndpoint (aliased from PyServiceEndpoint internally).

class ServiceEndpoint:
    def __init__(self, id: str, service_type: str, endpoint: str) -> None

    @property
    def id(self) -> str

    @property
    def service_type(self) -> str

    @property
    def endpoint(self) -> str

Example:

from kanoniv_agent_auth import AgentKeyPair, ServiceEndpoint
import json

kp = AgentKeyPair.generate()
identity = kp.identity()

services = [
    ServiceEndpoint("#messaging", "AgentMessaging", "https://example.com/msg"),
    ServiceEndpoint("#resolve", "KanonivResolve", "https://api.kanoniv.com/v1/resolve"),
]
doc = json.loads(identity.did_document_with_services(services))

---

Signing

Signed message envelopes with Ed25519 signatures.

SignedMessage

A cryptographically signed message envelope.

class SignedMessage:
    @property
    def payload(self) -> str
        """The payload as JSON string."""

    @property
    def signer_did(self) -> str
        """The signer's DID."""

    @property
    def nonce(self) -> str
        """UUID v4 nonce for replay protection."""

    @property
    def timestamp(self) -> str
        """RFC 3339 timestamp (milliseconds, Z suffix)."""

    @property
    def signature(self) -> str
        """Hex-encoded Ed25519 signature (128 hex chars)."""

    def verify(self, identity: AgentIdentity) -> None
        """Verify against a known identity. Raises ValueError on failure."""

    def content_hash(self) -> str
        """Compute the SHA-256 content hash (64 hex chars)."""

    def to_json(self) -> str
        """Serialize to JSON string."""

    @staticmethod
    def from_json(json: str) -> SignedMessage
        """Deserialize from JSON string."""

Example:

import json
from kanoniv_agent_auth import AgentKeyPair

kp = AgentKeyPair.generate()
payload = json.dumps({"action": "merge", "entity_id": "abc123"})
signed = kp.sign(payload)

# Verify
signed.verify(kp.identity())  # OK - no exception

# Content hash for chaining
print(signed.content_hash())  # 64-char hex string

# Serialize / deserialize
json_str = signed.to_json()
restored = SignedMessage.from_json(json_str)
restored.verify(kp.identity())  # still valid

---

Delegation

Cryptographic delegation with attenuated capabilities.

Delegation

A cryptographic delegation of authority from one agent to another.

class Delegation:
    @staticmethod
    def create_root(
        issuer_keypair: AgentKeyPair,
        delegate_did: str,
        caveats_json: str,
    ) -> Delegation
        """Create a root delegation (issuer is the root authority).

        caveats_json is a JSON array of caveat objects, e.g.:
        '[{"type": "action_scope", "value": ["resolve", "search"]}]'
        """

    @staticmethod
    def delegate(
        issuer_keypair: AgentKeyPair,
        delegate_did: str,
        additional_caveats_json: str,
        parent: Delegation,
    ) -> Delegation
        """Create a delegated delegation (with parent chain).
        Caveats accumulate - can only narrow, never widen."""

    @property
    def issuer_did(self) -> str
        """DID of the agent that granted authority."""

    @property
    def delegate_did(self) -> str
        """DID of the agent that received authority."""

    @property
    def depth(self) -> int
        """Chain depth (0 for root)."""

    def content_hash(self) -> str
        """Content hash of this delegation's proof (for revocation)."""

Caveat JSON Format

Caveats are passed as a JSON string containing an array of objects:

import json

caveats = json.dumps([
    {"type": "action_scope", "value": ["resolve", "search"]},
    {"type": "expires_at", "value": "2030-01-01T00:00:00.000Z"},
    {"type": "max_cost", "value": 10.0},
    {"type": "resource", "value": "entity:customer:*"},
    {"type": "context", "value": {"key": "task_id", "value": "task-abc"}},
    {"type": "custom", "value": {"key": "org", "value": "acme"}},
])

Invocation

An agent exercising delegated authority.

class Invocation:
    @staticmethod
    def create(
        invoker_keypair: AgentKeyPair,
        action: str,
        args_json: str,
        delegation: Delegation,
    ) -> Invocation
        """Create an invocation exercising delegated authority."""

    @property
    def invoker_did(self) -> str
        """DID of the agent performing the action."""

    @property
    def action(self) -> str
        """The action being performed."""

verify_invocation()

Verify an invocation's full authority chain.

def verify_invocation(
    invocation: Invocation,
    invoker_identity: AgentIdentity,
    root_identity: AgentIdentity,
) -> tuple[str, str, list[str], int]
    """Returns (invoker_did, root_did, chain, depth).
    Raises ValueError on verification failure."""

Example:

import json
from kanoniv_agent_auth import AgentKeyPair, Delegation, Invocation, verify_invocation

root = AgentKeyPair.generate()
agent = AgentKeyPair.generate()

# Root delegates to agent: resolve only
delegation = Delegation.create_root(
    root,
    agent.identity().did,
    json.dumps([{"type": "action_scope", "value": ["resolve"]}]),
)

# Agent invokes
invocation = Invocation.create(
    agent,
    "resolve",
    json.dumps({"entity_id": "123"}),
    delegation,
)

# Verify the full chain
invoker_did, root_did, chain, depth = verify_invocation(
    invocation,
    agent.identity(),
    root.identity(),
)
print(f"Verified: {invoker_did} authorized by {root_did} (depth={depth})")

---

MCP

MCP (Model Context Protocol) authentication middleware.

McpProof

Self-contained invocation proof for MCP transport. Contains everything an MCP server needs to verify the agent's identity and authority without any external key resolver.

class McpProof:
    @staticmethod
    def create(
        invoker_keypair: AgentKeyPair,
        action: str,
        args_json: str,
        delegation: Delegation,
    ) -> McpProof
        """Create an MCP proof for a tool call."""

    @property
    def invoker_public_key(self) -> str
        """The invoker's Ed25519 public key as hex string (64 chars)."""

    @property
    def invoker_did(self) -> str
        """The invoker's DID."""

    @property
    def action(self) -> str
        """The action this proof authorizes."""

    def to_json(self) -> str
        """Serialize to JSON string (for embedding in MCP tool arguments)."""

    @staticmethod
    def from_json(json: str) -> McpProof
        """Deserialize from JSON string."""

verify_mcp_call()

Verify an MCP proof against a root authority.

def verify_mcp_call(
    proof: McpProof,
    root_identity: AgentIdentity,
) -> tuple[str, str, list[str], int]
    """Returns (invoker_did, root_did, chain, depth).
    Raises ValueError on verification failure."""

extract_mcp_proof()

Extract an MCP proof from tool arguments JSON.

def extract_mcp_proof(args_json: str) -> tuple[McpProof | None, str]
    """Returns (proof_or_none, clean_args_json).
    The _proof field is always stripped from the returned args."""

inject_mcp_proof()

Inject an MCP proof into tool arguments JSON.

def inject_mcp_proof(proof: McpProof, args_json: str) -> str
    """Returns the arguments JSON string with _proof field added."""

Example:

import json
from kanoniv_agent_auth import (
    AgentKeyPair, Delegation, McpProof,
    verify_mcp_call, extract_mcp_proof, inject_mcp_proof,
)

root = AgentKeyPair.generate()
agent = AgentKeyPair.generate()

delegation = Delegation.create_root(
    root,
    agent.identity().did,
    json.dumps([{"type": "action_scope", "value": ["resolve"]}]),
)

# Agent creates proof
proof = McpProof.create(
    agent,
    "resolve",
    json.dumps({"source": "crm", "external_id": "123"}),
    delegation,
)

# Inject into tool args
args_with_proof = inject_mcp_proof(
    proof,
    json.dumps({"source": "crm", "external_id": "123"}),
)

# Server extracts and verifies
extracted, clean_args = extract_mcp_proof(args_with_proof)
if extracted is not None:
    invoker_did, root_did, chain, depth = verify_mcp_call(
        extracted,
        root.identity(),
    )
    print(f"Verified agent: {invoker_did}")

---

Provenance

Signed audit trail for agent actions.

ProvenanceEntry

A signed provenance entry in the audit chain.

class ProvenanceEntry:
    @staticmethod
    def create(
        keypair: AgentKeyPair,
        action: str,
        entity_ids: list[str],
        parent_ids: list[str],
        metadata_json: str,
    ) -> ProvenanceEntry
        """Create and sign a new provenance entry.

        action is a string: "resolve", "merge", "split", "mutate",
        "ingest", "delegate", "revoke", or "custom:name".
        """

    @property
    def agent_did(self) -> str

    @property
    def action(self) -> str

    @property
    def entity_ids(self) -> list[str]

    @property
    def parent_ids(self) -> list[str]

    @property
    def metadata(self) -> str
        """Metadata as JSON string."""

    @property
    def signed_envelope(self) -> SignedMessage

    def verify(self, identity: AgentIdentity) -> None
        """Verify signature AND outer field integrity.
        Raises ValueError on failure."""

    def content_hash(self) -> str
        """Get the content hash (64 hex chars, usable as parent_id)."""

Example:

import json
from kanoniv_agent_auth import AgentKeyPair, ProvenanceEntry

kp = AgentKeyPair.generate()

# Create entries
resolve_entry = ProvenanceEntry.create(
    kp,
    "resolve",
    ["entity-1"],
    [],
    json.dumps({"source": "crm"}),
)

# Chain entries via content_hash
merge_entry = ProvenanceEntry.create(
    kp,
    "merge",
    ["entity-1", "entity-2"],
    [resolve_entry.content_hash()],
    json.dumps({"confidence": 0.97}),
)

# Verify
merge_entry.verify(kp.identity())  # OK
print(f"Merge hash: {merge_entry.content_hash()}")

---

Error Handling

All classes raise ValueError on cryptographic failures (signature verification, key length mismatches, integrity check failures, caveat violations, delegation chain errors). The error message contains details about the specific failure.

from kanoniv_agent_auth import AgentKeyPair, ProvenanceEntry

kp1 = AgentKeyPair.generate()
kp2 = AgentKeyPair.generate()

entry = ProvenanceEntry.create(kp1, "resolve", ["e1"], [], "{}")

try:
    entry.verify(kp2.identity())  # wrong agent
except ValueError as e:
    print(f"Verification failed: {e}")

---

Full Export List

from kanoniv_agent_auth import (
    # Identity
    AgentKeyPair,
    AgentIdentity,
    ServiceEndpoint,      # aliased from PyServiceEndpoint

    # Signing
    SignedMessage,

    # Delegation
    Delegation,
    Invocation,
    verify_invocation,

    # MCP
    McpProof,
    verify_mcp_call,
    extract_mcp_proof,
    inject_mcp_proof,

    # Provenance
    ProvenanceEntry,
)

TIP

The Python SDK does not export ActionType as a separate type. Action types are passed as plain strings: "resolve", "merge", "split", "mutate", "ingest", "delegate", "revoke", or "custom:name".

TIP

Caveats, metadata, and tool arguments are passed as JSON strings (not dicts) because the Rust boundary uses serde_json serialization. Always use json.dumps() to convert Python dicts before passing them to the library.