bma-executor-sdk (Python)

Connect your machine (hands) or your own agent (brain) to BMA — the executor, harness and agent ends of the bma.executor protocol. bma-sdk drives and uses agents (REST / façade client); bma-executor-sdk connects hands or brains to BMA.

asyncio executor (ExecutorClient), harness (Harness) and agent ends of the BMA Executor Protocol bma.executor.v1.1, the counterpart of the Go executor module. Protocol types are TypedDicts generated into bma_executor_sdk/generated.py from the contract in contracts/bma/executor, so messages are plain JSON dicts.

pip install bma-executor-sdk==1.0.2

The REST / façade client is the separate package bma-sdk.

Executor quick start

import asyncio, os
from bma_executor_sdk import BashInput, BashResult, ExecutorClient, ToolContext

async def main() -> None:
    client = ExecutorClient(
        base_url=os.environ["BMA_BASE_URL"],  # required: see below
        environment_key=os.environ["BMA_ENVIRONMENT_KEY"],
        environment_id=os.environ["BMA_ENVIRONMENT_ID"],
        executor={"name": "my-executor", "os": "linux", "arch": "arm64", "version": "0.1.0"},
    )

    @client.tools.bash.handler  # one typed slot per built-in tool; only registered tools are advertised
    async def bash(input: BashInput, ctx: ToolContext) -> BashResult:
        proc = await asyncio.create_subprocess_shell(
            input["command"], stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        out, err = await proc.communicate()
        return BashResult(stdout=out.decode(), stderr=err.decode(), exit_code=proc.returncode or 0)

    await client.start()
    await client.wait_closed()  # raises when the client ends for good

asyncio.run(main())

Set base_url to your BMA deployment: the hosted service is in early access and the default DEFAULT_BASE_URL (https://api.bma.beeos.ai) does not resolve yet. It is the deployment origin, the same value as BMA_BASE_URL in bma-sdk (for example https://api.staging.bma.beeos.ai); the client posts the environment key to {base_url}/v1/executors/token and drops a trailing /api/v1. The SDK does not read BMA_BASE_URL itself. A token_source carries its own channel URL.

A complete version that runs bash and stops the process on cancellation is examples/executor_minimal.py; set BMA_HARNESS_URL (ws://… or unix:///path.sock) to connect it to a local harness instead of BMA.

Behavior matches the Go SDK:

• A handler returns a string, content blocks, ToolResult(content, is_error, artifacts) or (bash) BashResult, rendered like the Go Core; raise ToolError(text) reports a tool-level error (is_error: true), any other exception is outcome: "failed". The harness calls tools as await session.tools.bash.call({"command": "ls"}).
• tool.cancel cancels the handler's task and reports outcome: "canceled"; duplicate tool.call ids run once.
• tool.result retries -32009 / -32004 / retryable errors and hands a lost result to on_tool_result_lost.
• Reconnect resumes with resume.executor_id / active_sessions, until stop() by default (backoff 1 s doubling to 60 s, jitter); only close codes 4003 and 4004 stop the client. A positive max_reconnect_attempts bounds the attempts. await client.wait_closed() returns after stop() / executor.close and raises ChannelClosed (4003/4004), CredentialRejectedError (the token endpoint answered 401/403 on reconnect: revoked or invalid credential, not retried) or ReconnectExhaustedError; let it end the process so a supervisor can restart it.
• Call session_ready after setup and session_ended after cleanup. client.on_request(methods.PROCESS_START, handler) serves process.*, terminal.*, skill.install or workspace.backup yourself with the contract's TypedDicts; await client.call(methods.SESSION_EVENT, params) / await client.notify(methods.TOOL_PROGRESS, params) send typed methods (bma_executor_sdk.methods). There is no execution Core in Python.
• Input missing a required member fails the call before the handler runs; a tool.call for a tool without a handler is refused with -32012 before the ack; shell / files are enabled while a handler of the group is registered unless set enabled: False, and a handler registered after start() sends executor.capabilities.update. BashResult beyond INLINE_RESULT_BYTES (or truncated=True) is cut to Go Core's preview. The computer_* tools get slots once the contract defines their input.

Harness quick start

from bma_executor_sdk import Harness

harness = Harness(authenticate=authenticate)  # authenticate(ConnectRequest) -> Principal; omit only in-process or on a private socket
server = await harness.serve("127.0.0.1", 0)   # or await harness.serve_unix(path)

executor = await harness.wait_executor()
session = await executor.assign_session({"session_id": "ses_1"})
await session.ready()
result = await session.call_tool({"tool": "bash", "input": {"command": "ls"}})
mcp = await session.mcp("echo")  # raw MCP JSON-RPC
await session.stop("done")

Semantics match the Go Harness: call_tool is idempotent by tool_call_id; sessions and pending calls survive a reconnect within the lease; lease expiry fails pending calls with ToolOutcomeUnknownError.

Transports

Chosen by the channel URL of the token source:

The executor-listens local socket is not implemented in Python (use the Go or TypeScript SDK); named pipes need the Windows ProactorEventLoop (the asyncio default there) and are tested on a Windows runner in CI. See local connection.

Agent executors

client.serve_agent(handler, capability) serves the agent capability; handler implements open_turn, run_turn and close_session (async; cancellation is task cancellation), and run_turn uses turn.emit, await turn.delta(...), turn.inputs (an asyncio.Queue) and turn.request_interaction. The harness side is Harness(agent=platform) with an AgentPlatform. See agent capability.

Go ↔ Python

Examples

Run from examples after uv sync (or pip install bma-executor-sdk):

Tests

uv sync && uv run pytest -q

tests/test_executor_interop.py runs the Python executor against the Go harness (WebSocket and Unix socket) and the Python harness against the Go executor; it needs Go.