Files

68 lines
2.0 KiB
Python

"""Exercise `thread.extensions[name]` against the integration API.
Every node in the example graph writes `("progress", {...})` via
`get_stream_writer`. This script verifies the `extensions["progress"]`
projection yields each progress event in order.
"""
from __future__ import annotations
import asyncio
from _common import (
ASSISTANT_ID,
check_api_reachable,
header,
make_async_client,
make_sync_client,
)
async def run_async() -> None:
header("async extensions[progress]")
threads, raw = make_async_client()
try:
async with threads.stream(assistant_id=ASSISTANT_ID) as thread:
await thread.run.start(input={"messages": [], "value": "init", "items": []})
events: list[dict] = []
async for event in thread.extensions["progress"]:
print(f" progress: {event!r}")
events.append(event)
print(f" total progress events: {len(events)}")
assert events, "expected at least one progress event"
# Verify ordering covers the node sequence.
steps = [e.get("step") for e in events]
print(f" step sequence: {steps}")
finally:
await raw.aclose()
def run_sync() -> None:
header("sync extensions[progress]")
threads, raw = make_sync_client()
try:
with threads.stream(assistant_id=ASSISTANT_ID) as thread:
thread.run.start(input={"messages": [], "value": "init", "items": []})
events: list[dict] = []
for event in thread.extensions["progress"]:
print(f" progress: {event!r}")
events.append(event)
print(f" total progress events: {len(events)}")
assert events, "expected at least one progress event"
steps = [e.get("step") for e in events]
print(f" step sequence: {steps}")
finally:
raw.close()
def main() -> None:
check_api_reachable()
asyncio.run(run_async())
run_sync()
if __name__ == "__main__":
main()