Agent client - microsandbox

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.

Typical flow

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

Constants

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

Constructors

AgentClient.connect_sandbox()

@classmethod
async def connect_sandbox(cls, name: str, *, timeout: float | None = None) -> AgentClient

Example

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.

Parameters

namestr

Sandbox name, up to 128 UTF-8 bytes.

timeoutfloat | None

Connection timeout in seconds. None uses the default.

Returns

AgentClient

Connected client.

AgentClient.connect()

@classmethod
async def connect(cls, path: str, *, timeout: float | None = None) -> AgentClient

Example

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().

Parameters

pathstr

Path to the agentd relay socket.

timeoutfloat | None

Connection timeout in seconds. None uses the default.

Returns

AgentClient

Connected client.

AgentClient.socket_path()

@staticmethod
def socket_path(name: str) -> str

Example

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.

Parameters

namestr

Sandbox name, up to 128 UTF-8 bytes.

Returns

str

Filesystem path to the relay socket.

Instance methods

client.request()

async def request(self, flags: int, body: bytes) -> RawFrame

Example

frame = await client.request(0, body)
print(frame["id"], frame["flags"])

Send one raw frame and wait for one response frame.

Parameters

flagsint

Frame flag byte, e.g. a combination of FLAG_* constants.

bodybytes

CBOR-encoded protocol message body.

Returns

RawFrame

The response frame.

client.stream()

async def stream(self, flags: int, body: bytes) -> AgentStream

Example

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.

Parameters

flagsint

Frame flag byte; pass FLAG_SESSION_START to open a session.

bodybytes

CBOR-encoded protocol message body.

Returns

AgentStream

Open streaming session.

client.send()

async def send(self, id: int, flags: int, body: bytes) -> None

Example

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().

Parameters

idint

Correlation id of an open session, from stream.id.

flagsint

Frame flag byte.

bodybytes

CBOR-encoded protocol message body.

client.ready_bytes()

def ready_bytes(self) -> bytes

Example

ready = client.ready_bytes()

Return the cached handshake core.ready frame body as CBOR bytes.

Returns

bytes

CBOR-encoded core.ready frame body.

client.close()

async def close(self) -> None

Example

await client.close()

Close the client. Calling it more than once is safe.

Types

RawFrame

Returned by request() · yielded by AgentStream

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

AgentStream

Returned by stream()

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

Example

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

​Constants

​Constructors

​AgentClient.connect_sandbox()

​AgentClient.connect()

​AgentClient.socket_path()

​Instance methods

​client.request()

​client.stream()

​client.send()

​client.ready_bytes()

​client.close()

​Types

​RawFrame

​AgentStream

Constants

Constructors

AgentClient.connect_sandbox()

AgentClient.connect()

AgentClient.socket_path()

Instance methods

client.request()

client.stream()

client.send()

client.ready_bytes()

client.close()

Types

RawFrame

AgentStream