bma-executor-sdk (Python)

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==0.4.9

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 asyncio.Event().wait()

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; close codes 1008, 4002, 4003 and 4004 stop the client (4001 and 4005 reconnect).
• 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.