py-opencode-wrapper 0.1.4

Async Python wrapper for OpenCode CLI (opencode run --format json)

oc-py-harness

Python async wrapper around the OpenCode CLI (opencode run --format json). Intended as a subprocess-based executor for multi-agent workflow orchestration.

Requirements

• Python 3.10+
• opencode on PATH (or pass an absolute path to the binary)

Install (local tree)

pip install -e ".[dev]"

Usage

One-shot run with aggregated result

import asyncio
from pathlib import Path

from opencode_wrapper import AsyncOpenCodeClient, RunConfig

async def main():
    client = AsyncOpenCodeClient("opencode")
    cfg = RunConfig(
        model="anthropic/claude-sonnet-4-5",
        agent="plan",
        permission={"bash": "deny", "edit": "deny"},
        mcp={
            "demo": {
                "type": "local",
                "command": ["npx", "-y", "@modelcontextprotocol/server-everything"],
                "enabled": True,
            }
        },
    )
    result = await client.async_run(
        "Summarize the README in one sentence.",
        Path("/path/to/repo"),
        run_cfg=cfg,
        timeout_s=600,
    )
    print(result.exit_code, result.final_text)

asyncio.run(main())

Stream structured JSON events

async def stream_example():
    client = AsyncOpenCodeClient()
    cfg = RunConfig(permission={"*": "allow"})
    async for event in client.async_stream("List top-level files.", workspace=".", run_cfg=cfg):
        print(event)

Parallel agents (asyncio.gather)

async def multi():
    # startup_concurrency=1 serialises SQLite initialisation to avoid a known
    # WAL-pragma race in opencode when many instances start simultaneously.
    # startup_delay_s controls how long each slot is held before the next
    # process is allowed to start (default 0.3 s).
    client = AsyncOpenCodeClient(startup_concurrency=1, startup_delay_s=0.3)
    ws = Path("/path/to/monorepo")
    results = await asyncio.gather(*[
        client.async_run(
            f"Explain services/{svc}.",
            ws / "services" / svc,
            run_cfg=RunConfig(agent="explore"),
            timeout_s=600,
            # max_retries=2 (default): retry automatically if opencode crashes
            # during SQLite startup before giving up.
        )
        for svc in ["api", "worker", "gateway"]
    ])
    return results

Configuration injection

Per-call JSON is merged and passed as OPENCODE_CONFIG_CONTENT (see OpenCode config). Use RunConfig fields:

Field	Purpose
permission	permission map (allow / deny, patterns)
mcp	MCP server definitions
tools	Enable/disable tools (including MCP globs)
config_overrides	Any extra top-level config keys to deep-merge

Optional env tuning: disable_autoupdate=True sets OPENCODE_DISABLE_AUTOUPDATE=1. Note: ask is intentionally rejected in subprocess mode (no interactive terminal); use allow or deny.

CLI arguments

RunConfig maps to flags such as --agent, -m, -f, --attach, --title, etc. Prompt text is appended as the final opencode run message argument.

Tests

Unit tests (no real OpenCode / no API calls):

pytest -q -m "not integration"

Integration tests (real opencode run, needs working provider auth — slow, may incur API usage):

pytest -m integration -q tests/test_integration_opencode.py

Multi-agent weather workflow (10 parallel city lookups + 1 summary — 11 API calls, not run by default):

OPENCODE_MULTI_AGENT_WEATHER=1 pytest -m integration -v tests/test_integration_multi_agent_weather.py

Optional: OPENCODE_WEATHER_SEQUENTIAL=1 runs the 10 city calls one-by-one (easier on rate limits).
Per-stage timeouts: OPENCODE_WEATHER_PER_CITY_TIMEOUT_S, OPENCODE_WEATHER_SUMMARY_TIMEOUT_S (default: same as OPENCODE_INTEGRATION_TIMEOUT_S).

Env	Meaning
OPENCODE_BINARY	Absolute path to opencode if not on PATH
OPENCODE_INTEGRATION=0	Skip integration tests
OPENCODE_INTEGRATION_TIMEOUT_S	Per-test timeout seconds (default 300)
OPENCODE_MULTI_AGENT_WEATHER=1	Enable 11-call weather integration test
OPENCODE_ENABLE_EXA	Passed through / defaulted to 1 in that test for web search tools

Default pytest -q runs all tests; use -m "not integration" in CI without OpenCode.

Concurrency notes

When running many tasks with asyncio.gather, three protections are enabled by default:

Startup serialisation — startup_concurrency=1 and startup_delay_s=0.3 limit how many processes enter SQLite startup at once, reducing WAL-initialisation race crashes.

DB isolation — isolate_db=True gives each run a private XDG_DATA_HOME, so concurrent runs do not contend on the same opencode.db during tool execution.

Automatic retry — async_run(max_retries=2, retry_delay_s=1.0) retries known SQLite-startup crashes with short backoff. Non-SQLite failures still fail fast.

Set startup_concurrency=0, isolate_db=False, and max_retries=0 to opt out.

Notes

• Event shapes from --format json may change between OpenCode versions; unknown fields are preserved in each parsed dict.
• For fully non-interactive automation, prefer explicit permission (allow/deny) over relying on interactive ask prompts.