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: RuntimeError if the channel name is invalid (must match org/namespace/service format) or 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: RuntimeError 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!")

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)

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.

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 = 60.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.

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: 60s)

Raises: RuntimeError if the token is invalid, expired, or 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 the LangGraph, CrewAI, and MCP integrations use under the hood. Use it directly if you need custom framework support.

---

Envelope

New in v0.3.0. 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)

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

New in v0.3.0. Protocol-specific adapters for A2A, 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")

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, MCP ↔ SLIM, A2A ↔ MCP.

---

Error handling

All SDK methods raise RuntimeError on failure. Error messages indicate the subsystem:

Error prefix	Cause
MLS engine error:	MLS protocol failure (bad key package, invalid Welcome, decryption error)
transport error:	Network failure (relay unreachable, gRPC connection error)
invalid channel name:	Channel name not in org/namespace/service format
auth error:	API key invalid or expired
runtime error:	Internal SDK error

try:
    channel = client.create_channel("bad-name")
except RuntimeError as e:
    print(f"Failed: {e}")
    # "Failed: invalid channel name: bad-name (expected org/namespace/service)"

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()