Skip to content

Commit 84de7ca

Browse files
declan-scaleclaude
declan-scale
and
claude
committed
feat(claude-code): stream-json parser tap for the unified harness surface (convert_claude_code_to_agentex_events + ClaudeCodeTurn)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 2e820c7 commit 84de7ca

6 files changed

Lines changed: 1563 additions & 0 deletions

File tree

‎src/agentex/lib/adk/__init__.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,11 @@
1313
from agentex.lib.adk._modules._pydantic_ai_async import stream_pydantic_ai_events
1414
from agentex.lib.adk._modules._pydantic_ai_sync import convert_pydantic_ai_to_agentex_events
1515
from agentex.lib.adk._modules._pydantic_ai_tracing import create_pydantic_ai_tracing_handler
16+
from agentex.lib.adk._modules._claude_code_sync import convert_claude_code_to_agentex_events
17+
from agentex.lib.adk._modules._claude_code_turn import (
18+
ClaudeCodeTurn,
19+
claude_code_usage_to_turn_usage,
20+
)
1621
from agentex.lib.adk._modules.events import EventsModule
1722
from agentex.lib.adk._modules.messages import MessagesModule
1823
from agentex.lib.adk._modules.state import StateModule
@@ -54,6 +59,10 @@
5459
"stream_pydantic_ai_events",
5560
"convert_pydantic_ai_to_agentex_events",
5661
"create_pydantic_ai_tracing_handler",
62+
# Claude Code
63+
"convert_claude_code_to_agentex_events",
64+
"ClaudeCodeTurn",
65+
"claude_code_usage_to_turn_usage",
5766
# Providers
5867
"providers",
5968
# Utils

‎src/agentex/lib/adk/_modules/_claude_code_sync.py‎

Lines changed: 364 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,364 @@
1+
"""Claude Code stream-json parser tap for the unified harness surface.
2+
3+
Converts the newline-delimited JSON envelopes emitted by
4+
``claude -p --output-format stream-json`` into the canonical
5+
``StreamTaskMessage*`` stream consumed by the Agentex harness.
6+
7+
Envelope → canonical mapping
8+
-----------------------------
9+
system/init
10+
Ignored at this layer (session_id tracking is a provider concern).
11+
12+
assistant / user (content blocks)
13+
text block → Start(TextContent) + Delta(TextDelta)* + Done
14+
thinking block → Start(ReasoningContent) + Delta(ReasoningContentDelta)* + Done
15+
tool_use block → Start(ToolRequestContent) + Done (Full args in Start content)
16+
tool_result block → Full(ToolResponseContent)
17+
18+
stream_event / content_block_start
19+
type=text → Start(TextContent, empty)
20+
type=thinking → Start(ReasoningContent, empty)
21+
22+
stream_event / content_block_delta
23+
type=text_delta → Delta(TextDelta)
24+
type=thinking_delta → Delta(ReasoningContentDelta)
25+
26+
stream_event / content_block_stop
27+
(text open) → Done
28+
(thinking open) → Done (full text known here; update Full via Full event first)
29+
30+
result
31+
Fires ``on_result`` with the raw envelope so the caller can capture
32+
usage and cost. No StreamTaskMessage is emitted for the result itself.
33+
34+
Out of scope
35+
------------
36+
No deployable test agent is provided. claude-code requires the golden
37+
agent's sandbox/subprocess/secret/MCP orchestration to produce the stream.
38+
Live coverage is the golden agent, which will adopt this tap. Do NOT add an
39+
examples/ agent or CI live-matrix row for claude-code.
40+
"""
41+
42+
from __future__ import annotations
43+
44+
import json
45+
from typing import Any, Callable, Awaitable, AsyncIterator
46+
47+
from agentex.lib.utils.logging import make_logger
48+
from agentex.types.text_content import TextContent
49+
from agentex.types.reasoning_content import ReasoningContent
50+
from agentex.types.task_message_delta import TextDelta
51+
from agentex.types.task_message_update import (
52+
StreamTaskMessageDone,
53+
StreamTaskMessageFull,
54+
StreamTaskMessageDelta,
55+
StreamTaskMessageStart,
56+
)
57+
from agentex.types.tool_request_content import ToolRequestContent
58+
from agentex.types.tool_response_content import ToolResponseContent
59+
from agentex.types.reasoning_content_delta import ReasoningContentDelta
60+
61+
logger = make_logger(__name__)
62+
63+
_MAX_RESULT_LENGTH = 4000
64+
65+
66+
def _truncate(text: str) -> str:
67+
return str(text)[:_MAX_RESULT_LENGTH]
68+
69+
70+
def _extract_summary(text: str, max_len: int = 300) -> str:
71+
return text.strip().split("\n", 1)[0][:max_len]
72+
73+
74+
async def convert_claude_code_to_agentex_events(
75+
lines: AsyncIterator[str | dict[str, Any]],
76+
on_result: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
77+
) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]:
78+
"""Convert a claude-code ``stream-json`` line stream into Agentex ``StreamTaskMessage*`` events.
79+
80+
Each item in ``lines`` is either a raw JSON string (as read from the CLI's
81+
stdout) or an already-parsed dict. Empty strings are skipped; unparseable
82+
JSON is logged and skipped.
83+
84+
``on_result`` is called with the ``result`` envelope when it arrives so the
85+
caller can capture usage and cost. It is awaited before the generator
86+
continues. When ``None``, the result envelope is silently dropped.
87+
88+
Envelope → canonical mapping is documented in this module's docstring.
89+
"""
90+
next_index = 0
91+
tool_call_count = 0
92+
93+
# Streaming state for content_block_start / content_block_delta /
94+
# content_block_stop triples.
95+
_thinking_open = False
96+
_thinking_buf = ""
97+
_thinking_index: int | None = None
98+
_text_open = False
99+
_text_buf = ""
100+
_text_index: int | None = None
101+
# Track which assistant-message block indices were already streamed via
102+
# stream_event triples. Those blocks must not be re-emitted when the full
103+
# assistant message arrives.
104+
_streamed_block_indexes: set[int] = set()
105+
_saw_text_stream = False
106+
_saw_thinking_stream = False
107+
# For deferred ReasoningStarted: if a content_block_start(thinking) arrives
108+
# but no thinking_delta ever follows, the final assistant block's thinking
109+
# field fills the reasoning content instead.
110+
_pending_thinking_block_index: int | None = None
111+
112+
async for raw in lines:
113+
if not raw:
114+
continue
115+
116+
if isinstance(raw, dict):
117+
evt = raw
118+
else:
119+
line = raw.strip()
120+
if not line:
121+
continue
122+
try:
123+
evt = json.loads(line)
124+
except json.JSONDecodeError:
125+
logger.debug("claude-code: skipping non-JSON line: %r", line[:120])
126+
continue
127+
128+
evt_type = evt.get("type", "")
129+
130+
# -----------------------------------------------------------------------
131+
# assistant / user — materialised content blocks
132+
# -----------------------------------------------------------------------
133+
if evt_type in ("assistant", "user"):
134+
msg = evt.get("message", {})
135+
blocks = msg.get("content", [])
136+
if not isinstance(blocks, list):
137+
blocks = [blocks]
138+
139+
for idx, block in enumerate(blocks):
140+
if not isinstance(block, dict):
141+
continue
142+
block_type = block.get("type", "")
143+
144+
if block_type == "text":
145+
# Skip if already delivered via stream_event deltas
146+
if _saw_text_stream or idx in _streamed_block_indexes:
147+
continue
148+
text = block.get("text", "")
149+
if text:
150+
msg_index = next_index
151+
next_index += 1
152+
yield StreamTaskMessageStart(
153+
type="start",
154+
index=msg_index,
155+
content=TextContent(
156+
type="text",
157+
author="agent",
158+
content="",
159+
),
160+
)
161+
yield StreamTaskMessageDelta(
162+
type="delta",
163+
index=msg_index,
164+
delta=TextDelta(type="text", text_delta=text),
165+
)
166+
yield StreamTaskMessageDone(type="done", index=msg_index)
167+
168+
elif block_type == "thinking":
169+
# Skip if already delivered via stream_event deltas
170+
if _saw_thinking_stream or idx in _streamed_block_indexes:
171+
continue
172+
thinking_text = block.get("thinking", "")
173+
if thinking_text:
174+
summary = _extract_summary(thinking_text)
175+
msg_index = next_index
176+
next_index += 1
177+
yield StreamTaskMessageStart(
178+
type="start",
179+
index=msg_index,
180+
content=ReasoningContent(
181+
type="reasoning",
182+
author="agent",
183+
summary=[summary],
184+
content=[],
185+
style="active",
186+
),
187+
)
188+
yield StreamTaskMessageDelta(
189+
type="delta",
190+
index=msg_index,
191+
delta=ReasoningContentDelta(
192+
type="reasoning_content",
193+
content_index=0,
194+
content_delta=thinking_text,
195+
),
196+
)
197+
yield StreamTaskMessageDone(type="done", index=msg_index)
198+
199+
elif block_type == "tool_use":
200+
tool_call_count += 1
201+
tool_id = block.get("id", f"tool_{tool_call_count}")
202+
name = block.get("name", "unknown")
203+
arguments = block.get("input", {})
204+
if not isinstance(arguments, dict):
205+
arguments = {}
206+
msg_index = next_index
207+
next_index += 1
208+
yield StreamTaskMessageStart(
209+
type="start",
210+
index=msg_index,
211+
content=ToolRequestContent(
212+
type="tool_request",
213+
author="agent",
214+
tool_call_id=tool_id,
215+
name=name,
216+
arguments=arguments,
217+
),
218+
)
219+
yield StreamTaskMessageDone(type="done", index=msg_index)
220+
221+
elif block_type == "tool_result":
222+
tool_id = block.get("tool_use_id", "")
223+
content = block.get("content", "")
224+
is_error = block.get("is_error", False)
225+
if isinstance(content, list):
226+
content = "\n".join(b.get("text", str(b)) if isinstance(b, dict) else str(b) for b in content)
227+
result_str = _truncate(str(content))
228+
msg_index = next_index
229+
next_index += 1
230+
yield StreamTaskMessageFull(
231+
type="full",
232+
index=msg_index,
233+
content=ToolResponseContent(
234+
type="tool_response",
235+
author="agent",
236+
tool_call_id=tool_id,
237+
name="",
238+
content={"result": result_str, **({"is_error": True} if is_error else {})},
239+
),
240+
)
241+
242+
# -----------------------------------------------------------------------
243+
# stream_event — incremental streaming deltas
244+
# -----------------------------------------------------------------------
245+
elif evt_type == "stream_event":
246+
se = evt.get("event") or {}
247+
se_type = se.get("type", "")
248+
block_index = se.get("index")
249+
250+
if se_type == "content_block_start":
251+
block = se.get("content_block") or {}
252+
btype = block.get("type")
253+
254+
if btype == "thinking":
255+
_thinking_open = True
256+
_thinking_buf = ""
257+
# Defer marking the block as streamed until we actually
258+
# receive a thinking_delta. Some configurations emit a
259+
# thinking block_start but no deltas — in that case we want
260+
# the final assistant-message handler to fill the text.
261+
_pending_thinking_block_index = block_index if isinstance(block_index, int) else None
262+
msg_index = next_index
263+
next_index += 1
264+
_thinking_index = msg_index
265+
yield StreamTaskMessageStart(
266+
type="start",
267+
index=msg_index,
268+
content=ReasoningContent(
269+
type="reasoning",
270+
author="agent",
271+
summary=[],
272+
content=[],
273+
style="active",
274+
),
275+
)
276+
277+
elif btype == "text":
278+
_text_open = True
279+
_text_buf = ""
280+
_saw_text_stream = True
281+
if isinstance(block_index, int):
282+
_streamed_block_indexes.add(block_index)
283+
msg_index = next_index
284+
next_index += 1
285+
_text_index = msg_index
286+
yield StreamTaskMessageStart(
287+
type="start",
288+
index=msg_index,
289+
content=TextContent(
290+
type="text",
291+
author="agent",
292+
content="",
293+
),
294+
)
295+
296+
elif se_type == "content_block_delta":
297+
delta = se.get("delta") or {}
298+
dtype = delta.get("type")
299+
300+
if dtype == "thinking_delta":
301+
chunk = delta.get("thinking", "")
302+
if chunk and _thinking_open:
303+
if not _saw_thinking_stream:
304+
_saw_thinking_stream = True
305+
# Now mark the block as claimed so the assistant
306+
# message handler won't re-emit it.
307+
if _pending_thinking_block_index is not None:
308+
_streamed_block_indexes.add(_pending_thinking_block_index)
309+
_thinking_buf += chunk
310+
if _thinking_index is not None:
311+
yield StreamTaskMessageDelta(
312+
type="delta",
313+
index=_thinking_index,
314+
delta=ReasoningContentDelta(
315+
type="reasoning_content",
316+
content_index=0,
317+
content_delta=chunk,
318+
),
319+
)
320+
321+
elif dtype == "text_delta":
322+
chunk = delta.get("text", "")
323+
if chunk and _text_open:
324+
_text_buf += chunk
325+
if _text_index is not None:
326+
yield StreamTaskMessageDelta(
327+
type="delta",
328+
index=_text_index,
329+
delta=TextDelta(type="text", text_delta=chunk),
330+
)
331+
332+
elif se_type == "content_block_stop":
333+
if _thinking_open:
334+
full_text = _thinking_buf
335+
_thinking_open = False
336+
_thinking_buf = ""
337+
_pending_thinking_block_index = None
338+
if _thinking_index is not None:
339+
yield StreamTaskMessageDone(type="done", index=_thinking_index)
340+
_thinking_index = None
341+
elif _text_open:
342+
_text_open = False
343+
_text_buf = ""
344+
if _text_index is not None:
345+
yield StreamTaskMessageDone(type="done", index=_text_index)
346+
_text_index = None
347+
348+
# -----------------------------------------------------------------------
349+
# system / init — session metadata (ignored at this layer)
350+
# -----------------------------------------------------------------------
351+
elif evt_type == "system":
352+
# Session ID tracking and MCP status logging are provider concerns.
353+
# This pure parser layer intentionally emits nothing for system events.
354+
pass
355+
356+
# -----------------------------------------------------------------------
357+
# result — carries usage + cost; fired to on_result, not emitted as msgs
358+
# -----------------------------------------------------------------------
359+
elif evt_type == "result":
360+
if on_result is not None:
361+
await on_result(evt)
362+
363+
else:
364+
logger.debug("claude-code: unhandled envelope type %r", evt_type)

0 commit comments

Comments
 (0)