From cade2139bc12d31bfc750dba72d6dd3efba014e9 Mon Sep 17 00:00:00 2001 From: Marcus Schiesser Date: Wed, 21 Oct 2026 17:53:11 +0700 Subject: [PATCH 1/2] fix(examples): recover submit_result from MESSAGES_SNAPSHOT in structured output A reconnect behind the replay buffer answers with one MESSAGES_SNAPSHOT instead of the tool-call frames, which the read-stream snippet ignored, so a completed submit_result could be lost. A from_snapshot helper outside the docs region replaces the collected arguments with the snapshot's tool calls, as the JS, Rust and Swift snippets do. Co-Authored-By: Claude Opus 5.5 --- .../patterns/structured_output.py | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/examples/introspection_examples/patterns/structured_output.py b/examples/introspection_examples/patterns/structured_output.py index bc0eb38d..4ee7f3ff 100644 --- a/examples/introspection_examples/patterns/structured_output.py +++ b/examples/introspection_examples/patterns/structured_output.py @@ -21,12 +21,35 @@ import sys from typing import Literal -from ag_ui.core import ToolCallArgsEvent, ToolCallStartEvent +from ag_ui.core import ( + AssistantMessage, + MessagesSnapshotEvent, + ToolCallArgsEvent, + ToolCallStartEvent, +) from dotenv import load_dotenv from introspection_sdk import IntrospectionClient, Runner from pydantic import BaseModel, Field, ValidationError +def from_snapshot( + event: MessagesSnapshotEvent, args: dict[str, str] +) -> str | None: + """A reconnect behind the replay buffer sends the run so far instead of + the frames, so replace the collected arguments with its tool calls and + return the ``submit_result`` call's id.""" + args.clear() + submit_id = None + for message in event.messages: + if not isinstance(message, AssistantMessage): + continue + for call in message.tool_calls or []: + args[call.id] = call.function.arguments + if call.function.name == "submit_result": + submit_id = call.id + return submit_id + + def read_stream(runner: Runner) -> tuple[str, str | None]: # #region docs:patterns/structured-output/read-stream handle = runner.tasks.start(prompt="Triage ticket 4411: ...") @@ -34,6 +57,8 @@ def read_stream(runner: Runner) -> tuple[str, str | None]: submit_id: str | None = None for event in handle.stream(): + if isinstance(event, MessagesSnapshotEvent): + submit_id = from_snapshot(event, args) if ( isinstance(event, ToolCallStartEvent) and event.tool_call_name == "submit_result" From 6777d0eee71b09e58fb8c96cdbd7ee2dfa8843bc Mon Sep 17 00:00:00 2001 From: Marcus Schiesser Date: Wed, 21 Oct 2026 18:55:13 +0700 Subject: [PATCH 2/2] fix(examples): show the helpers each docs region calls Move from_snapshot into the structured-output read-stream region and replace hidden app placeholders (send_event_to_browser, save_agent_binding, render) with minimal stubs inside the regions that call them. AGENTS.md states the rule for later regions. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 7 ++++ .../patterns/add_to_your_app.py | 42 ++++++++----------- .../patterns/structured_output.py | 35 ++++++++-------- .../reference/python_sdk.py | 8 ++-- 4 files changed, 46 insertions(+), 46 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8c89aa60..609d997f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,6 +104,13 @@ Examples are documentation, not scratch space. Each example must: - Be invoked by `.github/workflows/examples.yml` so the examples workflow exercises it. +A `# #region docs:` block must show every helper it calls that the +example file defines, methods and closures included. Move SDK work (stream +parsing, snapshot recovery, text accumulation) inside the region, before the +code that uses it; stand in for the reader's own app code (`render`, +`save_agent_binding`, ...) with a minimal stub inside the region under a +one-line `# Your app's code: ...` comment. + ## Linting and formatting ```shell diff --git a/examples/introspection_examples/patterns/add_to_your_app.py b/examples/introspection_examples/patterns/add_to_your_app.py index 3676ce0f..eb167bb3 100644 --- a/examples/introspection_examples/patterns/add_to_your_app.py +++ b/examples/introspection_examples/patterns/add_to_your_app.py @@ -44,26 +44,6 @@ class AgentBinding: user_id: str -class Orders: - """Stands in for your database: which task works on which order.""" - - def __init__(self) -> None: - self.bindings: dict[str, AgentBinding] = {} - - def save_agent_binding( - self, order_id: str, *, task_id: str, user_id: str - ) -> None: - self.bindings[order_id] = AgentBinding(task_id, user_id) - - -orders = Orders() - - -def send_event_to_browser(event: AGUIEvent) -> None: - """Stands in for your server-sent events or WebSocket push.""" - print(f" -> browser: {event.type}") - - def open_runner( runtime: str, user_id: str ) -> tuple[IntrospectionClient, IntrospectionLogs, Runner]: @@ -85,23 +65,38 @@ def open_runner( return client, outcomes, runner -def start_task(runner: Runner, order_id: str, user_id: str) -> None: +def start_task(runner: Runner, order_id: str, user_id: str) -> AgentBinding: # #region docs:patterns/add-to-your-app/start-task + # Your app's code: store which task works on which order. + def save_agent_binding( + order_id: str, *, task_id: str, user_id: str + ) -> None: + print(f" saved task {task_id} for order {order_id} ({user_id})") + + # Your app's code: push the event over server-sent events or a WebSocket. + def send_event_to_browser(event: AGUIEvent) -> None: + print(f" -> browser: {event.type}") + handle = runner.tasks.start( prompt="Review order 1842 against the refund policy and draft the next response.", metadata={"order_id": "1842"}, ) - orders.save_agent_binding( + save_agent_binding( order_id, task_id=str(handle.run.task_id), user_id=user_id ) for event in handle.stream(): send_event_to_browser(event) # #endregion docs:patterns/add-to-your-app/start-task + return AgentBinding(str(handle.run.task_id), user_id) def reconnect(runner: Runner, binding: AgentBinding) -> None: # #region docs:patterns/add-to-your-app/reconnect + # Your app's code: push the event over server-sent events or a WebSocket. + def send_event_to_browser(event: AGUIEvent) -> None: + print(f" -> browser: {event.type}") + task = runner.tasks.get(binding.task_id) metadata = task.metadata or {} run_id = metadata.get("active_run_id") or metadata["latest_run_id"] @@ -144,8 +139,7 @@ def main() -> None: client, outcomes, runner = open_runner(runtime, user_id) try: - start_task(runner, "1842", user_id) - binding = orders.bindings["1842"] + binding = start_task(runner, "1842", user_id) print(f"task {binding.task_id} saved for order 1842") # After a reload: check the record belongs to the signed-in user. diff --git a/examples/introspection_examples/patterns/structured_output.py b/examples/introspection_examples/patterns/structured_output.py index 4ee7f3ff..6635c2a4 100644 --- a/examples/introspection_examples/patterns/structured_output.py +++ b/examples/introspection_examples/patterns/structured_output.py @@ -32,26 +32,25 @@ from pydantic import BaseModel, Field, ValidationError -def from_snapshot( - event: MessagesSnapshotEvent, args: dict[str, str] -) -> str | None: - """A reconnect behind the replay buffer sends the run so far instead of - the frames, so replace the collected arguments with its tool calls and - return the ``submit_result`` call's id.""" - args.clear() - submit_id = None - for message in event.messages: - if not isinstance(message, AssistantMessage): - continue - for call in message.tool_calls or []: - args[call.id] = call.function.arguments - if call.function.name == "submit_result": - submit_id = call.id - return submit_id - - def read_stream(runner: Runner) -> tuple[str, str | None]: # #region docs:patterns/structured-output/read-stream + def from_snapshot( + event: MessagesSnapshotEvent, args: dict[str, str] + ) -> str | None: + """A reconnect behind the replay buffer sends the run so far instead of + the frames, so replace the collected arguments with its tool calls and + return the ``submit_result`` call's id.""" + args.clear() + submit_id = None + for message in event.messages: + if not isinstance(message, AssistantMessage): + continue + for call in message.tool_calls or []: + args[call.id] = call.function.arguments + if call.function.name == "submit_result": + submit_id = call.id + return submit_id + handle = runner.tasks.start(prompt="Triage ticket 4411: ...") args: dict[str, str] = {} submit_id: str | None = None diff --git a/examples/introspection_examples/reference/python_sdk.py b/examples/introspection_examples/reference/python_sdk.py index 8aca734e..e57a661a 100644 --- a/examples/introspection_examples/reference/python_sdk.py +++ b/examples/introspection_examples/reference/python_sdk.py @@ -35,14 +35,14 @@ from introspection_sdk.schemas.agui import AGUIEvent, RunFinishedEvent -def render(event: AGUIEvent) -> None: - print(event.model_dump_json(by_alias=True, exclude_none=True)) - - # The regions keep the docs' line lengths, not the formatter's. # fmt: off async def reattach(runner: AsyncRunner, task_id: str) -> None: # #region docs:reference/python/reattach + # Your app's code: show the event in your UI. + def render(event: AGUIEvent) -> None: + print(event.model_dump_json(by_alias=True, exclude_none=True)) + task = await runner.tasks.get(task_id) metadata = task.metadata or {} run_id = metadata.get("active_run_id") or metadata["latest_run_id"]