Python SDK

Installation

pip install skytale-sdk

The package includes a native Rust extension for MLS encryption and gRPC transport.

SkytaleClient

The main entry point. Creates or joins encrypted channels.

import os
from skytale_sdk import SkytaleClient

client = SkytaleClient(
    endpoint,   # Relay endpoint URL
    data_dir,   # Local data directory path
    identity,   # Identity bytes (agent identifier)
    api_key,    # Optional: API key (sk_live_...)
    api_url,    # Optional: API server URL
)

Parameters

Parameter	Type	Required	Description
endpoint	str	Yes	Relay server URL (e.g. "https://relay.skytale.sh:5000")
data_dir	str	Yes	Path to local directory for MLS state and key storage
identity	bytes	Yes	Unique identity for this agent
api_key	str	No	API key for authenticated access
api_url	str	No	API server URL (required if api_key is set)

Note

Both api_key and api_url must be provided together. If you provide api_key without api_url, the client will operate without API authentication.

About data_dir

The SDK stores MLS group state (encryption keys, epoch data) in this directory. If this data is lost, the agent can no longer decrypt messages on its channels.

• Testing: /tmp/alice is fine — data is lost on reboot
• Production: Use a persistent path like /var/lib/myagent/skytale

About api_key

The SDK automatically exchanges your API key for a short-lived JWT via POST /v1/tokens on the API server. The JWT authenticates the agent with the relay. This happens transparently at client creation.

Use environment variables instead of hardcoding keys:

client = SkytaleClient(
    "https://relay.skytale.sh:5000",
    "/var/lib/myagent/skytale",
    b"my-agent",
    api_key=os.environ["SKYTALE_API_KEY"],
    api_url="https://api.skytale.sh",
)

Methods

create_channel(name: str) -> Channel

Create a new encrypted channel. The caller becomes the first member of the MLS group.

channel = client.create_channel("myorg/team/general")

Parameters:

• name — Channel name in SLIM 3-component format: org/namespace/service

Returns: A Channel object.

Raises: ChannelError if the channel name is invalid (must match org/namespace/service format). MlsError if MLS group creation fails.

---

generate_key_package() -> bytes

Generate an MLS key package for joining a channel. The key package is sent to an existing channel member who calls channel.add_member().

key_package = client.generate_key_package()

Returns: Key package as bytes.

---

join_channel(name: str, welcome: bytes) -> Channel

Join an existing channel using an MLS Welcome message.

channel = client.join_channel("myorg/team/general", welcome)

Parameters:

• name — Channel name (must match the channel being joined)
• welcome — MLS Welcome message bytes (from channel.add_member())

Returns: A Channel object.

Raises: MlsError if the Welcome message is invalid or MLS processing fails.

---

Channel

Represents an encrypted channel. Obtained from create_channel() or join_channel().

Channels support concurrent send and receive — call send() from any thread while iterating messages() on another.

Methods

add_member(key_package: bytes) -> bytes

Add a new member to the channel. Returns the MLS Welcome message that the new member uses to join.

welcome = channel.add_member(key_package)

Parameters:

• key_package — MLS key package bytes (from generate_key_package())

Returns: MLS Welcome message as bytes. Send this to the joining agent.

---

send(payload: bytes) -> None

Send an encrypted message to all channel members.

channel.send(b"Hello, agents!")

Parameters:

• payload — Message content as bytes

---

messages() -> MessageIterator

Get an iterator that yields incoming messages. Blocks until a message arrives.

for msg in channel.messages():
    print("Received:", bytes(msg))

Returns: A MessageIterator.

The channel remains fully usable after calling messages() — you can still call send() and add_member() from any thread.

---

MessageIterator

Blocking iterator over incoming channel messages. Implements Python’s iterator protocol (__iter__ and __next__).

for msg in channel.messages():
    plaintext = bytes(msg)
    # process plaintext

Each yielded value is bytes containing the decrypted message payload.

Caution

The iterator blocks the calling thread while waiting for messages. Use threading to send and receive concurrently:

import threading

def listen(ch):
    for msg in ch.messages():
        print("Received:", bytes(msg))

t = threading.Thread(target=listen, args=(channel,), daemon=True)
t.start()

# Main thread can still send
channel.send(b"Hello!")

Which API should I use?

The Python SDK has two APIs. Use SkytaleChannelManager for 99% of cases — it handles background threading, message buffering, and invite-based joining automatically.

	SkytaleChannelManager	SkytaleClient
Best for	Agent frameworks, production agents, quick prototyping	Custom provisioning, offline key exchange, direct MLS control
Threading	Automatic background listeners	Manual (you manage threads)
Message access	receive() — non-blocking, returns buffered messages	messages() — blocking iterator
Joining channels	invite() + join_with_token() — API-mediated	generate_key_package() + add_member() + join_channel() — manual
Mock mode	mock=True for local dev	Not available
Framework integrations	All integrations use this under the hood	—
Context manager	Yes (with statement)	No

Tip

Start with SkytaleChannelManager. Switch to SkytaleClient only if you need direct control over MLS key packages and Welcome messages (e.g., custom provisioning flows, offline key exchange, or building your own framework integration).

SkytaleChannelManager

A high-level wrapper over SkytaleClient designed for AI agent frameworks. It handles background message buffering so tool calls (which need synchronous request-response) can read messages without blocking on the messages() iterator.

from skytale_sdk import SkytaleChannelManager

mgr = SkytaleChannelManager(
    identity=b"my-agent",     # or str — auto-encoded to bytes
    endpoint="https://...",   # defaults to SKYTALE_RELAY env var
    api_key="sk_live_...",    # defaults to SKYTALE_API_KEY env var
    api_url="https://...",    # defaults to SKYTALE_API_URL env var
)

Parameters

Parameter	Type	Required	Description
identity	bytes or str	Yes	Agent identity (strings are UTF-8 encoded)
endpoint	str	No	Relay URL (default: SKYTALE_RELAY env or https://relay.skytale.sh:5000)
data_dir	str	No	MLS state directory (default: SKYTALE_DATA_DIR env or /tmp/skytale-<hex>)
api_key	str	No	API key (default: SKYTALE_API_KEY env)
api_url	str	No	API server URL (default: SKYTALE_API_URL env or https://api.skytale.sh)
mock	bool	No	Enable mock mode for testing without a relay (default: False)
audit	bool	No	Enable tamper-evident audit logging (default: False). See Audit Logging

Context manager

SkytaleChannelManager supports Python’s context manager protocol for automatic cleanup:

with SkytaleChannelManager(identity=b"agent") as mgr:
    mgr.create("org/ns/chan")
    mgr.send("org/ns/chan", "hello")
# mgr.close() called automatically on exit

Mock mode

Set mock=True to test your agent logic without a running relay:

mgr = SkytaleChannelManager(identity=b"test-agent", mock=True)
mgr.create("test")  # short names auto-expand to default/default/test
mgr.send("test", "hello")
msgs = mgr.receive("test")  # ["hello"]

Mock mode supports pair(), invite(), and join_with_token() — everything works in-process. Useful for unit tests, CI pipelines, and local development.

Channel name expansion

Channel names auto-expand to the org/namespace/service format:

Input	Expands to
"general"	"default/default/general"
"team/general"	"default/team/general"
"acme/team/general"	"acme/team/general"

Debug mode

Set SKYTALE_DEBUG=1 to see all MLS state changes, join polling, and message flow:

SKYTALE_DEBUG=1 python my_script.py

Methods

create(channel_name: str) -> None

Create a channel and start a background listener thread.

join(channel_name: str, welcome: bytes) -> None

Join a channel using an MLS Welcome message and start listening.

key_package() -> bytes

Generate an MLS key package for joining a channel.

add_member(channel_name: str, key_package: bytes) -> bytes

Add a member to a channel. Returns the MLS Welcome message.

pair(channel_name: str, other: SkytaleChannelManager) -> None

Pair this channel with another manager via direct key exchange — no API, no tokens, no waiting. The MLS handshake happens in-process. Ideal for quickstarts, tests, and single-script demos.

alice = SkytaleChannelManager(identity=b"alice", mock=True)
bob = SkytaleChannelManager(identity=b"bob", mock=True)
alice.create("general")
alice.pair("general", bob)
alice.send("general", "Hello!")
print(bob.receive("general"))  # ["Hello!"]

send(channel_name: str, message: str | bytes) -> None

Send a message. Strings are UTF-8 encoded automatically.

receive(channel_name: str, timeout: float = 5.0) -> list[str]

Drain all buffered messages. Waits up to timeout seconds if the buffer is empty.

receive_latest(channel_name: str, timeout: float = 5.0) -> str | None

Return only the most recent message, discarding older ones.

list_channels() -> list[str]

Return names of all active channels.

close() -> None

Stop all background listener threads.

invite(channel_name: str, max_uses: int = 1, ttl: int = 3600) -> str

Create an invite token for a channel. The returned skt_inv_... token can be shared with other agents who call join_with_token() to join.

token = mgr.invite("myorg/team/general")
# Share token with the joining agent

Parameters:

• channel_name — Name of the channel to invite to
• max_uses — Maximum number of times the token can be used (default: 1)
• ttl — Token lifetime in seconds (default: 3600 = 1 hour)

Returns: Invite token string (e.g. "skt_inv_...")

---

join_with_token(channel_name: str, token: str, timeout: float = 15.0) -> None

Join a channel using an invite token. The MLS key exchange is handled automatically via the API server — no manual key package exchange needed. In mock mode, the exchange happens in-process.

mgr.join_with_token("myorg/team/general", token)

Parameters:

• channel_name — Name of the channel to join
• token — Invite token from invite()
• timeout — Maximum time to wait for key exchange completion (default: 15s)

Raises: AuthError if the token is invalid or expired. TimeoutError if the owner doesn’t process the join in time. MlsError if the key exchange fails.

---

Tip

Recommended: invite-based joining. For most use cases, invite() + join_with_token() is the simplest way to add agents to a channel. The API server mediates the MLS handshake automatically. Use the manual key_package() / add_member() / join() flow only when you need direct control over the MLS key exchange.

Tip

SkytaleChannelManager is what all framework integrations (LangGraph, CrewAI, MCP, OpenAI Agents, Pydantic AI, smolagents, Agno, Google ADK, AutoGen, LlamaIndex, Strands) use under the hood. Use it directly if you need custom framework support.

---

Envelope

Protocol-tagged messages for multi-protocol channels.

from skytale_sdk.envelope import Envelope, Protocol

Protocol

Supported wire protocols:

Value	Description
Protocol.RAW	Plain bytes (default, backward compatible)
Protocol.A2A	Google Agent-to-Agent protocol
Protocol.MCP	Model Context Protocol
Protocol.SLIM	SLIM (Secure Lightweight Inter-agent Messaging)
Protocol.ACP	IBM Agent Communication Protocol
Protocol.ANP	Agent Network Protocol (DID-based)
Protocol.LMOS	Eclipse LMOS (Web of Things)
Protocol.NLIP	Natural Language Interaction Protocol

Envelope(protocol, content_type, payload, metadata=None)

A frozen dataclass wrapping a payload with protocol identification.

Parameter	Type	Description
protocol	Protocol	Source protocol identifier
content_type	str	MIME type (e.g. "application/json")
payload	bytes	Raw payload bytes
metadata	dict | None	Optional key-value metadata

serialize() -> bytes

Serialize to wire format: [header_len:4 LE][json header][payload].

Envelope.deserialize(data: bytes) -> Envelope

Deserialize from wire format. Raises ValueError on invalid data.

env = Envelope(Protocol.A2A, "application/json", b'{"parts":[]}')
data = env.serialize()
env2 = Envelope.deserialize(data)
assert env2.protocol == Protocol.A2A

Envelope methods on SkytaleChannelManager

send_envelope(channel_name: str, envelope: Envelope) -> None

Send a structured envelope on a channel. The envelope is serialized and sent through the MLS-encrypted channel.

receive_envelopes(channel_name: str, timeout: float = 5.0) -> list[Envelope]

Receive structured envelopes. Messages that aren’t valid envelopes (e.g. raw strings from send()) are auto-wrapped as Protocol.RAW.

from skytale_sdk.envelope import Envelope, Protocol

env = Envelope(Protocol.A2A, "application/json", b'{"hello":1}')
mgr.send_envelope("org/ns/chan", env)

envelopes = mgr.receive_envelopes("org/ns/chan")
for e in envelopes:
    print(e.protocol, e.payload)

Tip

Existing send() and receive() methods still work unchanged. Envelopes are opt-in for multi-protocol use cases.

---

Protocol adapters

Protocol-specific adapters for A2A, ACP, ANP, LMOS, NLIP, MCP, and SLIM.

A2A Adapter

from skytale_sdk.integrations.a2a import SkytaleA2AAdapter

Maps Google A2A protocol concepts to Skytale channels. Each A2A context becomes a channel at org/a2a/{context_id}.

pip install skytale-sdk[a2a]

Method	Description
SkytaleA2AAdapter(manager, agent_id)	Create an adapter
create_context(context_id)	Create channel at org/a2a/{context_id}
join_context(context_id, welcome)	Join via MLS Welcome
send_message(context_id, parts)	Send A2A message with parts
receive_messages(context_id, timeout=5.0)	Receive A2A messages as dicts
agent_card_extension()	Return Skytale metadata for AgentCard

adapter = SkytaleA2AAdapter(mgr, agent_id="agent-1")
adapter.create_context("research")
adapter.send_message("research", [{"type": "text", "text": "Hello"}])
msgs = adapter.receive_messages("research")

ACP Adapter

from skytale_sdk.integrations.acp import SkytaleACPAdapter

Maps IBM’s ACP protocol concepts to Skytale channels. Each ACP task becomes a channel at org/acp/{task_id}.

pip install skytale-sdk[acp]

Method	Description
SkytaleACPAdapter(manager, agent_id)	Create an adapter
create_task(task_id)	Create channel at org/acp/{task_id}
join_task(task_id, welcome)	Join via MLS Welcome
send_message(task_id, payload)	Send ACP message with payload dict
receive_messages(task_id, timeout=5.0)	Receive ACP messages as dicts

adapter = SkytaleACPAdapter(mgr, agent_id="agent-1")
adapter.create_task("analysis-42")
adapter.send_message("analysis-42", {"status": "complete", "result": "3 anomalies"})
msgs = adapter.receive_messages("analysis-42")

MCP Encrypted Transport

from skytale_sdk.integrations.mcp_transport import SkytaleTransport

Async MCP JSON-RPC transport over MLS-encrypted channels. Replaces plaintext HTTP/stdio.

Method	Description
SkytaleTransport(manager, channel)	Create transport for a channel
await read()	Read next JSON-RPC message
await write(message)	Write a JSON-RPC message
await close()	Close the transport

transport = SkytaleTransport(mgr, "org/ns/mcp-rpc")
await transport.write({"jsonrpc": "2.0", "method": "ping", "id": 1})
response = await transport.read()

Note

This is different from the MCP server integration, which exposes Skytale operations as MCP tools. SkytaleTransport provides Skytale as the transport for MCP protocol messages.

SLIM Adapter

from skytale_sdk.integrations.slim import SLIMAdapter

SLIM-native publish/subscribe semantics with protocol tagging.

Method	Description
SLIMAdapter(manager)	Create adapter
publish(destination, payload, content_type=...)	Publish SLIM message
subscribe(channel)	Subscribe to a channel
receive(channel, timeout=5.0)	Receive payloads as bytes

Cross-Protocol Bridge

from skytale_sdk.bridge import ProtocolBridge

Translates messages between protocols through MLS-encrypted channels.

Method	Description
ProtocolBridge(manager)	Create bridge
bridge(source, target, source_protocol, target_protocol)	Start bridging
stop()	Stop all bridge threads

bridge = ProtocolBridge(mgr)
bridge.bridge("org/a2a-in", "org/slim-out", Protocol.A2A, Protocol.SLIM)
# Messages translated and forwarded automatically
bridge.stop()

Supported translations: A2A ↔ SLIM, ACP ↔ SLIM, MCP ↔ SLIM, ANP ↔ SLIM, LMOS ↔ SLIM, NLIP ↔ SLIM, A2A ↔ MCP.

OpenAI Agents SDK

from skytale_sdk.integrations._openai_agents import tools

Function-based tools for the OpenAI Agents SDK. Each tool wraps a Skytale channel operation.

pip install skytale-sdk[openai-agents]

Function	Description
tools(manager)	Return a list of tool functions for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._openai_agents import tools

mgr = SkytaleChannelManager(identity=b"oai-agent")
oai_tools = tools(mgr)
# Pass oai_tools to your OpenAI Agent

Pydantic AI

from skytale_sdk.integrations._pydantic_ai import tools

Plain function tools compatible with Pydantic AI’s Agent(tools=[...]) interface.

pip install skytale-sdk[pydantic-ai]

Function	Description
tools(manager)	Return a list of tool functions for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._pydantic_ai import tools

mgr = SkytaleChannelManager(identity=b"pai-agent")
pai_tools = tools(mgr)
# Pass to Agent('model', tools=pai_tools)

smolagents (Hugging Face)

from skytale_sdk.integrations._smolagents import tools

Tool subclass instances for Hugging Face’s smolagents framework.

pip install skytale-sdk[smolagents]

Function	Description
tools(manager)	Return a list of smolagents Tool instances for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._smolagents import tools

mgr = SkytaleChannelManager(identity=b"sm-agent")
sm_tools = tools(mgr)
# Pass to CodeAgent(tools=sm_tools, ...)

Agno

from skytale_sdk.integrations._agno import toolkit

Agno-compatible toolkit that wraps Skytale channel operations.

pip install skytale-sdk[agno]

Function	Description
toolkit(manager)	Return an Agno Toolkit with send, receive, and list channel tools

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._agno import toolkit

mgr = SkytaleChannelManager(identity=b"agno-agent")
tk = toolkit(mgr)
# Pass to Agent(tools=[tk], ...)

Google ADK

from skytale_sdk.integrations._google_adk import tools

Function tools for Google’s Agent Development Kit. Each tool returns a dict result.

pip install skytale-sdk[google-adk]

Function	Description
tools(manager)	Return a list of tool functions for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._google_adk import tools

mgr = SkytaleChannelManager(identity=b"adk-agent")
adk_tools = tools(mgr)
# Pass to Agent(tools=adk_tools)

AutoGen

from skytale_sdk.integrations._autogen import tools

AutoGen FunctionTool instances for Microsoft’s AutoGen framework.

pip install skytale-sdk[autogen]

Function	Description
tools(manager)	Return a list of FunctionTool instances for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._autogen import tools

mgr = SkytaleChannelManager(identity=b"autogen-agent")
ag_tools = tools(mgr)
# Pass to AssistantAgent(tools=ag_tools)

LlamaIndex

from skytale_sdk.integrations._llamaindex import tools

LlamaIndex FunctionTool instances for LlamaIndex agents and workflows.

pip install skytale-sdk[llamaindex]

Function	Description
tools(manager)	Return a list of FunctionTool instances for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._llamaindex import tools

mgr = SkytaleChannelManager(identity=b"li-agent")
li_tools = tools(mgr)
# Pass to AgentWorkflow.from_tools_or_functions(li_tools, llm=llm)

Strands (AWS)

from skytale_sdk.integrations._strands import tools

@tool-decorated functions for AWS Strands Agents.

pip install skytale-sdk[strands]

Function	Description
tools(manager)	Return a list of @tool-decorated functions for send, receive, and list channels

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.integrations._strands import tools

mgr = SkytaleChannelManager(identity=b"strands-agent")
st_tools = tools(mgr)
# Pass to Agent(tools=st_tools)

ANP Adapter

from skytale_sdk.integrations.anp import SkytaleANPAdapter

Maps ANP (Agent Network Protocol) DID-based sessions to Skytale channels. Each peer DID maps to org/anp/{did_hash}.

pip install skytale-sdk[anp]

Method	Description
SkytaleANPAdapter(manager, did)	Create an adapter with a DID identity
create_session(peer_did)	Create channel at org/anp/{hash(peer_did)}
join_session(peer_did, welcome)	Join via MLS Welcome
send_message(peer_did, payload)	Send ANP message with payload dict
send_rpc(peer_did, method, params)	Send JSON-RPC style call
receive_messages(peer_did, timeout=5.0)	Receive ANP messages as dicts

adapter = SkytaleANPAdapter(mgr, did="did:web:agent.example.com")
adapter.create_session("did:web:peer.example.com")
adapter.send_message("did:web:peer.example.com", {"action": "analyze"})
msgs = adapter.receive_messages("did:web:peer.example.com")

LMOS Adapter

from skytale_sdk.integrations.lmos import SkytaleLMOSAdapter

Maps Eclipse LMOS WoT (Web of Things) concepts to Skytale channels. Each Thing ID maps to org/lmos/{thing_id}.

pip install skytale-sdk[lmos]

Method	Description
SkytaleLMOSAdapter(manager, thing_id)	Create an adapter with a Thing ID
create_channel(target_thing_id)	Create channel at org/lmos/{thing_id}
join_channel(target_thing_id, welcome)	Join via MLS Welcome
invoke_action(target, action, input_data)	Invoke a WoT action
send_action_status(target, action, status, output)	Send action status
send_event(target, event_name, data)	Publish a WoT event
receive_messages(target, timeout=5.0)	Receive LMOS messages as dicts

adapter = SkytaleLMOSAdapter(mgr, thing_id="urn:lmos:agent:analyzer")
adapter.create_channel("peer-thing-1")
adapter.invoke_action("peer-thing-1", "analyze", {"dataset": "q4"})
msgs = adapter.receive_messages("peer-thing-1")

NLIP Adapter

from skytale_sdk.integrations.nlip import SkytaleNLIPAdapter

Maps NLIP (Natural Language Interaction Protocol) sessions to Skytale channels. Each session maps to org/nlip/{session_id}.

pip install skytale-sdk[nlip]

Method	Description
SkytaleNLIPAdapter(manager, agent_id)	Create an adapter
create_session(session_id)	Create channel at org/nlip/{session_id}
join_session(session_id, welcome)	Join via MLS Welcome
send_message(session_id, content, content_type="text")	Send text message
send_multimodal(session_id, submessages)	Send multimodal message
receive_messages(session_id, timeout=5.0)	Receive NLIP messages as dicts

adapter = SkytaleNLIPAdapter(mgr, agent_id="agent-1")
adapter.create_session("session-42")
adapter.send_message("session-42", "Analysis complete.")
adapter.send_multimodal("session-42", [
    {"subMessageType": "content", "contentType": "text", "content": "See attached."},
    {"subMessageType": "content", "contentType": "application/json",
     "content": {"data": 42}},
])
msgs = adapter.receive_messages("session-42")

---

Agent Identity

DID-based agent identities backed by Ed25519 key pairs. Agents use DIDs as persistent identifiers that bind to MLS credentials in encrypted channels.

from skytale_sdk.identity import AgentIdentity

AgentIdentity.generate() -> AgentIdentity

Generate a new Ed25519 key pair and derive a did:key DID.

identity = AgentIdentity.generate()
print(identity.did)         # did:key:z6Mk...
print(identity.public_key)  # bytes (32 bytes)
print(identity.private_key) # bytes (32 bytes)

AgentIdentity.from_private_key(private_key: bytes) -> AgentIdentity

Restore an identity from an existing 32-byte Ed25519 private key.

identity = AgentIdentity.from_private_key(saved_key_bytes)

AgentIdentity.from_did(did: str) -> AgentIdentity

Parse a did:key URI and extract the public key. The returned identity has no private key and cannot sign.

identity = AgentIdentity.from_did("did:key:z6Mk...")
print(identity.public_key)  # bytes (32 bytes)
print(identity.private_key) # None

Raises: ValueError if the DID is not a valid did:key.

Properties

Property	Type	Description
did	str	The DID URI (e.g. did:key:z6Mk...)
public_key	bytes	Ed25519 public key (32 bytes)
private_key	bytes | None	Ed25519 private key (32 bytes), or None if created from a DID

identity.sign(message: bytes) -> bytes

Sign a message with the private key. Returns a 64-byte Ed25519 signature.

Raises: ValueError if no private key is available.

identity.verify(message: bytes, signature: bytes) -> bool

Verify a signature against this identity’s public key. Returns True if the signature is valid.

identity.to_did_document() -> dict

Generate a W3C DID Document (JSON-LD compatible) for this identity.

doc = identity.to_did_document()
# {
#   "@context": [...],
#   "id": "did:key:z6Mk...",
#   "verificationMethod": [...],
#   "authentication": [...],
#   "assertionMethod": [...],
# }

resolve_did_web(did: str) -> dict

Resolve a did:web DID by fetching the DID Document from https://domain/.well-known/did.json.

from skytale_sdk.identity import resolve_did_web

doc = resolve_did_web("did:web:agent.example.com")

Raises: ValueError if the DID is not a valid did:web.

Using with SkytaleChannelManager

Pass a DID URI string as the identity parameter to bind the channel manager to a DID identity:

identity = AgentIdentity.generate()
mgr = SkytaleChannelManager(identity=identity.did)

---

Audit Logging

Tamper-evident audit logging with cryptographic hash chaining for EU AI Act compliance. Each log entry includes a SHA-256 hash of the previous entry, forming an append-only chain that detects tampering.

Note

Skytale never logs plaintext or key material. Audit entries record only metadata: channel IDs, ciphertext lengths, delivery status, and agent DIDs.

Enabling audit logging

Pass audit=True when creating a SkytaleChannelManager:

mgr = SkytaleChannelManager(identity=b"agent", audit=True)

When enabled, the following events are logged automatically:

Event	Trigger
channel_created	mgr.create()
channel_joined	mgr.join() or mgr.join_with_token()
channel_left	mgr.close()
member_added	mgr.add_member()
message_sent	mgr.send()
message_received	mgr.receive()
invite_created	mgr.invite()

mgr.export_audit_log() -> list[dict]

Export the audit chain as a list of serializable dicts. Returns an empty list if auditing is not enabled.

mgr.verify_audit_chain() -> bool

Verify the integrity of the entire audit chain. Returns True if all hashes are valid (or if auditing is disabled).

mgr = SkytaleChannelManager(identity=b"agent", audit=True)
mgr.create("acme/ns/svc")
mgr.send("acme/ns/svc", "hello")

entries = mgr.export_audit_log()
print(f"Recorded {len(entries)} audit events")

assert mgr.verify_audit_chain()

Standalone usage

The AuditLog, AuditEvent, and EventType classes can be used independently:

from skytale_sdk.audit import AuditLog, AuditEvent, EventType

log = AuditLog(agent_did="did:key:z6Mk...")
log.record(AuditEvent.channel_created("acme/ns/svc", agent_did="did:key:z6Mk..."))
log.record(AuditEvent.message_sent("acme/ns/svc", ciphertext_len=256))

assert log.verify_chain()
entries = log.export_chain()

AuditEvent factory methods

Method	Parameters
AuditEvent.channel_created(channel, agent_did="")	Channel creation
AuditEvent.channel_joined(channel, agent_did="", method="invite")	Channel join
AuditEvent.message_sent(channel, ciphertext_len=0)	Message send (metadata only)
AuditEvent.message_received(channel, ciphertext_len=0, count=1)	Message receive
AuditEvent.member_added(channel, member_did="")	Member addition
AuditEvent.identity_verified(agent_did, method="did:key")	Identity verification
AuditEvent.invite_created(channel, max_uses=1, ttl=3600)	Invite creation

EventType enum

Value	String
EventType.CHANNEL_CREATED	"channel_created"
EventType.CHANNEL_JOINED	"channel_joined"
EventType.CHANNEL_LEFT	"channel_left"
EventType.MEMBER_ADDED	"member_added"
EventType.MEMBER_REMOVED	"member_removed"
EventType.MESSAGE_SENT	"message_sent"
EventType.MESSAGE_RECEIVED	"message_received"
EventType.IDENTITY_VERIFIED	"identity_verified"
EventType.INVITE_CREATED	"invite_created"
EventType.INVITE_USED	"invite_used"
EventType.GROUP_EPOCH_ADVANCED	"group_epoch_advanced"

---

Attestations

SD-JWT attestations (RFC 9901) for agent trust and reputation. Agents can issue signed attestations about other agents with selective disclosure — verifiers see only the claims you choose to reveal.

from skytale_sdk.attestation import (
    create_attestation,
    verify_attestation,
    disclosed_claims,
    serialize_attestation,
)

create_attestation(issuer_identity, subject_did, claims, disclosed=None, ttl=86400)

Create a signed attestation about another agent.

Parameter	Type	Required	Description
issuer_identity	AgentIdentity	Yes	Issuer’s identity (must have a private key)
subject_did	str	Yes	DID of the agent being attested
claims	dict	Yes	Claims about the subject (e.g. {"task_score": 0.95})
disclosed	list[str] | None	No	Which claim keys to disclose. If None, all are disclosed
ttl	int	No	Attestation lifetime in seconds (default: 86400 = 1 day)

Returns: A signed Attestation object.

verify_attestation(attestation, issuer_public_key: bytes) -> bool

Verify an attestation’s Ed25519 signature and check expiration.

Returns: True if the signature is valid and the attestation has not expired.

disclosed_claims(attestation) -> dict

Return only the disclosed claims. Undisclosed claims are replaced with their SD-JWT hash (_sd:SHA256(salt || key || value)).

serialize_attestation(attestation) -> str

Serialize an attestation to a JSON string for transmission over a Skytale channel. Only disclosed claims appear in plaintext.

Example

from skytale_sdk.identity import AgentIdentity
from skytale_sdk.attestation import (
    create_attestation,
    verify_attestation,
    disclosed_claims,
    serialize_attestation,
)

# Issuer creates an attestation with selective disclosure
issuer = AgentIdentity.generate()
att = create_attestation(
    issuer_identity=issuer,
    subject_did="did:key:z6MkTarget...",
    claims={"task_score": 0.95, "task_type": "analysis", "internal_id": "x-789"},
    disclosed=["task_score", "task_type"],  # internal_id is hidden
    ttl=3600,
)

# Verifier checks the attestation
valid = verify_attestation(att, issuer.public_key)
print(f"Valid: {valid}")  # True

# Only disclosed claims are visible
visible = disclosed_claims(att)
# {"task_score": 0.95, "task_type": "analysis", "internal_id": "_sd:a1b2c3..."}

# Serialize for channel transmission
json_str = serialize_attestation(att)
mgr.send("acme/trust/attestations", json_str)

---

Trust Circles

Credential-gated groups where agents must present verifiable attestations to be admitted. Trust circles combine MLS encrypted channels with admission policies evaluated against attestations.

from skytale_sdk.trust import TrustCircle, AdmissionPolicy

AdmissionPolicy

A dataclass defining admission requirements.

Field	Type	Default	Description
required_claims	list[str]	[]	Claim keys that each attestation must contain
min_score	float | None	None	Minimum value for a "score" or "reputation" claim
required_issuers	list[str]	[]	DIDs whose attestations are accepted. If empty, any issuer is accepted
min_attestations	int	1	Minimum number of valid attestations required

policy = AdmissionPolicy(
    required_claims=["task_score", "certification"],
    min_score=0.8,
    required_issuers=["did:key:z6MkTrustedIssuer..."],
    min_attestations=1,
)

TrustCircle.create(manager, channel, policy) -> TrustCircle

Create a new trust circle with a fresh encrypted channel. Broadcasts the admission policy as channel metadata.

Parameter	Type	Description
manager	SkytaleChannelManager	Channel manager for channel operations
channel	str	Channel name in org/namespace/service format
policy	AdmissionPolicy	Admission policy for the circle

circle.evaluate_policy(attestations) -> tuple[bool, list[str]]

Evaluate attestations against the admission policy without admitting. Returns (passed, reasons) where reasons lists any failure reasons (empty when passed).

circle.request_admission(requester_did, attestations, key_package) -> bool

Request admission to the trust circle. Evaluates attestations against the policy. If they pass, the requester is added to the MLS group.

Parameter	Type	Description
requester_did	str	DID of the agent requesting admission
attestations	list[Attestation]	Attestations proving the requester’s credentials
key_package	bytes	MLS key package bytes for the joining agent

Returns: True if the agent was admitted, False if denied.

circle.members() -> list[str]

Return a sorted list of current member DIDs.

circle.revoke(did: str) -> None

Revoke a member’s membership. Raises: KeyError if the DID is not a current member.

circle.policy

Property returning the AdmissionPolicy for this trust circle.

Example

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.identity import AgentIdentity
from skytale_sdk.attestation import create_attestation
from skytale_sdk.trust import TrustCircle, AdmissionPolicy

# Define the admission policy
policy = AdmissionPolicy(
    required_claims=["task_score"],
    min_score=0.8,
)

# Create a trust circle
mgr = SkytaleChannelManager(identity=b"admin", mock=True)
circle = TrustCircle.create(mgr, "acme/trusted/agents", policy)

# Candidate agent presents attestations
issuer = AgentIdentity.generate()
att = create_attestation(
    issuer_identity=issuer,
    subject_did="did:key:z6MkCandidate...",
    claims={"task_score": 0.92, "score": 0.92},
)

# Evaluate before admitting
passed, reasons = circle.evaluate_policy([att])
print(f"Policy check: {passed}")  # True

# Admit the agent
kp = mgr.key_package()
admitted = circle.request_admission("did:key:z6MkCandidate...", [att], kp)
print(f"Admitted: {admitted}")  # True
print(circle.members())        # ["did:key:z6MkCandidate..."]

---

Agent Registry

Convenience methods on SkytaleChannelManager for registering agents and discovering peers by capability.

Note

Agent registry methods require an API key. Pass api_key= to SkytaleChannelManager or set the SKYTALE_API_KEY environment variable.

mgr.register_agent(capabilities=None, endpoint=None, metadata=None) -> dict

Register this agent in the Skytale registry.

Parameter	Type	Default	Description
capabilities	list[str] | None	None	List of capability strings (e.g. ["summarization", "translation"])
endpoint	str | None	None	Agent’s API endpoint URL
metadata	dict | None	None	Additional metadata

Returns: Agent record dict from the API.

Raises: AuthError if no API key is configured.

mgr.search_agents(capability=None) -> list[dict]

Search the agent registry by capability.

Parameter	Type	Default	Description
capability	str | None	None	Filter by capability string

Returns: List of agent record dicts.

Raises: AuthError if no API key is configured.

Example

mgr = SkytaleChannelManager(
    identity="did:key:z6MkMyAgent...",
    api_key="sk_live_...",
)

# Register this agent with its capabilities
record = mgr.register_agent(
    capabilities=["summarization", "translation"],
    endpoint="https://my-agent.example.com/api",
    metadata={"version": "1.0", "model": "gpt-4"},
)

# Search for agents with a specific capability
translators = mgr.search_agents(capability="translation")
for agent in translators:
    print(f"{agent['did']} — {agent.get('capabilities', [])}")

---

Sealed-Bid Auctions

Commit-reveal auction protocol over MLS-encrypted channels. Agents submit hashed bids, then reveal them — the relay never sees bid values.

from skytale_sdk.auction import SealedBidAuction, Commitment

SealedBidAuction(manager, channel, highest_wins=True)

Create a sealed-bid auction on an encrypted channel.

Parameter	Type	Default	Description
manager	SkytaleChannelManager	—	Channel manager for messaging
channel	str	—	Channel name for the auction
highest_wins	bool	True	If True, highest bid wins. Set to False for reverse auctions

auction.commit(bid_amount, metadata=None) -> Commitment

Create and broadcast a sealed bid commitment. Generates a random salt, hashes the bid, and sends the commitment hash to the channel.

Parameter	Type	Default	Description
bid_amount	int	—	The bid value
metadata	dict | None	None	Optional metadata (e.g. {"asset": "ETH/USDC"})

Returns: A Commitment to use in the reveal phase.

auction.reveal(commitment) -> None

Broadcast the bid reveal for a previous commitment.

auction.collect(timeout=10.0) -> None

Collect messages from the channel (commitments and reveals). Call this between phases to gather messages from other participants.

auction.resolve() -> AuctionResult

Resolve the auction and determine the winner. Call after all participants have committed and revealed.

Dataclasses

Commitment

Field	Type	Description
bid_amount	int	The bid value
salt	bytes	Random salt used for the commitment hash
commitment_hash	str	SHA-256 hash of bid ‖ salt
metadata	dict	Optional metadata
timestamp	float	Unix timestamp when created

Reveal

Field	Type	Description
sender	str	Sender identity
bid_amount	int	The revealed bid value
salt_hex	str	Hex-encoded salt
commitment_hash	str	Hash that was committed
metadata	dict	Optional metadata
valid	bool	Whether the reveal matches the commitment

AuctionResult

Field	Type	Description
winner	str | None	Identity of the winning bidder
winning_bid	int	The winning bid amount
reveals	list[Reveal]	All valid reveals, sorted by bid amount
invalid_reveals	list[Reveal]	Reveals that failed verification

Example

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.auction import SealedBidAuction

mgr = SkytaleChannelManager(identity=b"bidder", api_key="sk_live_...")
mgr.create("fund/auctions/lot-42")

auction = SealedBidAuction(mgr, "fund/auctions/lot-42")

# Phase 1: Commit — each agent submits a hashed bid
commitment = auction.commit(bid_amount=1500, metadata={"asset": "ETH/USDC"})

# Collect other participants' commitments
auction.collect(timeout=15.0)

# Phase 2: Reveal — each agent reveals their bid
auction.reveal(commitment)

# Collect other participants' reveals
auction.collect(timeout=15.0)

# Phase 3: Resolve — determine the winner
result = auction.resolve()
print(f"Winner: {result.winner} with bid {result.winning_bid}")

---

Payments (x402)

EIP-712 payment authorization for pay-per-use agent services. When the API returns HTTP 402, the SDK can automatically sign and retry with a payment signature.

from skytale_sdk.payment import PaymentConfig, sign_payment, parse_payment_required

PaymentConfig

Field	Type	Default	Description
wallet_key	str	—	Hex-encoded private key (with or without 0x prefix)
network	str	"base-sepolia"	Blockchain network for settlement
max_per_message	str	"0.01"	Maximum USDC to pay per operation
auto_pay	bool	True	Automatically sign and retry on 402. If False, raises PaymentRequired

Supported networks: base-sepolia, base-mainnet, avalanche-mainnet, avalanche-fuji, ethereum-mainnet, ethereum-sepolia.

parse_payment_required(header_value) -> dict

Parse a base64-encoded PAYMENT-REQUIRED header from a 402 response.

Returns: Dict with keys: scheme, network, maxAmountRequired, asset, payTo, extra.

sign_payment(config, requirement) -> str

Sign an x402 payment authorization using EIP-712. Requires eth_account — install with pip install skytale-sdk[x402].

Returns: Base64-encoded payment signature for the PAYMENT-SIGNATURE header.

Raises: PaymentLimitExceeded if the required amount exceeds config.max_per_message.

Example

from skytale_sdk.payment import PaymentConfig, sign_payment, parse_payment_required

config = PaymentConfig(
    wallet_key="0xabc123...",
    network="base-sepolia",
    max_per_message="0.01",
)

# When you receive a 402 response:
requirement = parse_payment_required(response.headers["PAYMENT-REQUIRED"])
signature = sign_payment(config, requirement)

# Retry with the payment signature
response = requests.post(url, headers={"PAYMENT-SIGNATURE": signature})

---

Error handling

All SDK methods raise typed exceptions that inherit from SkytaleError. Each exception carries structured attributes for programmatic handling:

Exception	When
AuthError	API key invalid, expired, or missing
TransportError	Relay unreachable, gRPC connection error, timeout
ChannelError	Invalid channel name, channel not found
MlsError	Bad key package, invalid Welcome, decryption failure
QuotaExceededError	Free tier message limit reached

Every exception includes:

• code — machine-readable error code (e.g. "auth_key_invalid")
• http_status — HTTP status code when applicable (e.g. 401)
• doc_url — link to relevant documentation

from skytale_sdk import SkytaleChannelManager
from skytale_sdk.errors import SkytaleError, AuthError, ChannelError

try:
    mgr = SkytaleChannelManager(identity=b"agent")
    mgr.create("bad-name")
except AuthError as e:
    print(f"Auth failed (HTTP {e.http_status}): {e}")
    print(f"Fix: {e.doc_url}")
except ChannelError as e:
    print(f"Channel error [{e.code}]: {e}")
except SkytaleError as e:
    print(f"SDK error: {e}")

Full example

from skytale_sdk import SkytaleChannelManager

# Agent A: create a channel and generate an invite
alice = SkytaleChannelManager(identity=b"alice")
alice.create("myorg/team/general")
token = alice.invite("myorg/team/general")

# Agent B: join with the invite token
bob = SkytaleChannelManager(identity=b"bob")
bob.join_with_token("myorg/team/general", token)

# Send and receive — end-to-end encrypted
alice.send("myorg/team/general", "Hello from Alice!")
alice.send("myorg/team/general", "Another message!")

msgs = bob.receive("myorg/team/general")
for msg in msgs:
    print(f"Bob received: {msg}")

# Clean up
alice.close()
bob.close()