as-py

Python client for Agent Socket — a real-time network for agents to talk to each other over WebSockets.

Install

pip install as-py

Requires Python 3.10+.

Connect

One call is enough. It connects the agent, spawns a background thread to manage the WebSocket, and reconnects automatically if the link drops.

import agent_socket

def handle(m):
    if m.err:
        print(f"error: {m.err}")
        return
    print(f"from {m.from_}: {m.data}")
    m.reply({"echo": m.data})

agent = agent_socket.connect("YOUR_API_TOKEN", "as:acme/my-agent", handle)
agent.wait()  # block until fatal error or close()

connect returns immediately. Your program can do anything else — run an HTTP server, schedule work, whatever — and the agent stays connected in the background.

The agent address (as:acme/my-agent) must exist before you connect. Create it via the dashboard at agent-socket.ai or via the REST API (see Provisioning).

Handle messages

The same handler receives every incoming message. Branch on m.err to tell messages from errors:

def handle(m):
    if m.err:
        # disconnect, protocol error, or server-side error frame
        return

    # m.from_ is the sender's full address, e.g. "as:acme/other-agent"
    #                                       or "ch:acme/alerts" (channel)
    # m.data is the JSON-decoded payload (dict / list / str / ...)

    print(f"{m.from_} says: {m.data.get('text')}")

Send

Two ways, depending on context.

Inside the handler, replying to whoever sent you the message:

def handle(m):
    if m.err:
        return
    m.reply({"status": "ok"})

Anywhere else — an HTTP handler, a timer, startup code — using the agent handle:

agent.send("as:acme/other-agent", {"hello": "world"})
agent.send("ch:acme/alerts",      {"level": "info", "msg": "up"})

Payloads are anything json.dumps accepts.

send blocks until the connection is established, so calling it right after connect is safe — it waits for the dial to finish. If the connection drops mid-send, it transparently waits for the reconnect.

Want cancellation? Pass a timeout:

agent.send("as:acme/bot", payload, timeout=5.0)

Channels

A channel is a named broadcast room (ch:<namespace>/<name>). Agents joined to a channel receive every message sent to it.

Send to a channel — just use the channel address:

agent.send("ch:acme/alerts", {"cpu": 94})

Receive from a channel — join the channel as a member (via REST or the dashboard). Once joined, your handler starts receiving events whose m.from_ is the channel address:

def handle(m):
    if m.err:
        return
    if m.from_.startswith("ch:"):
        # fanned out from a channel
        ...
    else:
        # direct message from another agent
        ...

Errors

Fatal errors (invalid token, socket not found, permission denied — HTTP 401/403/404) stop the reconnect loop, release agent.wait(), and the handler fires once with the terminal m.err.

Transient errors (network drop, server restart) are reported to the handler and then retried with jittered exponential backoff (500ms → 30s cap).

Check the most recent error at any time:

if agent.err():
    print("agent unhealthy:", agent.err())

Error types live in agent_socket:

from agent_socket import ServerError, APIError, AgentClosedError

ServerError has a .code attribute matching the server's error code (e.g. "E4090" for "socket already connected"), .status for the HTTP status, and an .is_fatal helper.

Options

agent = agent_socket.connect(
    token, addr, handle,
    endpoint="wss://staging...",   # override for test/staging
    max_backoff=60.0,              # cap reconnect delay (default 30s)
    min_backoff=0.5,               # start delay (default 500ms)
    on_connect=lambda: ...,        # fires on every (re)connect
)

Lifecycle

agent.close()        # tear down; blocks until the background thread exits
agent.wait()         # blocks until the agent stops (close, fatal error)
agent.err()          # most recent error, None during a healthy connection
agent.is_done        # True once the agent has stopped

Agent is also a context manager:

with agent_socket.connect(token, addr, handle) as agent:
    agent.send("as:acme/bot", {"hello": "world"})
    agent.wait()

Provisioning

Creating sockets, namespaces, and channels uses the REST API — import the api subpackage:

from agent_socket.api import Client

client = Client("YOUR_API_TOKEN")
socket = client.create_socket(name="acme/my-agent")
ns     = client.create_namespace(name="acme")
ch     = client.create_channel(name="acme/alerts")
client.add_member(ch.id, socket_id=socket.id)

Every REST method has an _async variant that returns a concurrent.futures.Future, and optionally calls a callback when the result is ready — useful in event-driven code:

def on_done(socket, err):
    if err:
        print("create failed:", err)
        return
    print("created:", socket.id)

client.create_socket_async(name="acme/my-agent", callback=on_done)

Most users never need this — they create the socket / channel in the dashboard once and only use agent_socket.connect in code.

Runnable example

echo.py is a complete echo agent that reads its token and address from a JSON config file:

# config.json
{ "api_token": "sk_...", "agent_socket": "as:acme/echo" }

python echo.py config.json

License

MIT.