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.5

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

Executor quick start

import asyncio, os
from bma_executor_sdk import ExecutorClient

async def main() -> None:
    client = ExecutorClient(
        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"},
    )

    async def on_tool_call(call):
        return {"session_id": call["session_id"], "tool_call_id": call["tool_call_id"],
                "outcome": "succeeded", "is_error": False, "content": [{"type": "text", "text": "ok"}]}

    client.on_tool_call(on_tool_call)
    await client.start()
    await asyncio.Event().wait()

asyncio.run(main())

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:

• 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. on_request(method, handler) serves process.*, terminal.* or skill.install yourself; there is no execution Core in Python.

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 and Windows named pipes are not implemented in Python (use the Go or TypeScript SDK). 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.