opencode

opencode

OpenCodeAgent -- wraps the opencode coding agent via its headless HTTP server.

Spawns opencode serve (https://opencode.ai) and drives a session over its HTTP API, configured to use OpenJarvis's local engine through an OpenAI-compatible provider. This keeps coding-agent work local-first: opencode handles the agentic loop / tools, OpenJarvis supplies the model.

opencode is an external binary (install: npm i -g opencode-ai or brew install anomalyco/tap/opencode). It is not bundled; :meth:run raises a clear error if it is not on PATH.

Classes

OpenCodeAgent

OpenCodeAgent(engine: InferenceEngine, model: str, *, bus: Optional[EventBus] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None, workspace: str = '', agent: str = 'build', provider_id: str = 'openjarvis', provider_base_url: str = '', model_id: str = '', api_key: str = '', permission: Optional[Any] = None, hostname: str = '127.0.0.1', port: int = 0, server_password: str = '', timeout: int = 600, opencode_bin: str = '')

Bases: BaseAgent

Agent that delegates coding tasks to a local opencode server.

The engine is used to wire opencode at an OpenAI-compatible provider so inference runs on OpenJarvis's selected local model. agent selects opencode's built-in agent: build (full access) or plan (read-only).

Source code in src/openjarvis/agents/opencode.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    bus: Optional[EventBus] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    workspace: str = "",
    agent: str = "build",
    provider_id: str = "openjarvis",
    provider_base_url: str = "",
    model_id: str = "",
    api_key: str = "",
    permission: Optional[Any] = None,
    hostname: str = "127.0.0.1",
    port: int = 0,
    server_password: str = "",
    timeout: int = 600,
    opencode_bin: str = "",
) -> None:
    super().__init__(
        engine,
        model,
        bus=bus,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    self._workspace = workspace or os.getcwd()
    self._agent = agent
    self._provider_id = provider_id
    self._provider_base_url = provider_base_url or _derive_openai_base_url(engine)
    self._model_id = model_id or model
    self._api_key = api_key
    self._permission = permission
    self._hostname = hostname
    self._port = port
    self._server_password = server_password or os.environ.get(
        "OPENCODE_SERVER_PASSWORD", ""
    )
    self._timeout = timeout
    self._opencode_bin = opencode_bin or shutil.which("opencode") or "opencode"
    self._proc: Optional[subprocess.Popen] = None
    self._base: str = ""
    self._config_dir: Optional[str] = None

Functions

close

close() -> None

Dispose the opencode session/server and terminate the process.

Source code in src/openjarvis/agents/opencode.py

def close(self) -> None:
    """Dispose the opencode session/server and terminate the process."""
    if self._base:
        try:
            with self._client() as c:
                c.post("/global/dispose")
        except Exception:
            pass
    if self._proc and self._proc.poll() is None:
        self._proc.terminate()
        try:
            self._proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            self._proc.kill()
    self._proc = None
    self._base = ""
    if self._config_dir:
        shutil.rmtree(self._config_dir, ignore_errors=True)
        self._config_dir = None

run

run(input: str, context: Optional[AgentContext] = None, **kwargs: Any) -> AgentResult

Run a coding task through opencode and return the assistant result.

Source code in src/openjarvis/agents/opencode.py

def run(
    self,
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult:
    """Run a coding task through opencode and return the assistant result."""
    self._emit_turn_start(input)

    # Resolve which opencode provider/model to address. Fail clearly rather
    # than letting opencode 500 on an unregistered provider.
    if self._provider_base_url:
        model_spec = {"providerID": self._provider_id, "modelID": self._model_id}
    elif "/" in self._model_id:
        prov, _, mid = self._model_id.partition("/")
        model_spec = {"providerID": prov, "modelID": mid}
    else:
        self._emit_turn_end(turns=1, error=True)
        return AgentResult(
            content=(
                f"OpenCodeAgent could not determine an opencode provider for "
                f"model {self._model_id!r}: no OpenAI-compatible base URL could "
                f"be derived from the engine. Pass provider_base_url=..., or use "
                f"a 'provider/model' that opencode already knows."
            ),
            turns=1,
            metadata={"error": True},
        )

    try:
        self._ensure_server()
    except RuntimeError as exc:
        self._emit_turn_end(turns=1, error=True)
        return AgentResult(
            content=str(exc), turns=1, metadata={"error": True}
        )

    data: dict = {}
    turn_parts: List[dict] = []
    try:
        with self._client() as c:
            ses = c.post("/session", json={"title": input[:80]})
            ses.raise_for_status()
            session_id = ses.json()["id"]

            body: dict = {
                "agent": self._agent,
                "model": model_spec,
                "parts": [{"type": "text", "text": input}],
            }

            resp = c.post(f"/session/{session_id}/message", json=body)
            resp.raise_for_status()
            data = resp.json()

            # The prompt POST returns only the final assistant message;
            # tool executions live in intermediate messages of the turn, so
            # pull the whole session to recover them (verified against a
            # live opencode session). Falls back to the final message.
            turn_parts = list(data.get("parts", []))
            try:
                msgs = c.get(f"/session/{session_id}/message").json()
                if isinstance(msgs, list):
                    turn_parts = [
                        part
                        for mm in msgs
                        if isinstance(mm, dict)
                        for part in mm.get("parts", [])
                    ]
            except Exception as get_exc:
                logger.debug("opencode message fetch failed: %s", get_exc)
    except Exception as exc:
        logger.error("opencode run failed: %s", exc, exc_info=True)
        self._emit_turn_end(turns=1, error=True)
        return AgentResult(
            content=f"opencode agent failed: {exc}",
            turns=1,
            metadata={"error": True},
        )

    info = data.get("info", {}) if isinstance(data, dict) else {}
    content = _extract_text(data.get("parts", []))
    tool_results = _extract_tool_results(turn_parts)

    self._emit_turn_end(turns=1)
    return AgentResult(
        content=content,
        tool_results=tool_results,
        turns=1,
        metadata={
            "finish": info.get("finish"),
            "tokens": info.get("tokens"),
            "provider_id": info.get("providerID", self._provider_id),
            "model_id": info.get("modelID", self._model_id),
            "session_id": info.get("sessionID", ""),
            "agent": self._agent,
        },
    )

Functions

is_opencode_available

is_opencode_available() -> bool

Return True if the opencode binary is on PATH.

Source code in src/openjarvis/agents/opencode.py

def is_opencode_available() -> bool:
    """Return True if the ``opencode`` binary is on PATH."""
    return shutil.which("opencode") is not None