py-opencode-wrapper 0.2.1

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)
instructions	Instruction file paths / glob patterns to inject
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.

User config isolation

By default, RunConfig.inherit_user_config=False makes each child opencode process see a sanitized copy of the host's global OpenCode config. The wrapper keeps only provider-selection keys ($schema, provider, disabled_providers, enabled_providers) and drops capability/configuration keys such as mcp, agent, command, tools, plugin, skills, instructions, permission, and model.

This keeps benchmark and orchestration runs reproducible while still allowing provider configuration and opencode auth credentials to work. Project-level config discovered from the workspace is not suppressed.

Set inherit_user_config=True to restore the legacy behavior of inheriting the host OpenCode config as-is. For reproducible runs, pass model, permission, mcp, tools, and instructions explicitly through RunConfig.

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.