Agent client

Low-level raw CBOR transport for communicating with agentd through a sandbox relay.

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

AgentClient

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.

AgentStream

Returned by stream()

An async iterator of raw agent frames.

stream.id

int Protocol correlation id; pass to send() for follow-up frames

stream.next()

next()

Read the next frame; returns None at EOF

Returns

Awaitable[RawFrame \| None]

stream.close()

close()

Release the stream handle early; safe to call more than once

Returns

Awaitable[None]

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

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

Getting Started

SDKs

CLI

Constants

AgentClient

AgentClient.connect_sandbox()

AgentClient.connect()

AgentClient.socket_path()

client.request()

client.stream()

client.send()

client.ready_bytes()

client.close()

AgentStream

stream.id

stream.next()

stream.close()

Types

RawFrame

​Constants

​AgentClient

​AgentClient.connect_sandbox()

​AgentClient.connect()

​AgentClient.socket_path()

​client.request()

​client.stream()

​client.send()

​client.ready_bytes()

​client.close()

​AgentStream

​stream.id

​stream.next()

​stream.close()

​Types

​RawFrame

Constants

AgentClient

AgentClient.connect_sandbox()

AgentClient.connect()

AgentClient.socket_path()

client.request()

client.stream()

client.send()

client.ready_bytes()

client.close()

AgentStream

stream.id

stream.next()

stream.close()

Types

RawFrame