docs/sdk/python/agent-client.mdx
Low-level raw CBOR transport for communicating with agentd through a sandbox relay.
| Name | Value | Description |
|---|---|---|
FLAG_TERMINAL | 0b0000_0001 | Last frame for a correlation id |
FLAG_SESSION_START | 0b0000_0010 | First frame of a streaming session |
FLAG_SHUTDOWN | 0b0000_0100 | Shutdown frame |
<Tooltip tip="Connecting an agent client by socket path is local-only; on microsandbox cloud the SDK uses the agent WebSocket route internally."><span className="msb-badge-local">Local-only <Icon icon="circle-info" size={11} /></span></Tooltip>
@classmethod
async def connect_sandbox(cls, name: str, *, timeout: float | None = None) -> AgentClient
client = await AgentClient.connect_sandbox("dev", timeout=5.0)
Connect to a running sandbox by name. Sandbox names are limited to 128 UTF-8 bytes.
<p className="msb-label">Parameters</p> <div className="msb-params"> <div className="msb-param"> <div className="msb-param-key"><code>name</code><span className="msb-type">str</span></div> <div className="msb-param-desc">Sandbox name, up to 128 UTF-8 bytes.</div> </div> <div className="msb-param"> <div className="msb-param-key"><code>timeout</code><span className="msb-type">float | None</span></div> <div className="msb-param-desc">Connection timeout in seconds. <code>None</code> uses the default.</div> </div> </div> <p className="msb-label">Returns</p> <div className="msb-params"> <div className="msb-param"> <div className="msb-param-key"><span className="msb-type">AgentClient</span></div> <div className="msb-param-desc">Connected client.</div> </div> </div><Tooltip tip="Connecting an agent client by socket path is local-only; on microsandbox cloud the SDK uses the agent WebSocket route internally."><span className="msb-badge-local">Local-only <Icon icon="circle-info" size={11} /></span></Tooltip>
@classmethod
async def connect(cls, path: str, *, timeout: float | None = None) -> AgentClient
path = AgentClient.socket_path("dev")
client = await AgentClient.connect(path)
Connect to an agent relay socket by path. Use this when you already know the socket path, for example one returned by socket_path().
<Tooltip tip="Connecting an agent client by socket path is local-only; on microsandbox cloud the SDK uses the agent WebSocket route internally."><span className="msb-badge-local">Local-only <Icon icon="circle-info" size={11} /></span></Tooltip>
@staticmethod
def socket_path(name: str) -> str
path = AgentClient.socket_path("dev")
Resolve a sandbox's agentd relay socket path without connecting. Returns the same path connect_sandbox() would dial, so you can talk to agentd over a raw byte transport (for example a transparent relay that splices bytes to and from the socket) instead of this frame client. The sandbox need not be running. Sandbox names are limited to 128 UTF-8 bytes.
async def request(self, flags: int, body: bytes) -> RawFrame
frame = await client.request(0, body)
print(frame["id"], frame["flags"])
Send one raw frame and wait for one response frame.
<p className="msb-label">Parameters</p> <div className="msb-params"> <div className="msb-param"> <div className="msb-param-key"><code>flags</code><span className="msb-type">int</span></div> <div className="msb-param-desc">Frame flag byte, e.g. a combination of <code>FLAG_*</code> constants.</div> </div> <div className="msb-param"> <div className="msb-param-key"><code>body</code><span className="msb-type">bytes</span></div> <div className="msb-param-desc">CBOR-encoded protocol message body.</div> </div> </div> <p className="msb-label">Returns</p> <div className="msb-params"> <div className="msb-param"> <div className="msb-param-key"><a className="msb-type" href="#rawframe">RawFrame</a></div> <div className="msb-param-desc">The response frame.</div> </div> </div>async def stream(self, flags: int, body: bytes) -> AgentStream
from microsandbox import FLAG_SESSION_START, FLAG_TERMINAL
stream = await client.stream(FLAG_SESSION_START, body)
async for frame in stream:
if frame["flags"] & FLAG_TERMINAL:
break
Open a raw streaming session. The returned AgentStream carries the protocol correlation id and is also an async iterator of raw frames.
async def send(self, id: int, flags: int, body: bytes) -> None
stream = await client.stream(FLAG_SESSION_START, body)
await client.send(stream.id, 0, follow_up_body)
Send a follow-up frame on an existing correlation id. Use the id from the AgentStream returned by stream().
def ready_bytes(self) -> bytes
ready = client.ready_bytes()
Return the cached handshake core.ready frame body as CBOR bytes.
async def close(self) -> None
await client.close()
Close the client. Calling it more than once is safe.
An async iterator of raw agent frames.
int
Protocol correlation id; pass to send() for follow-up frames
next()
Read the next frame; returns None at EOF
Awaitable[RawFrame \| None]
close()
Release the stream handle early; safe to call more than once
<p className="msb-label">Returns</p>Awaitable[None]
from microsandbox import FLAG_SESSION_START, FLAG_TERMINAL
async with await client.stream(FLAG_SESSION_START, body) as stream:
async for frame in stream:
if frame["flags"] & FLAG_TERMINAL:
break
A raw protocol frame with a CBOR-encoded body.
class RawFrame(TypedDict):
id: int
flags: int
body: bytes
| Field | Type | Description |
|---|---|---|
| id | int | Protocol correlation id |
| flags | int | Frame flag byte (combination of FLAG_* constants) |
| body | bytes | CBOR-encoded protocol message body |