skytale-sdk 0.3.2

Encrypted channels for AI agents

Skytale SDK

Encrypted channels for AI agents. Create MLS-encrypted communication channels with multi-protocol support (SLIM, A2A, ACP, MCP).

Install

pip install skytale-sdk

From source (development):

cd sdk
python3 -m venv .venv && source .venv/bin/activate
pip install maturin
maturin develop

Setup

Get an API key with the Skytale CLI:

# Install the CLI (one-time)
cd cli && cargo build --release
cp target/release/skytale ~/.cargo/bin/

# Create your account
skytale signup [e-mail]

Or set the key directly:

export SKYTALE_API_KEY="sk_live_..."

Quickstart

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")
print(token)  # "skt_inv_..." — share this with Agent B

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

# Send and receive
alice.send("myorg/team/general", "Hello from Alice!")
msgs = bob.receive("myorg/team/general")  # ["Hello from Alice!"]

API Reference

SkytaleChannelManager (recommended)

High-level API for most use cases. Handles background message buffering and environment-based configuration.

from skytale_sdk import SkytaleChannelManager
mgr = SkytaleChannelManager(identity=b"my-agent")

• identity (bytes | str) — agent identity (strings are UTF-8 encoded)
• endpoint (str) — relay URL (default: SKYTALE_RELAY env)
• data_dir (str) — MLS state directory (default: auto-generated)
• api_key (str) — API key (default: SKYTALE_API_KEY env)
• api_url (str) — API server URL (default: SKYTALE_API_URL env)

create(channel_name) -> None

Create a channel and start a background listener.

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

Generate an invite token for a channel. Returns an skt_inv_... token to share with other agents.

join_with_token(channel_name, token, timeout=60.0) -> None

Join a channel using an invite token. Handles MLS key exchange automatically.

send(channel_name, message) -> None

Send a message (strings are UTF-8 encoded automatically).

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

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

receive_latest(channel_name, timeout=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.

SkytaleClient (low-level)

Direct control over MLS key packages, Welcome messages, and channel objects. Use this when you need manual key exchange or custom MLS operations.

from skytale_sdk import SkytaleClient
client = SkytaleClient(endpoint, data_dir, identity, api_key=None, api_url=None)

• endpoint (str) — gRPC endpoint, e.g. "https://relay.skytale.sh:5000"
• data_dir (str) — directory for MLS state persistence (created if needed)
• identity (bytes) — unique agent identity (e.g. agent name or UUID)
• api_key (str) — API key for authenticated access (sk_live_...)
• api_url (str) — API server URL (required with api_key)

client.create_channel(name) -> Channel

Create and subscribe to a new encrypted channel.

client.generate_key_package() -> bytes

Generate a KeyPackage for other agents to add you to their channels.

client.join_channel(name, welcome_bytes) -> Channel

Join an existing channel from a Welcome message.

Channel

channel.add_member(key_package) -> bytes

Add an agent to this channel. Returns Welcome bytes for the new member.

channel.send(payload)

Send an encrypted message (bytes).

channel.messages() -> MessageIterator

Return a blocking iterator over decrypted incoming messages.

MessageIterator

Implements Python's iterator protocol. Each item is bytes (the decrypted payload).

for msg in channel.messages():
    print(bytes(msg))

Architecture

Your Agent --> SkytaleClient --> gRPC --> Relay --> gRPC --> SkytaleClient --> Their Agent
                  |                                              |
                  +-- MLS encrypt                  MLS decrypt --+

All messages are encrypted end-to-end with MLS (RFC 9420). The relay cannot read message contents.

Multi-Protocol Support (v0.3.0)

Envelope

Protocol-tagged messages for multi-protocol channels:

from skytale_sdk.envelope import Envelope, Protocol

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

SkytaleChannelManager envelope methods

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

Send a structured envelope on a channel.

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

Receive envelopes from a channel. Raw messages (from send()) are auto-wrapped as Protocol.RAW.

A2A Adapter

from skytale_sdk.integrations.a2a import SkytaleA2AAdapter

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

Maps A2A contexts to channels (org/a2a/{context_id}). Messages are JSON-serialized, envelope-tagged, and MLS-encrypted.

MCP Encrypted Transport

from skytale_sdk.integrations.mcp_transport import SkytaleTransport

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

MCP JSON-RPC over MLS-encrypted channels instead of plaintext HTTP/stdio.

ACP Adapter

from skytale_sdk.integrations.acp import SkytaleACPAdapter

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

Maps ACP tasks to channels (org/acp/{task_id}). Messages are JSON-serialized, envelope-tagged, and MLS-encrypted.

SLIM Adapter

from skytale_sdk.integrations.slim import SLIMAdapter

slim = SLIMAdapter(mgr)
slim.subscribe("org/ns/chan")
slim.publish("org/ns/chan", b"hello")
payloads = slim.receive("org/ns/chan")

Cross-Protocol Bridge

from skytale_sdk.bridge import ProtocolBridge
from skytale_sdk.envelope import Protocol

bridge = ProtocolBridge(mgr)
bridge.bridge("org/a2a-agents", "org/slim-agents", Protocol.A2A, Protocol.SLIM)
# Messages flow automatically with translation
bridge.stop()

Translates between A2A, ACP, MCP, and SLIM. Runs background threads per bridge link.

License

Apache 2.0