PlatoolsClient (WebSocket transport)

platools.connect() is a thin wrapper around PlatoolsClient. Most users never touch the client directly; it’s documented here so advanced setups (custom registries, multi-tenant dispatch, test harnesses) know where the seams are.

Connection model

Implementation lives in platools/transport/client.py. Follows PRD §5.2:

• Outbound WebSocket. The SDK dials ${PLATOS_URL}/ws/sdk - you open no inbound ports. http:// / https:// URLs are rewritten to ws:// / wss:// automatically.
• JWT auth. The PLATOS_SECRET is sent as Authorization: Bearer <secret> on the WebSocket upgrade request.
• Heartbeat every 30s. Fails closed if the platform stops responding.
• Exponential backoff reconnect. Base 1s, cap 60s, up to 10 attempts before the SDK gives up for the current session.
• Registration on reconnect. The full tool list is sent again after every reconnect so the platform’s view of your registry can never go stale.

Basic usage

import asyncio
from my_app.tools import platools  # the Platools() instance

asyncio.run(platools.connect())

connect() runs forever. To shut down gracefully from another task (say, in response to a SIGTERM handler):

import asyncio
import signal
from my_app.tools import platools

async def main() -> None:
    stop = asyncio.Event()

    def _handle_signal(*_args: object) -> None:
        stop.set()

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, _handle_signal)

    connect_task = asyncio.create_task(platools.connect())
    await stop.wait()
    connect_task.cancel()
    try:
        await connect_task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

Direct PlatoolsClient

For tests or tooling that wants to instantiate a client without going through the Platools class:

from platools import PlatoolsClient
from platools.core.registry import ToolRegistry

registry = ToolRegistry()
# ... register tools ...

client = PlatoolsClient(
    url="https://platform.example.com",
    secret="platos_agent_...",
    registry=registry,
)
await client.run_forever()

The constructor validates url and secret upfront - empty strings raise ValueError so a misconfigured deploy fails at startup, not at first-call time.

Wire protocol (at a glance)

The SDK and platform exchange Pydantic-validated JSON messages. The full shapes live in platools/transport/protocol.py.

Direction	Message	Purpose
SDK -> platform	tool_register	Sends ToolSchemaPayload[] after every (re)connect.
SDK -> platform	heartbeat	Every 30s to keep the connection alive.
SDK -> platform	tool_result	Success response for a tool_call, with latency_ms.
SDK -> platform	tool_error	Error response (exception message + traceback).
platform -> SDK	welcome	Handshake ack.
platform -> SDK	heartbeat_ack	Heartbeat ack.
platform -> SDK	tool_call	call_id, tool_name, params - dispatched locally.

Unknown incoming messages are logged and ignored so a platform upgrade that adds new message types can’t crash an old SDK.

Error handling

When a tool raises, the client catches the exception, builds a ToolErrorMessage with the stringified error and traceback.format_exc(), and sends it back over the wire. The session stays up - one bad tool call never kills the connection.

If the WebSocket itself drops (ConnectionClosed, network error), run_forever() logs the failure, backs off exponentially, and reconnects. Reconnects reset the attempt counter after a successful session, so a long-lived SDK that flaps once doesn’t get stuck in “attempt 9” mode forever.

Next steps

• CLI - platools doctor and platools test validate tools before you connect.
• Local mode - platools serve for development without a platform.
• TypeScript Client - same wire protocol, same backoff curve, different language.