docs/sdk/python/agent-client.mdx
AgentClient is the low-level raw transport for talking to agentd through a running sandbox's relay socket. For most applications, use Sandbox, exec, and fs instead. Reach for this API when you are building protocol-level tools or higher-level SDK helpers.
All request and response bodies are raw CBOR bytes. The SDK handles framing and correlation ids, but it does not encode or decode the CBOR message body for you. The raw body is the full CBOR-encoded protocol Message body (v, t, and p), not just the inner payload.
from microsandbox import AgentClient
client = await AgentClient.connect_sandbox("dev") # 1. connect
ready = client.ready_bytes() # 2. inspect handshake
frame = await client.request(0, body) # 3. raw request/response
await client.close() # 4. close
| 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 |
@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>@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().
@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.
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 |
An open raw agent stream. Carries the protocol correlation id and is both an async context manager and an async iterator of RawFrame. Iteration stops once a frame with FLAG_TERMINAL set is delivered or the stream reaches EOF.
| Property / Method | Type | Description |
|---|---|---|
| id | int | Protocol correlation id; pass to send() for follow-up frames |
| next() | Awaitable[RawFrame | None] | Read the next frame; returns None at EOF |
| close() | Awaitable[None] | Release the stream handle early; safe to call more than once |
async with | AgentStream | Async context manager; closes the stream on exit |
async for | RawFrame | Async iterator over frames until terminal or EOF |
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