feat: added Pydantic AI sync, async, temporal integration

[michael-chou359]

Pydantic AI Temporal:

Temporal logs also properly populated:

Pydantic AI Async:

Pydantic AI Sync:

Events Mapping:

Pydantic AI Event	Sync (HTTP yield)	Async (Redis)
PartStartEvent(TextPart)	yield StreamTaskMessageStart(TextContent) + initial delta if part.content	open streaming_task_message_context(TextContent); emit initial delta if part.content
PartDeltaEvent(TextPartDelta)	yield StreamTaskMessageDelta(TextDelta)	stream_update(StreamTaskMessageDelta(TextDelta)) on text context
PartEndEvent (text part)	yield StreamTaskMessageDone	close text context
PartStartEvent(ThinkingPart)	yield StreamTaskMessageStart(TextContent) + initial reasoning delta	open streaming_task_message_context(ReasoningContent) + initial reasoning delta
PartDeltaEvent(ThinkingPartDelta)	yield StreamTaskMessageDelta(ReasoningContentDelta) (skipped if empty)	stream_update(...) on reasoning context (skipped if empty)
PartEndEvent (thinking part)	yield StreamTaskMessageDone	close reasoning context
PartStartEvent(ToolCallPart)	yield StreamTaskMessageStart(ToolRequestContent) (+ initial args delta if args is a non-empty string)	record (tool_call_id, tool_name) for later; no immediate emission
PartDeltaEvent(ToolCallPartDelta)	yield StreamTaskMessageDelta(ToolRequestDelta) (token-by-token args)	(ignored — args accumulated; surfaced on PartEnd)
PartEndEvent (tool-call part)	yield StreamTaskMessageDone + tracing_handler.on_tool_start(...)	adk.messages.create(ToolRequestContent) with full args + tracing_handler.on_tool_start(...)
FunctionToolResultEvent	yield StreamTaskMessageFull(ToolResponseContent) + tracing_handler.on_tool_end(...)	adk.messages.create(ToolResponseContent) + tracing_handler.on_tool_end(...)
FunctionToolCallEvent	(ignored — redundant with Start/Delta/End)	(ignored)
FinalResultEvent	(ignored — informational)	(ignored)
AgentRunResultEvent	(ignored — terminator)	(ignored)

Edits to streaming.py:

CoalescingBuffer.close() was cancelling the background flush ticker mid-await leading to a race condition bug. When the cancellation landed after Redis had already accepted the write but before the await returned, the in-flight batch was re-enqueued and republished by close's final drain, producing a duplicated tail on the live UI stream. Fix: stop cancelling the ticker — signal _closed, let it finish whatever flush it's mid-way through, and let it exit naturally after its next drain pass. Each delta is now published exactly once. This bug showed up constantly with pydantic ai but not with the other integrations like langgraph or open ai sdk. Also this fix shouldn't lead to more silent losses: the "await self._on_flush(u)" is no longer interrupted so we know that the write will happen exactly once.

✅1. Sync streaming converter + tests (this PR)
✅2. Sync hello-world example agent (text-only streaming)
✅3. Sync agent with tools (tool-call delta streaming)
✅4. Tracing wired to SGP
✅5. Sync CLI template (agentex new-agent --type sync-pydantic-ai)
✅6. Async (non-Temporal) example
✅7. Temporal example using TemporalAgent + PydanticAIPlugin)
✅8. Async, Sync, Temporal template

Greptile Summary

This PR adds full Pydantic AI support to the Agentex SDK — sync HTTP streaming, async Redis streaming, a Temporal-compatible integration, CLI scaffold templates, and a bug fix for a race condition in CoalescingBuffer.close() that produced duplicate stream tails.

• Streaming converter pair: _pydantic_ai_sync.py converts the Pydantic AI event stream into StreamTaskMessage* events for HTTP yield; _pydantic_ai_async.py drives adk.streaming contexts and adk.messages.create for Redis-based delivery — both wired into adk/__init__.py.
• Tracing handler: _pydantic_ai_tracing.py records one deterministic child span per tool call using a stateless span-ID derivation (uuid5(trace_id, tool_call_id)) so spans pair correctly across Temporal activity boundaries.
• CoalescingBuffer fix: the ticker task in _run() now exits naturally after its next drain pass rather than being cancelled mid-await, eliminating the duplicate-tail symptom caused by a Redis write completing before the CancelledError landed; the regression test is updated to assert exactly-once delivery.

Confidence Score: 5/5

Safe to merge — no data loss, no incorrect message routing, and the CoalescingBuffer fix is well-tested.

The streaming race-condition fix is correct: the ticker exits naturally after its own drain pass, the final drain in close() catches any items enqueued after the last tick, and the updated regression test asserts exactly-once delivery end-to-end. Both new converters faithfully map every Pydantic AI event type and handle edge cases (initial PartStart content, empty ThinkingPartDelta, dict vs. string tool args, RetryPromptPart). The tracing handler's stateless span-ID derivation is the right approach for cross-activity Temporal runs. Coverage is thorough.

No files require special attention.

Important Files Changed

Filename	Overview
src/agentex/lib/core/services/adk/streaming.py	CoalescingBuffer race-condition fix: _run() now checks _closed after each drain pass instead of being cancelled mid-flush; close() awaits natural task exit and CancelledError propagates correctly; logic is sound and regression test is updated.
src/agentex/lib/adk/_modules/_pydantic_ai_sync.py	Sync streaming converter correctly maps all Pydantic AI part events to StreamTaskMessage* events; ThinkingPart correctly uses ReasoningContent; defensive fallbacks for unknown part indices; TextContent created without format="markdown" (matches LangGraph sync convention but differs from async counterpart).
src/agentex/lib/adk/_modules/_pydantic_ai_async.py	Async helper opens/closes streaming contexts correctly; tool-result content serialization matches the sync _tool_return_content helper; lazy pydantic-ai imports are now redundant since pydantic-ai-slim is a direct dependency.
src/agentex/lib/adk/_modules/_pydantic_ai_tracing.py	Stateless span-ID derivation via uuid5(namespace, trace_id:tool_call_id) correctly pairs on_tool_start/on_tool_end across Temporal activity boundaries; client is lazily constructed to match event-loop lifetime.
tests/lib/adk/test_pydantic_ai_sync.py	Comprehensive test coverage for sync converter using async generator fakes; covers text, thinking, tool-call, and multi-step scenarios.
tests/lib/adk/test_pydantic_ai_async.py	Comprehensive test coverage for async helper using FakeStreamingModule/FakeMessagesModule in-memory fakes; covers text, thinking, tool-call, retry-prompt, and tracing scenarios.
tests/lib/core/services/adk/test_streaming.py	Regression test updated from "may duplicate" to exact-once assertion; properly exercises the race condition with a blocking slow_flush and asyncio.sleep(0) to park close() inside the flush loop.

Sequence Diagram

sequenceDiagram
participant PA as PydanticAI Agent
participant SC as Sync Converter
participant AC as Async Helper
participant CB as CoalescingBuffer
participant Redis
participant Client
Note over PA,Client: Sync path - HTTP yield
PA->>SC: PartStartEvent(TextPart)
SC->>Client: StreamTaskMessageStart(TextContent)
PA->>SC: PartDeltaEvent(TextPartDelta)
SC->>Client: StreamTaskMessageDelta(TextDelta)
PA->>SC: PartEndEvent
SC->>Client: StreamTaskMessageDone
Note over PA,Redis: Async path - Redis streaming
PA->>AC: PartStartEvent(TextPart)
AC->>CB: open streaming_task_message_context
PA->>AC: PartDeltaEvent(TextPartDelta)
AC->>CB: buffer.add(StreamTaskMessageDelta)
CB-->>Redis: flush batch every 50ms
PA->>AC: PartEndEvent
AC->>CB: context.close()
Note over CB: set _closed and signal ticker then await natural exit
CB-->>Redis: final drain exactly once
Note over PA,Redis: Tool calls async path
PA->>AC: PartEndEvent(ToolCallPart with full args)
AC->>Redis: adk.messages.create(ToolRequestContent)
PA->>AC: FunctionToolResultEvent
AC->>Redis: adk.messages.create(ToolResponseContent)

Loading

Comments Outside Diff (1)

1. src/agentex/lib/adk/_modules/_pydantic_ai_sync.py, line 431-450 (link)

   ThinkingPart start event uses TextContent instead of ReasoningContent

   The sync converter emits StreamTaskMessageStart(content=TextContent(...)) for ThinkingPart events (line 434–440), but immediately follows with StreamTaskMessageDelta(delta=ReasoningContentDelta(...)) deltas. The async counterpart (_pydantic_ai_async.py) consistently uses ReasoningContent as the initial content type for the same event. The type mismatch — a TextContent start receiving ReasoningContentDelta updates — means any server-side logic that requires the accumulated message type to match the delta type will either fail silently or render thinking content as a plain text bubble rather than a collapsible reasoning block. ReasoningContent is not imported in this file at all, confirming the type was never set intentionally.

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: src/agentex/lib/adk/_modules/_pydantic_ai_sync.py
   Line: 431-450
   Comment:
   **ThinkingPart start event uses `TextContent` instead of `ReasoningContent`**
   The sync converter emits `StreamTaskMessageStart(content=TextContent(...))` for `ThinkingPart` events (line 434–440), but immediately follows with `StreamTaskMessageDelta(delta=ReasoningContentDelta(...))` deltas. The async counterpart (`_pydantic_ai_async.py`) consistently uses `ReasoningContent` as the initial content type for the same event. The type mismatch — a `TextContent` start receiving `ReasoningContentDelta` updates — means any server-side logic that requires the accumulated message type to match the delta type will either fail silently or render thinking content as a plain text bubble rather than a collapsible reasoning block. `ReasoningContent` is not imported in this file at all, confirming the type was never set intentionally.
   How can I resolve this? If you propose a fix, please make it concise.

   

_{Reviews (9): Last reviewed commit: "Merge branch 'next' into mc/pydantic-ai-..." | Re-trigger Greptile}

[socket-security]

Review the following changes in direct dependencies. Learn more about Socket for GitHub.

Diff	Package	Supply Chain Security	Vulnerability	Quality	Maintenance	License
	pydantic-ai-slim@​1.98.0

View full report

[danielmillerp]

overall looks good!! can you just add the new templates as well see: src/agentex/lib/cli/

[danielmillerp]

also please fix linting !

[danielmillerp]

also could you figure out why the tests aren't running in CI should look like this:

[danielmillerp]

and address any greptile comments !