Executor client

The executor client is the hand end: it fetches a short-lived executor token, connects out to BMA (or a local harness), registers with its capabilities, heartbeats, reconnects with resume, and serves the session and tool requests sent to it. Register tool handlers and protocol handlers before start.

Generated from bma-executor-sdk 1.3.0 and its contract (contracts/bma/executor, protocol bma.executor.v1.1). Install, credentials and quick starts are on the bma-executor-sdk overview; the protocol itself is described in Executor Protocol.

Create a client

Validates the options and returns a disconnected client; the long-lived credential (environment key or machine token) is only sent to the token endpoint.

constructor(options: ExecutorClientOptions)

def __init__(
    self,
    *,
    environment_id: str,
    executor: ExecutorInfo,
    capabilities: Capabilities | None = None,
    machine_id: str = "",
    environment_key: str = "",
    base_url: str = DEFAULT_BASE_URL,
    ws_url: str = "",
    token_source: TokenSource | None = None,
    dialer: Dialer | None = None,
    http_client: httpx.AsyncClient | None = None,
    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    heartbeat_interval: float = DEFAULT_HEARTBEAT_INTERVAL,
    heartbeat_load: Callable[[], ExecutorLoad | None] | None = None,
    max_reconnect_attempts: int = DEFAULT_MAX_RECONNECT_ATTEMPTS,
    reconnect_base_delay: float = DEFAULT_RECONNECT_BASE_DELAY,
    reconnect_max_delay: float = _MAX_RECONNECT_DELAY,
    random: Callable[[], float] = _random.random,
    tool_result_retry_attempts: int = DEFAULT_TOOL_RESULT_RETRY_ATTEMPTS,
    tool_result_retry_base_delay: float = DEFAULT_TOOL_RESULT_RETRY_BASE_DELAY,
    on_tool_result_lost: Callable[[ToolResultParams, BaseException], None] | None = None,
    on_log: Callable[[LogEvent], None] | None = None,
) -> None:
    # environment_id may be empty: a machine-scoped executor registers
    # without one (executor.register in the v1.1 contract). environment_id
    # and machine_id are alternatives; machine_id defaults to BMA_MACHINE_ID
    # when environment_id is empty. With neither, the legacy alias
    # POST /api/v1/executors/token serves a machine daemon token.

func NewClient(options Options) (*Client, error)

Client options

Base URL, credential, executor identity, capabilities, reconnect and tool-result retry budgets, and the hook that receives a tool result every retry failed to deliver.

export interface ExecutorClientOptions extends Omit<WebSocketDialerOptions, 'environmentKey' | 'tokenSource'> {
  environmentKey?: string
  tokenSource?: TokenSource
  capabilities?: Capabilities
  /** Replaces the WebSocket dialer, for example listenExecutor().dialer or Harness.inProcessDialer. */
  dialer?: Dialer
  /** false for a connection that cannot be redialed (an accepted local connection). */
  reconnect?: boolean
  /** Overrides the interval returned by executor.register. */
  heartbeatIntervalMs?: number
  heartbeatLoad?: () => ExecutorLoad | undefined
  /**
   * Bounds consecutive failed reconnects; the client then ends and
   * waitClosed() rejects with ReconnectExhaustedError. 0 (the default) or a
   * negative value reconnects until stop().
   */
  maxReconnectAttempts?: number
  /** The first reconnect backoff (default 1 s); it doubles per failed attempt up to reconnectMaxDelayMs, with jitter. */
  reconnectBaseDelayMs?: number
  /** Caps the reconnect backoff before jitter; defaults to 60 s. */
  reconnectMaxDelayMs?: number
  random?: () => number
  /** Total tool.result attempts including the first; 1 disables retries. */
  toolResultRetryAttempts?: number
  toolResultRetryBaseDelayMs?: number
  /**
   * Called when a tool result was not accepted after all retries or the error
   * is not retryable. Persist it and replay it with the same tool_call_id.
   */
  onToolResultLost?: (result: ToolResultParams, error: unknown) => void
  onLog?: Logger
}

# Python takes these options as the keyword arguments of `ExecutorClient(...)` above.

type Options struct {
	// EnvironmentKey is the bearer of the token exchange: an environment key
	// (POST /v1/environments/{EnvironmentID}/token), or a machine daemon token
	// for a machine-scoped executor (POST /v1/machines/{MachineID}/token).
	EnvironmentKey string
	// EnvironmentID is the selfhost environment. EnvironmentID and MachineID
	// are alternatives: setting both is an error. With neither the client
	// registers a machine-scoped executor on the legacy alias
	// POST /api/v1/executors/token (removed in contract 2.0).
	EnvironmentID string
	// MachineID is the machine of a machine-scoped executor, taken from
	// BMA_MACHINE_ID (MachineIDEnv) when empty and EnvironmentID is empty.
	// The id must equal the machine of the credential, otherwise the server
	// answers 404.
	MachineID    string
	Executor     ExecutorInfo
	Capabilities Capabilities
	// Protocol is the WebSocket subprotocol and executor.register protocol.
	// Defaults to ProtocolV11.
	Protocol string

	BaseURL string
	WSURL   string
	// TokenSource replaces the token exchange bootstrap, for example
	// StaticToken for a local harness. When set, EnvironmentKey and BaseURL
	// are not used.
	TokenSource func(ctx context.Context) (ExecutorTokenResponse, error)
	// Dial opens the control connection. Defaults to WebSocketDialer with
	// HTTPClient; see UnixSocketDialer, ListenUnixSocket and InProcessDialer.
	Dial Dialer
	// HTTPClient is used for the token request, artifact uploads and the
	// default WebSocket dial. Defaults to http.DefaultClient.
	HTTPClient *http.Client
	// RequestTimeout bounds one JSON-RPC request. Defaults to 30s.
	RequestTimeout time.Duration
	// HeartbeatInterval overrides the interval returned by executor.register.
	HeartbeatInterval time.Duration
	// HeartbeatLoad is sampled for each heartbeat.
	HeartbeatLoad func() *ExecutorLoad
	// MaxReconnectAttempts bounds consecutive failed reconnects; the client
	// then ends with ErrReconnectExhausted (see Done and Err). Zero (the
	// default) or a negative value reconnects until Stop.
	MaxReconnectAttempts int
	// ReconnectBaseDelay is the first reconnect backoff (default 1s); it
	// doubles per failed attempt up to ReconnectMaxDelay, with jitter.
	ReconnectBaseDelay time.Duration
	// ReconnectMaxDelay caps the reconnect backoff before jitter. Defaults
	// to 60s.
	ReconnectMaxDelay time.Duration
	Random            func() float64
	// ToolResultRetryAttempts is the total number of attempts for tool.result,
	// session.ready and session.ended, including the first. It defaults to 5;
	// set it to 1 to disable retries.
	ToolResultRetryAttempts int
	// ToolResultRetryBaseDelay is the first backoff for retryable failures.
	// It defaults to 250ms and doubles per attempt up to 10s.
	ToolResultRetryBaseDelay time.Duration
	// OnToolResultLost is called by the dispatch path when a tool result could
	// not be accepted after all retries, or the error is not retryable. The
	// adapter must persist the result and replay it with the same
	// tool_call_id; the client never drops a result silently.
	OnToolResultLost func(params ToolResultParams, err error)
	OnLog            func(LogEvent)
}

start

Fetches an executor token, connects, sends executor.register and starts heartbeats. After a failed start the client may be started again.

async start(): Promise<ExecutorRegisterResult>

async def start(self) -> ExecutorRegisterResult

func (c *Client) Start(ctx context.Context) (ExecutorRegisterResult, error)

stop

Sends executor.close, closes the connection and fails pending calls.

stop(reason = 'stopped'): void

async def stop(self, reason: str = "") -> None

func (c *Client) Stop(ctx context.Context, reason string) error

Done / Err / waitClosed

Report when the client stops for good (stop, executor.close, a terminal close code such as 4004 replaced, a rejected credential, or exhausted reconnects). In Go, Done closes and Err explains why; in Python and TypeScript, wait_closed() / waitClosed() return after a stop and raise on any other end, so the process can exit and a supervisor restart it.

async waitClosed(): Promise<void>

async def wait_closed(self) -> None

func (c *Client) Done() <-chan struct

func (c *Client) Err() error

WaitReady / waitReady / wait_ready

Wait until the executor is registered and the registration stayed open for one heartbeat interval. The state turns ready as soon as executor.register returns, which a refused registration also does (a machine-scoped executor that declares neither the process nor the agent capability is closed with 4002 right after), so only this wait separates a usable executor from one about to be refused. It returns the error of a client that ended, and fails on timeout.

async waitReady(timeoutMs?: number): Promise<void>

async def wait_ready(self, timeout: float | None = None) -> None

func (c *Client) WaitReady(ctx context.Context) error

state

The current connection state.

get state(): ConnectionState

@property
def state(self) -> ConnectionState

func (c *Client) State() ConnectionState

executorId / connectionId / protocol

The executor id assigned by executor.register, the current connection id and the negotiated subprotocol.

get executorId(): string

get connectionId(): string

get protocol(): string | undefined

@property
def executor_id(self) -> str

@property
def connection_id(self) -> str

func (c *Client) ExecutorID() string

func (c *Client) ConnectionID() string

activeSessions

The sessions currently assigned to this executor.

get activeSessions(): string[]

@property
def active_sessions(self) -> list[str]

func (c *Client) ActiveSessions() []string

sessionContext / sessionContextById

Read the session a process or session id belongs to, with its metadata, over the machine token of a machine-scoped executor (no environment id). Answers are cached per session; a missing session and a client that is not machine-scoped are distinct errors.

sessionContext(processId: string): Promise<SessionContext>

sessionContextById(sessionId: string): Promise<SessionContext>

async def session_context(self, process_id: str) -> SessionContext

async def session_context_by_id(self, session_id: str) -> SessionContext

func (c *Client) SessionContext(ctx context.Context, processID string) (SessionContext, error)

func (c *Client) SessionContextByID(ctx context.Context, sessionID string) (SessionContext, error)

RefreshModelToken

Asks the platform for a fresh model credential for a running turn (model_token.refresh): call it before the turn.start model_token expires. One attempt, the caller owns retry and backoff; an ended turn answers turn_inactive, a platform that issues this machine no model token answers forbidden or model_token_unavailable.

// Not in the TypeScript SDK 1.3.0.

# Not in the Python SDK 1.3.0.

func (c *Client) RefreshModelToken(ctx context.Context, sessionID, turnID string) (ModelTokenRefreshResult, error)

onSessionAssign

Registers the session.assign handler, which prepares the workspace and MCP servers of a new session.

onSessionAssign(handler: SessionAssignHandler): void

def on_session_assign(self, handler: SessionAssignHandler) -> None

func (c *Client) OnSessionAssign(handler SessionAssignHandler)

onSessionStop

Registers the session.stop handler; stop the session’s processes and MCP servers before answering.

onSessionStop(handler: SessionStopHandler): void

def on_session_stop(self, handler: SessionStopHandler) -> None

func (c *Client) OnSessionStop(handler SessionStopHandler)

onDrain

Registers the executor.drain handler; a draining executor accepts no new sessions.

onDrain(handler: DrainHandler): void

def on_drain(self, handler: DrainHandler) -> None

func (c *Client) OnDrain(handler DrainHandler)

OnToolCallAccept

Go only: a check run before a tool.call is acknowledged; an error answers the request (an *RPCError keeps its code) and the tool does not run.

// Not in the TypeScript SDK 1.3.0.

# Not in the Python SDK 1.3.0.

func (c *Client) OnToolCallAccept(accept func(ToolCallParams) error)

sessionReady

Reports session.ready after workspace and MCP setup complete; only your code knows when that is.

sessionReady(params: SessionReadyParams): Promise<Ack>

async def session_ready(self, params: SessionReadyParams) -> Ack

func (c *Client) SessionReady(ctx context.Context, params SessionReadyParams) (Ack, error)

sessionEnded

Reports session.ended once every process and MCP server of the session has stopped; the session then leaves the active sessions.

async sessionEnded(params: SessionEndedParams): Promise<Ack>

async def session_ended(self, params: SessionEndedParams) -> Ack

func (c *Client) SessionEnded(ctx context.Context, params SessionEndedParams) (Ack, error)

reportToolResult

Sends tool.result once per tool_call_id, retrying executor_busy, timeout and disconnects with backoff across reconnects. Tool handlers call it for you; use it for results you produce yourself.

async reportToolResult(params: ToolResultParams): Promise<Ack>

async def report_tool_result(self, params: ToolResultParams) -> Ack

func (c *Client) ReportToolResult(ctx context.Context, params ToolResultParams) (Ack, error)

UploadArtifact

Go only: uploads an artifact with the current executor token (POST /v1/executors/artifacts, idempotent by session and digest); finish it before a tool.result references the artifact.

// Not in the TypeScript SDK 1.3.0.

# Not in the Python SDK 1.3.0.

func (c *Client) UploadArtifact(ctx context.Context, sessionID, name, mimeType string, content []byte) (ArtifactRef, error)

openMcpChannel

Opens the MCP channel of one environment-owned MCP server from session.assign.mcp_servers; frames are raw MCP JSON-RPC.

async openMcpChannel(sessionId: string, serverKey: string): Promise<FrameChannel>

async def open_mcp_channel(self, session_id: str, server_key: str) -> MCPChannel

func (c *Client) OpenMCPChannel(ctx context.Context, sessionID, serverKey string) (*MCPChannel, error)

updateCapabilities

TypeScript: publishes changed capabilities with executor.capabilities.update. Registering a tool handler after start does this in every language.

updateCapabilities(capabilities: Capabilities): void

# Not in the Python SDK 1.3.0.

// Not in the Go SDK 1.3.0.

onRequest

Serves one harness-to-executor capability request (process.*, terminal.*, skill.install, workspace.backup); see Process, terminal & machine.

onRequest<M extends ExecutorRequestMethod>(method: M, handler: RequestHandler<M>): void

def on_request(self, method: Request[P, R], handler: Callable[[P], Awaitable[R]]) -> None

func (m RequestMethod[P, R]) Handle(client *Client, handler func(ctx context.Context, params P) (R, error))

call / notify

Sends a typed executor-to-harness request or notification (for example session.event, tool.progress); see Protocol methods.

call<M extends ExecutorToHarnessRequestMethod>(method: M, params: MethodParamsMap[M]): Promise<MethodResultMap[M]>

notify<M extends ExecutorToHarnessNotificationMethod>(method: M, params: MethodParamsMap[M]): void

async def call(self, method: Call[P, R], params: P) -> R

async def notify(self, method: Notification[P], params: P) -> None

func (m CallMethod[P, R]) Call(ctx context.Context, client *Client, params P) (R, error)

func (m NotificationMethod[P]) Notify(ctx context.Context, client *Client, params P) error

sessionEvent / sessionStatus / toolProgress

TypeScript shortcuts for session.event, session.status and tool.progress; the other languages use call / notify with the method.

sessionEvent(params: SessionEventParams): Promise<SessionEventResult>

sessionStatus(params: SessionStatusParams): void

toolProgress(params: ToolProgressParams): void

# Not in the Python SDK 1.3.0.

// Not in the Go SDK 1.3.0.

serveAgent

Advertises the agent capability and routes the agent methods to your handler; call it before start. See Agent.

serveAgent(handler: AgentHandler, capability: Omit<AgentCapability, 'enabled'> = {}): void

def serve_agent(self, handler: AgentHandler, capability: AgentCapability | None = None) -> None

func (c *Client) ServeAgent(handler AgentHandler, capability AgentCapability)

staticToken

A token source for a harness reached directly, typically a local harness; the token may be empty when the harness has no authenticate hook.

export function staticToken(wsUrl: string, token = ''): TokenSource

def static_token(ws_url: str, token: str = "") -> TokenSource

func StaticToken(wsURL, token string) func(context.Context) (ExecutorTokenResponse, error)

MCP channel

The executor end of an MCP channel; read and write one raw MCP JSON-RPC frame at a time.

export class FrameChannel {
  constructor(readonly conn: Conn)
  get done(): Promise<void>
  read(signal?: AbortSignal): Promise<string>
  write(frame: string): void
  close(): void
}

class MCPChannel:
    """Executor end of one MCP channel: raw MCP JSON-RPC frames. MCP process"""
    def __init__(self, conn: Connection) -> None: ...
    async def read(self) -> str: ...
    async def write(self, frame: str) -> None: ...
    async def close(self) -> None: ...

func (m *MCPChannel) Read(ctx context.Context) ([]byte, error)

func (m *MCPChannel) Write(ctx context.Context, payload []byte) error

func (m *MCPChannel) Close() error

Local transports

Listen as the executor on a local Unix socket or Windows named pipe, so the next local harness (brain) that dials it drives this hand; the roles stay the same: the executor still registers.

export async function listenExecutor(options: ListenExecutorOptions): Promise<ExecutorListener>

# Not in the Python SDK 1.3.0.

func ListenUnixSocket(path, token string) (Dialer, func() error, error)

func ListenNamedPipe(name, token string) (Dialer, func() error, error)

func UnixSocketDialer(path string) Dialer

func NamedPipeDialer(name string) Dialer

func WebSocketDialer(httpClient *http.Client) Dialer

SDKs

Anthropic SDKs

bma-executor-sdk

Create a client

Client options

start

stop

Done / Err / waitClosed

WaitReady / waitReady / wait_ready

state

executorId / connectionId / protocol

activeSessions

sessionContext / sessionContextById

RefreshModelToken

onSessionAssign

onSessionStop

onDrain

OnToolCallAccept

sessionReady

sessionEnded

reportToolResult

UploadArtifact

openMcpChannel

updateCapabilities

onRequest

call / notify

sessionEvent / sessionStatus / toolProgress

serveAgent

staticToken

MCP channel

Local transports

​Create a client

​Client options

​start

​stop

​Done / Err / waitClosed

​WaitReady / waitReady / wait_ready

​state

​executorId / connectionId / protocol

​activeSessions

​sessionContext / sessionContextById

​RefreshModelToken

​onSessionAssign

​onSessionStop

​onDrain

​OnToolCallAccept

​sessionReady

​sessionEnded

​reportToolResult

​UploadArtifact

​openMcpChannel

​updateCapabilities

​onRequest

​call / notify

​sessionEvent / sessionStatus / toolProgress

​serveAgent

​staticToken

​MCP channel

​Local transports

Create a client

Client options

start

stop

Done / Err / waitClosed

WaitReady / waitReady / wait_ready

state

executorId / connectionId / protocol

activeSessions

sessionContext / sessionContextById

RefreshModelToken

onSessionAssign

onSessionStop

onDrain

OnToolCallAccept

sessionReady

sessionEnded

reportToolResult

UploadArtifact

openMcpChannel

updateCapabilities

onRequest

call / notify

sessionEvent / sessionStatus / toolProgress

serveAgent

staticToken

MCP channel

Local transports