Harness

The harness is the brain end: it accepts executors and drives them exactly as the BMA service does, so an executor cannot tell a local harness from the cloud. Assign a session, wait for it to be ready, call tools (typed session.tools.<name>.call or callTool), cancel them, open MCP channels and stop the session. A disconnected executor keeps its sessions and pending calls for the lease (60 s by default).

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 harness

A harness with defaults applied; authenticate verifies each upgrade request and may pin the executor identity (omit it only on loopback or in-process).

constructor(options: HarnessOptions = {})

export interface HarnessOptions {
  /**
   * Verifies every WebSocket upgrade (control and MCP). Throwing answers 401.
   * Omitted accepts every connection: use that only on loopback.
   */
  authenticate?: (request: IncomingMessage) => Principal | Promise<Principal>
  /** Returned by executor.register. A disconnected executor keeps its sessions and calls for leaseTtlMs. */
  heartbeatIntervalMs?: number
  leaseTtlMs?: number
  requestTimeoutMs?: number
  /** Accepted subprotocols in preference order; defaults to v1.1 then v1. */
  protocols?: string[]
  onExecutor?: (executor: HarnessExecutor) => void
  /** Persists a tool executor's session.event; omitted accepts with a generated id. */
  onSessionEvent?: (executor: HarnessExecutor, params: SessionEventParams) => Promise<SessionEventResult>
  onSessionStatus?: (executor: HarnessExecutor, params: SessionStatusParams) => void
  onToolProgress?: (executor: HarnessExecutor, params: ToolProgressParams) => void
  onLog?: Logger
  /** Receives agent-capability traffic; omitted rejects it as unknown methods. */
  agent?: AgentPlatform
}

def __init__(
    self,
    *,
    authenticate: Authenticate | None = None,
    heartbeat_interval: float = DEFAULT_HARNESS_HEARTBEAT_INTERVAL,
    lease_ttl: float = DEFAULT_HARNESS_LEASE_TTL,
    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    protocols: Sequence[str] = (PROTOCOL_V11, PROTOCOL_V1),
    on_executor: Callable[[Executor], Any] | None = None,
    on_session_event: Callable[[Executor, SessionEventParams], Awaitable[SessionEventResult]] | None = None,
    on_session_status: Callable[[Executor, SessionStatusParams], None] | None = None,
    on_tool_progress: Callable[[Executor, ToolProgressParams], None] | None = None,
    on_log: Callable[[LogEvent], None] | None = None,
    agent: AgentPlatform | None = None,
) -> None

func NewHarness(opts HarnessOptions) *Harness

type HarnessOptions struct {
	// Authenticate verifies every WebSocket upgrade (control and MCP). nil
	// accepts every connection: use that only on a loopback listener.
	Authenticate func(r *http.Request) (Principal, error)
	// HeartbeatInterval and LeaseTTL are returned by executor.register. A
	// disconnected executor keeps its sessions and pending tool calls for
	// LeaseTTL; then they end and pending calls fail with
	// ErrToolOutcomeUnknown.
	HeartbeatInterval time.Duration
	LeaseTTL          time.Duration
	// RequestTimeout bounds harness-to-executor requests. Defaults to 30s.
	RequestTimeout time.Duration
	// Protocols are the accepted subprotocols in preference order. Defaults
	// to ProtocolV11 then ProtocolV1.
	Protocols []string

	// OnExecutor runs after each successful executor.register.
	OnExecutor func(*Executor)
	// OnSessionEvent persists session.event. nil accepts with a generated id.
	// The harness answers a retried JSON-RPC id with the first result.
	OnSessionEvent  func(ctx context.Context, executor *Executor, params SessionEventParams) (SessionEventResult, error)
	OnSessionStatus func(executor *Executor, params SessionStatusParams)
	OnToolProgress  func(executor *Executor, params ToolProgressParams)
	// OnProcessStatus receives the advisory process.status notification.
	OnProcessStatus func(executor *Executor, params ProcessStatusParams)
	OnLog           func(LogEvent)
	// Agent receives agent-capability traffic: agent session events,
	// previews and interactions. nil rejects them as unknown methods.
	Agent AgentPlatform
}

Serve executors

Accept executor connections: on a TCP port, a local Unix socket or Windows named pipe, an existing HTTP server, or in-process.

async listen(port = 0, host = '127.0.0.1'): Promise<{ url: string; close(): Promise<void> }>

async listenLocal(path: string): Promise<{ url: string; close(): Promise<void> }>

attach(server: Server): void

async handleUpgrade(request: IncomingMessage, socket: Duplex, head: Buffer): Promise<void>

inProcessDialer(principal: Principal = {}): Dialer

async def serve(self, host: str = "127.0.0.1", port: int = 0) -> Any

async def serve_unix(self, path: str) -> Any

async def serve_npipe(self, name: str) -> Any

def inprocess_dialer(self) -> Dialer

func (h *Harness) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (h *Harness) ServeConn(transport Conn, principal Principal)

func InProcessDialer(h *Harness, principal Principal) Dialer

dialExecutor

Connects this harness to an executor that listens on a local socket or named pipe, so a local brain drives the hand already running on the machine.

async dialExecutor(url: string, options: { token?: string } = {}): Promise<void>

async dialMcp(executor: HarnessExecutor, sessionId: string, serverKey: string): Promise<void>

# Not in the Python SDK 1.3.0.

func (h *Harness) DialExecutor(ctx context.Context, path, token string) error

waitExecutor

Returns a connected executor, waiting for one to register.

async waitExecutor(signal?: AbortSignal): Promise<HarnessExecutor>

async def wait_executor(self) -> Executor

func (h *Harness) WaitExecutor(ctx context.Context) (*Executor, error)

executors

The executors currently registered (connected or within their reconnect lease).

executors(): HarnessExecutor[]

def executors(self) -> list[Executor]

func (h *Harness) Executors() []*Executor

Executor identity and state

The id, executor info, negotiated protocol, advertised capabilities, connection state and bound sessions of one executor.

get connected(): boolean

sessions(): HarnessSession[]

@property
def connected(self) -> bool

def sessions(self) -> list[Session]

func (e *Executor) ID() string

func (e *Executor) Info() ExecutorInfo

func (e *Executor) Protocol() string

func (e *Executor) Capabilities(out any) error

func (e *Executor) Sessions() []*Session

assignSession

Binds a session to the executor and sends session.assign; the work id and lease token are generated when empty. The session is bound before the request is sent, so a racing session.ready is accepted.

async assignSession(
  params: Omit<SessionAssignParams, 'work_id' | 'environment_id' | 'lease'> & Partial<SessionAssignParams>,
): Promise<HarnessSession>

async def assign_session(self, params: SessionAssignParams) -> Session

func (e *Executor) AssignSession(ctx context.Context, params SessionAssignParams) (*Session, error)

drain

Asks the executor to stop accepting sessions; existing sessions run to completion.

drain(reason: string): Promise<ExecutorDrainResult>

async def drain(self, reason: str = "") -> ExecutorDrainResult

func (e *Executor) Drain(ctx context.Context, reason string) (ExecutorDrainResult, error)

call

Sends a harness-to-executor capability request (process.*, terminal.*, skill.install, …) and decodes its result; a JSON-RPC error answer is an RPC error.

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

# Not in the Python SDK 1.3.0.

func (e *Executor) Call(ctx context.Context, method string, params, out any) error

ready

Waits for the executor’s session.ready.

async ready(signal?: AbortSignal): Promise<void>

async def ready(self) -> None

func (s *Session) Ready(ctx context.Context) error

callTool

Sends tool.call and waits for its tool.result. Idempotent by tool_call_id (generated when empty); canceling the wait does not abort the tool, use cancelTool for that.

async callTool<N extends ToolName>(tool: N, input: ToolCallInputMap[N], options: CallToolOptions = {}): Promise<ToolResultParams>

async def call_tool(
    self,
    tool: ToolName | str,
    input: Mapping[str, Any],
    *,
    tool_call_id: str = "",
    deadline_ms: int | None = None,
    permission: ToolPermission | None = None,
) -> ToolResultParams

func (s *Session) CallTool(ctx context.Context, params ToolCallParams) (ToolResultParams, error)

cancelTool

Sends tool.cancel; the waiting call still receives the executor’s tool.result (outcome canceled when the abort won).

async cancelTool(toolCallId: string, reason: string): Promise<boolean>

async def cancel_tool(self, tool_call_id: str, reason: str = "") -> bool

func (s *Session) CancelTool(ctx context.Context, toolCallID, reason string) (bool, error)

stop

Sends session.stop; the executor terminates the session’s processes and MCP servers before answering.

async stop(reason: string): Promise<void>

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

func (s *Session) Stop(ctx context.Context, reason string) error

Session end

Resolves when the session ends, with the end reason.

get done(): Promise<void>

@property
def ended(self) -> bool

async def wait_ended(self) -> str

func (s *Session) Done() <-chan struct

func (s *Session) EndReason() string

func (s *Session) ID() string

func (s *Session) WorkID() string

func (s *Session) Executor() *Executor

mcp

Waits for the MCP channel of one session.assign MCP server and returns it; frames are raw MCP JSON-RPC and the harness is the MCP client. A newer channel for the same key replaces the old one.

async mcp(serverKey: string, signal?: AbortSignal): Promise<HarnessMcpChannel>

async def mcp(self, server_key: str) -> HarnessMCPChannel

func (s *Session) MCP(ctx context.Context, serverKey string) (*HarnessMCPChannel, error)

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

func (m *HarnessMCPChannel) Write(ctx context.Context, frame []byte) error

SDKs

Anthropic SDKs

bma-executor-sdk

Create a harness

Serve executors

dialExecutor

waitExecutor

executors

Executor identity and state

assignSession

drain

call

ready

callTool

cancelTool

stop

Session end

mcp

​Create a harness

​Serve executors

​dialExecutor

​waitExecutor

​executors

​Executor identity and state

​assignSession

​drain

​call

​ready

​callTool

​cancelTool

​stop

​Session end

​mcp

Create a harness

Serve executors

dialExecutor

waitExecutor

executors

Executor identity and state

assignSession

drain

call

ready

callTool

cancelTool

stop

Session end

mcp