skytale-sdk 0.4.0

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, ANP, LMOS, NLIP).

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.

Context Manager

SkytaleChannelManager supports Python's context manager for automatic cleanup:

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

Mock Mode

Test agent logic without a running relay:

mgr = SkytaleChannelManager(identity=b"test", mock=True)
mgr.create("org/ns/test")
mgr.send("org/ns/test", "hello")
msgs = mgr.receive("org/ns/test")  # ["hello"]

Error Handling

All methods raise typed exceptions inheriting from SkytaleError:

from skytale_sdk.errors import SkytaleError, AuthError, TransportError

try:
    mgr.send("org/ns/chan", "hello")
except AuthError as e:
    print(f"Auth failed [{e.code}] (HTTP {e.http_status}): {e}")
except TransportError as e:
    print(f"Transport error: {e}")
except SkytaleError as e:
    print(f"SDK error: {e}")

Exception types: AuthError, TransportError, ChannelError, MlsError, QuotaExceededError. All carry code, http_status, and doc_url attributes.

Multi-Protocol Support

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, SLIM, ANP, LMOS, and NLIP. Runs background threads per bridge link. API requests auto-retry on transient failures (408, 429, 5xx) with configurable max_retries.

OpenAI Agents SDK

from skytale_sdk.integrations.openai_agents import tools

agent = Agent(name="my-agent", tools=tools(mgr))

Pydantic AI

from skytale_sdk.integrations.pydantic_ai import tools

agent = Agent('openai:gpt-4.1', tools=tools(mgr))

smolagents (Hugging Face)

from skytale_sdk.integrations.smolagents import tools

agent = CodeAgent(tools=tools(mgr), model=model)

Agno

from skytale_sdk.integrations.agno import toolkit

agent = Agent(model=OpenAIChat(id="gpt-4.1"), tools=[toolkit(mgr)])

Google ADK

from skytale_sdk.integrations.google_adk import tools

agent = Agent(model='gemini-2.0-flash', name='agent', tools=tools(mgr))

ANP Adapter

from skytale_sdk.integrations.anp import SkytaleANPAdapter

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": "hello"})
msgs = adapter.receive_messages("did:web:peer.example.com")

Maps ANP DID-based sessions to channels (org/anp/{did_hash}).

LMOS Adapter

from skytale_sdk.integrations.lmos import SkytaleLMOSAdapter

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

Maps Eclipse LMOS Thing IDs to channels (org/lmos/{thing_id}).

NLIP Adapter

from skytale_sdk.integrations.nlip import SkytaleNLIPAdapter

adapter = SkytaleNLIPAdapter(mgr, agent_id="agent-1")
adapter.create_session("session-42")
adapter.send_message("session-42", "Hello from NLIP!")
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")

Maps NLIP sessions to channels (org/nlip/{session_id}).

License

Apache 2.0