feat(claude): expose the headless stream-json transport, with real streaming
`ClaudeCodeOptions.transport="stream_json"` runs the agent on `claude -p` with stream-json on both pipes instead of driving the TUI through a pseudo-tty. Opt-in: the default stays `pty`, so nothing moves until a config asks for it. The reason to ask for it is that the PTY path pays a multi-second readiness wait before every spawn and can silently lose a prompt to a swallowed paste; the headless path writes to a pipe and gets a native `result` record back. `--remote-control` and friends are dropped by the transport when set, since they'd take claude out of print mode. `include_partial_messages` then gives clients genuine token-level SSE. Blocks are emitted from the `StreamEvent`s rather than forwarded raw: the payload is already Anthropic-shaped but arrives as a dict, so it would need validating against the SDK union anyway, and rebuilding it through our own builders drops an unrecognized block or delta type the same way `_emit_block` already does instead of raising mid-stream on a claude release that adds one. When streaming, the whole-block records are kept for TurnCapture but not re-emitted — that would duplicate every block on the wire. Verified end-to-end against a live claude: both paths accumulate to the identical Message (same blocks, same stop_reason, same usage), the envelope stays 1:1, and block indices stay collision-free across a turn that spans several API requests.
This commit is contained in:
@@ -1,8 +1,13 @@
|
|||||||
"""Claude Code agent definition.
|
"""Claude Code agent definition.
|
||||||
|
|
||||||
Notice the absence of a ``streaming`` field — claude-code does not emit
|
There is no ``streaming`` field, but there is
|
||||||
token-level deltas, and that fact is encoded in the type, not in a
|
:attr:`ClaudeCodeOptions.include_partial_messages`, and the difference
|
||||||
runtime branch.
|
matters. Token-level deltas are not a property of the agent, they are a
|
||||||
|
property of the transport: the PTY transport reads a JSONL that only
|
||||||
|
ever contains finished blocks, so it cannot stream no matter what the
|
||||||
|
agent asks for, while ``transport="stream_json"`` can. The flag lives
|
||||||
|
next to the transport that grants it rather than pretending to be a
|
||||||
|
free-standing capability.
|
||||||
|
|
||||||
``BaseAgent.system_prompt`` maps onto the claude CLI's
|
``BaseAgent.system_prompt`` maps onto the claude CLI's
|
||||||
``--system-prompt`` — i.e. it really *is* the agent's system prompt,
|
``--system-prompt`` — i.e. it really *is* the agent's system prompt,
|
||||||
@@ -57,6 +62,9 @@ HistoryInjectionMode = Literal["native_jsonl", "concat_message"]
|
|||||||
cycle on the user-facing path (configs may be loaded without
|
cycle on the user-facing path (configs may be loaded without
|
||||||
``claude-code-api`` installed, e.g. ``--extra prod`` minus claude)."""
|
``claude-code-api`` installed, e.g. ``--extra prod`` minus claude)."""
|
||||||
|
|
||||||
|
Transport = Literal["pty", "stream_json"]
|
||||||
|
"""Mirrors ``claude_code_api.Transport``, for the same reason."""
|
||||||
|
|
||||||
|
|
||||||
class ClaudeCodeOptions(BaseModel):
|
class ClaudeCodeOptions(BaseModel):
|
||||||
"""Per-agent passthrough for ``claude_code_api.BackendOptions``.
|
"""Per-agent passthrough for ``claude_code_api.BackendOptions``.
|
||||||
@@ -85,6 +93,32 @@ class ClaudeCodeOptions(BaseModel):
|
|||||||
|
|
||||||
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
|
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
|
||||||
|
|
||||||
|
transport: Transport = "pty"
|
||||||
|
"""Which claude process backs this agent.
|
||||||
|
|
||||||
|
``stream_json`` runs ``claude -p`` with stream-json on stdin and
|
||||||
|
stdout: prompts go into a pipe, events come back as they happen, and
|
||||||
|
a native ``result`` record closes each turn. ``pty`` (the default,
|
||||||
|
kept for backwards compatibility) drives the interactive TUI and
|
||||||
|
tails the session JSONL, which costs a multi-second readiness wait
|
||||||
|
per spawn — the thing ``startup_delay`` bounds — and brings the
|
||||||
|
swallowed-paste failure mode with it.
|
||||||
|
|
||||||
|
Switching also makes the PTY-only knobs below inert
|
||||||
|
(``startup_delay`` / ``file_wait_timeout`` /
|
||||||
|
``turn_duration_timeout`` / ``wait_for_turn_duration``) and drops
|
||||||
|
interactive-only ``extra_args`` such as ``--remote-control``, which
|
||||||
|
would take claude out of print mode. Everything else — model,
|
||||||
|
prompts, tools, MCPs, history seeding — behaves identically."""
|
||||||
|
|
||||||
|
include_partial_messages: bool = False
|
||||||
|
"""Stream token-level deltas to the client instead of one delta per
|
||||||
|
finished block. Requires ``transport="stream_json"``; inert on the
|
||||||
|
PTY transport, whose JSONL contains only finished blocks. Off by
|
||||||
|
default because it multiplies the SSE event count — worth it for
|
||||||
|
interactive chat, pointless for a client that waits for the whole
|
||||||
|
message anyway."""
|
||||||
|
|
||||||
append_system_prompt: str | None = None
|
append_system_prompt: str | None = None
|
||||||
"""Maps to claude CLI's ``--append-system-prompt``. Opting in
|
"""Maps to claude CLI's ``--append-system-prompt``. Opting in
|
||||||
re-attaches claude-code's full built-in prompt (persona, planning
|
re-attaches claude-code's full built-in prompt (persona, planning
|
||||||
|
|||||||
@@ -13,11 +13,20 @@ Per :meth:`complete` we:
|
|||||||
* hand the full Anthropic-style ``messages`` list to
|
* hand the full Anthropic-style ``messages`` list to
|
||||||
``ClaudeCodeBackend.complete`` — it does its own fingerprint-based
|
``ClaudeCodeBackend.complete`` — it does its own fingerprint-based
|
||||||
session lookup, so we never need to track sessions ourselves;
|
session lookup, so we never need to track sessions ourselves;
|
||||||
* re-emit each ``AssistantMessage`` as ``content_block_start`` +
|
* turn its events into one ``message_start`` … ``message_stop``
|
||||||
one delta + ``content_block_stop`` per content block, with
|
envelope per ``complete`` call, with content-block indices increasing
|
||||||
monotonically increasing indices spanning the entire turn (one
|
monotonically across the whole turn;
|
||||||
``message_start`` … ``message_stop`` envelope per ``complete`` call);
|
* close the envelope on the ``ResultMessage``.
|
||||||
* close the envelope on the synthesized ``ResultMessage``.
|
|
||||||
|
How the blocks inside that envelope are produced depends on the agent's
|
||||||
|
transport. By default each ``AssistantMessage`` is re-emitted as
|
||||||
|
``content_block_start`` + one delta + ``content_block_stop`` per block —
|
||||||
|
a "stream" of finished blocks, because the PTY transport has nothing
|
||||||
|
finer to offer. With ``transport="stream_json"`` *and*
|
||||||
|
``include_partial_messages``, the backend instead feeds us the real
|
||||||
|
token-level ``StreamEvent``s and we emit from those; the whole-block
|
||||||
|
records still arrive and are still kept for :class:`TurnCapture`, but
|
||||||
|
emitting from both would duplicate every block on the wire.
|
||||||
|
|
||||||
The per-request ``system`` parameter is intentionally **ignored** —
|
The per-request ``system`` parameter is intentionally **ignored** —
|
||||||
``BackendOptions.system_prompt`` is fixed at session-spawn time, and the
|
``BackendOptions.system_prompt`` is fixed at session-spawn time, and the
|
||||||
@@ -29,6 +38,7 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import uuid
|
import uuid
|
||||||
|
from collections.abc import Mapping # runtime import: isinstance in _emit_stream_event
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from typing import TYPE_CHECKING, Any, Self
|
from typing import TYPE_CHECKING, Any, Self
|
||||||
|
|
||||||
@@ -37,6 +47,7 @@ from claude_code_api import (
|
|||||||
BackendOptions,
|
BackendOptions,
|
||||||
ClaudeCodeBackend,
|
ClaudeCodeBackend,
|
||||||
ResultMessage,
|
ResultMessage,
|
||||||
|
StreamEvent,
|
||||||
TextBlock,
|
TextBlock,
|
||||||
ThinkingBlock,
|
ThinkingBlock,
|
||||||
ToolUseBlock,
|
ToolUseBlock,
|
||||||
@@ -60,7 +71,7 @@ from beaver_gateway.core.events import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from collections.abc import AsyncIterator, Iterable, Mapping
|
from collections.abc import AsyncIterator, Iterable
|
||||||
|
|
||||||
from anthropic.types import MessageParam
|
from anthropic.types import MessageParam
|
||||||
|
|
||||||
@@ -184,6 +195,8 @@ def _build_backend_options(
|
|||||||
append_system_prompt=opt.append_system_prompt,
|
append_system_prompt=opt.append_system_prompt,
|
||||||
allowed_tools=allowed_tools,
|
allowed_tools=allowed_tools,
|
||||||
mcp_servers=_build_mcp_servers(agent, mcp_internal_urls),
|
mcp_servers=_build_mcp_servers(agent, mcp_internal_urls),
|
||||||
|
transport=opt.transport,
|
||||||
|
include_partial_messages=opt.include_partial_messages,
|
||||||
disallowed_tools=opt.disallowed_tools,
|
disallowed_tools=opt.disallowed_tools,
|
||||||
permission_mode=opt.permission_mode,
|
permission_mode=opt.permission_mode,
|
||||||
dangerously_skip_permissions=opt.dangerously_skip_permissions,
|
dangerously_skip_permissions=opt.dangerously_skip_permissions,
|
||||||
@@ -215,9 +228,16 @@ class ClaudeCodeBackendAdapter:
|
|||||||
self, *, agent: ClaudeAgent, mcp_internal_urls: Mapping[str, str]
|
self, *, agent: ClaudeAgent, mcp_internal_urls: Mapping[str, str]
|
||||||
) -> None:
|
) -> None:
|
||||||
self._agent = agent
|
self._agent = agent
|
||||||
self._backend = ClaudeCodeBackend(
|
options = _build_backend_options(agent, mcp_internal_urls)
|
||||||
_build_backend_options(agent, mcp_internal_urls)
|
self._backend = ClaudeCodeBackend(options)
|
||||||
|
self._streaming = (
|
||||||
|
options.transport == "stream_json" and options.include_partial_messages
|
||||||
)
|
)
|
||||||
|
"""Whether the backend feeds us token-level ``StreamEvent``s.
|
||||||
|
|
||||||
|
When it does, blocks are emitted from those and the whole-block
|
||||||
|
``AssistantMessage`` records are used only for bookkeeping —
|
||||||
|
emitting from both would duplicate every block on the wire."""
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def agent(self) -> ClaudeAgent:
|
def agent(self) -> ClaudeAgent:
|
||||||
@@ -229,12 +249,16 @@ class ClaudeCodeBackendAdapter:
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def live_sessions(self) -> dict[str, Any]:
|
def live_sessions(self) -> dict[str, Any]:
|
||||||
"""Live PTY processes keyed by claude session_id.
|
"""Live claude processes keyed by claude session_id.
|
||||||
|
|
||||||
Pass-through to the underlying ``ClaudeCodeBackend``. The value
|
Pass-through to the underlying ``ClaudeCodeBackend``. The value
|
||||||
type is the claude-code-api ``PtyClaudeProcess``; admin code
|
is a ``PtyClaudeProcess`` or a ``StreamClaudeProcess`` depending
|
||||||
consumes ``captured_output()`` / ``add_output_listener`` /
|
on the agent's transport; both implement the
|
||||||
``write`` from it. Typed as ``Any`` to avoid leaking the lower
|
``captured_output()`` / ``add_output_listener`` / ``write``
|
||||||
|
surface the admin terminal consumes, so it works either way —
|
||||||
|
with the caveat that the stream transport's "terminal" is a
|
||||||
|
read-only view of the JSON event stream rather than a TUI you
|
||||||
|
can type into. Typed as ``Any`` to avoid leaking the lower
|
||||||
layer's type into the gateway's public surface.
|
layer's type into the gateway's public surface.
|
||||||
"""
|
"""
|
||||||
return self._backend.live_sessions
|
return self._backend.live_sessions
|
||||||
@@ -302,7 +326,13 @@ class ClaudeCodeBackendAdapter:
|
|||||||
|
|
||||||
async for event in self._backend.complete(msgs_list):
|
async for event in self._backend.complete(msgs_list):
|
||||||
raw_events.append(event)
|
raw_events.append(event)
|
||||||
if isinstance(event, AssistantMessage):
|
if isinstance(event, StreamEvent):
|
||||||
|
# Only ever arrives when ``self._streaming`` — the
|
||||||
|
# backend doesn't emit these otherwise. Indices are
|
||||||
|
# already rebased onto the whole turn by claude-code-api.
|
||||||
|
for ev in _emit_stream_event(event.event):
|
||||||
|
yield ev
|
||||||
|
elif isinstance(event, AssistantMessage):
|
||||||
for block in event.content:
|
for block in event.content:
|
||||||
if isinstance(block, TextBlock):
|
if isinstance(block, TextBlock):
|
||||||
n_text += 1
|
n_text += 1
|
||||||
@@ -310,8 +340,9 @@ class ClaudeCodeBackendAdapter:
|
|||||||
n_thinking += 1
|
n_thinking += 1
|
||||||
elif isinstance(block, ToolUseBlock):
|
elif isinstance(block, ToolUseBlock):
|
||||||
n_tool_use += 1
|
n_tool_use += 1
|
||||||
for ev in _emit_block(block, next_index):
|
if not self._streaming:
|
||||||
yield ev
|
for ev in _emit_block(block, next_index):
|
||||||
|
yield ev
|
||||||
next_index += 1
|
next_index += 1
|
||||||
elif isinstance(event, ResultMessage):
|
elif isinstance(event, ResultMessage):
|
||||||
# ResultMessage is the terminal event from TurnManager
|
# ResultMessage is the terminal event from TurnManager
|
||||||
@@ -349,6 +380,67 @@ class ClaudeCodeBackendAdapter:
|
|||||||
yield build_message_stop()
|
yield build_message_stop()
|
||||||
|
|
||||||
|
|
||||||
|
def _emit_stream_event(event: Mapping[str, Any]) -> Iterable[MessageStreamEvent]:
|
||||||
|
"""Render one raw Anthropic streaming event through our own builders.
|
||||||
|
|
||||||
|
The payload is already Anthropic-shaped, so forwarding it verbatim
|
||||||
|
is tempting — but it arrives as a plain dict and the frontend
|
||||||
|
serializes pydantic models, so it would have to be validated against
|
||||||
|
the SDK's ``RawMessageStreamEvent`` union anyway. Rebuilding it from
|
||||||
|
the fields we recognize is the same amount of work and fails the
|
||||||
|
same way :func:`_emit_block` already does: a block or delta type
|
||||||
|
this gateway doesn't know is dropped rather than raising mid-stream
|
||||||
|
on a claude release that added one.
|
||||||
|
|
||||||
|
Only ``content_block_*`` events reach here — claude-code-api strips
|
||||||
|
the inner per-request message envelopes, because the caller owns the
|
||||||
|
single envelope spanning the whole turn.
|
||||||
|
"""
|
||||||
|
etype = event.get("type")
|
||||||
|
index = event.get("index")
|
||||||
|
if not isinstance(index, int):
|
||||||
|
return ()
|
||||||
|
|
||||||
|
if etype == "content_block_start":
|
||||||
|
block = event.get("content_block")
|
||||||
|
if not isinstance(block, Mapping):
|
||||||
|
return ()
|
||||||
|
btype = block.get("type")
|
||||||
|
if btype == "text":
|
||||||
|
return (build_text_block_start(index),)
|
||||||
|
if btype == "thinking":
|
||||||
|
return (build_thinking_block_start(index),)
|
||||||
|
if btype == "tool_use":
|
||||||
|
return (
|
||||||
|
build_tool_use_block_start(
|
||||||
|
index,
|
||||||
|
tool_use_id=str(block.get("id", "")),
|
||||||
|
name=str(block.get("name", "")),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return ()
|
||||||
|
|
||||||
|
if etype == "content_block_delta":
|
||||||
|
delta = event.get("delta")
|
||||||
|
if not isinstance(delta, Mapping):
|
||||||
|
return ()
|
||||||
|
dtype = delta.get("type")
|
||||||
|
if dtype == "text_delta":
|
||||||
|
return (build_text_delta(index, str(delta.get("text", ""))),)
|
||||||
|
if dtype == "thinking_delta":
|
||||||
|
return (build_thinking_delta(index, str(delta.get("thinking", ""))),)
|
||||||
|
if dtype == "signature_delta":
|
||||||
|
return (build_signature_delta(index, str(delta.get("signature", ""))),)
|
||||||
|
if dtype == "input_json_delta":
|
||||||
|
return (build_input_json_delta(index, str(delta.get("partial_json", ""))),)
|
||||||
|
return ()
|
||||||
|
|
||||||
|
if etype == "content_block_stop":
|
||||||
|
return (build_content_block_stop(index),)
|
||||||
|
|
||||||
|
return ()
|
||||||
|
|
||||||
|
|
||||||
def _emit_block(
|
def _emit_block(
|
||||||
block: TextBlock | ThinkingBlock | ToolUseBlock | Any, index: int
|
block: TextBlock | ThinkingBlock | ToolUseBlock | Any, index: int
|
||||||
) -> Iterable[MessageStreamEvent]:
|
) -> Iterable[MessageStreamEvent]:
|
||||||
|
|||||||
@@ -287,7 +287,7 @@ local = [
|
|||||||
{ name = "raycast-api", version = "0.1.0", source = { editable = "../raycast-api" } },
|
{ name = "raycast-api", version = "0.1.0", source = { editable = "../raycast-api" } },
|
||||||
]
|
]
|
||||||
prod = [
|
prod = [
|
||||||
{ name = "claude-code-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#aa7beea1e014ba0dd8cd4980fee4b9302c57b212" } },
|
{ name = "claude-code-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#76799179d9437dcd133a40ffcf5eecc524063c2f" } },
|
||||||
{ name = "raycast-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/raycast-api.git#e73894c8e435da5c0709f92f69f11bcd0dab9afe" } },
|
{ name = "raycast-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/raycast-api.git#e73894c8e435da5c0709f92f69f11bcd0dab9afe" } },
|
||||||
]
|
]
|
||||||
|
|
||||||
@@ -419,7 +419,7 @@ wheels = [
|
|||||||
[[package]]
|
[[package]]
|
||||||
name = "claude-code-api"
|
name = "claude-code-api"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#aa7beea1e014ba0dd8cd4980fee4b9302c57b212" }
|
source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#76799179d9437dcd133a40ffcf5eecc524063c2f" }
|
||||||
resolution-markers = [
|
resolution-markers = [
|
||||||
"python_full_version >= '3.14'",
|
"python_full_version >= '3.14'",
|
||||||
"python_full_version < '3.14'",
|
"python_full_version < '3.14'",
|
||||||
|
|||||||
Reference in New Issue
Block a user