|
| 1 | +"""Send a text task to server.py and display its ADK event stream.""" |
| 2 | + |
| 3 | +import argparse |
| 4 | +import asyncio |
| 5 | +import json |
| 6 | +import sys |
| 7 | +from urllib.parse import quote |
| 8 | + |
| 9 | +import httpx |
| 10 | + |
| 11 | + |
| 12 | +async def session_id_for(client, app_name, user_id, session_id=None): |
| 13 | + path = f"/apps/{quote(app_name, safe='')}/users/{quote(user_id, safe='')}/sessions" |
| 14 | + if session_id: |
| 15 | + response = await client.get(f"{path}/{quote(session_id, safe='')}") |
| 16 | + response.raise_for_status() |
| 17 | + return response.json()["id"] |
| 18 | + response = await client.post(path, json={}) |
| 19 | + response.raise_for_status() |
| 20 | + return response.json()["id"] |
| 21 | + |
| 22 | + |
| 23 | +async def stream_events(client, *, app_name, user_id, session_id, task): |
| 24 | + body = { |
| 25 | + "appName": app_name, |
| 26 | + "userId": user_id, |
| 27 | + "sessionId": session_id, |
| 28 | + "newMessage": {"role": "user", "parts": [{"text": task}]}, |
| 29 | + "streaming": True, |
| 30 | + } |
| 31 | + async with client.stream("POST", "/run_sse", json=body) as response: |
| 32 | + response.raise_for_status() |
| 33 | + data = [] |
| 34 | + async for line in response.aiter_lines(): |
| 35 | + if line.startswith("data:"): |
| 36 | + data.append(line[5:].lstrip(" ")) |
| 37 | + elif not line and data: |
| 38 | + yield json.loads("\n".join(data)) |
| 39 | + data.clear() |
| 40 | + if data: |
| 41 | + yield json.loads("\n".join(data)) |
| 42 | + |
| 43 | + |
| 44 | +class EventPrinter: |
| 45 | + def __init__(self): |
| 46 | + self.segment = None |
| 47 | + self.author = None |
| 48 | + |
| 49 | + def finish(self): |
| 50 | + if self.segment is not None: |
| 51 | + print(flush=True) |
| 52 | + self.segment = None |
| 53 | + |
| 54 | + def __call__(self, event): |
| 55 | + error = event.get("error") or event.get("errorMessage") |
| 56 | + if error: |
| 57 | + self.finish() |
| 58 | + raise RuntimeError(str(error)) |
| 59 | + final = False |
| 60 | + parts = (event.get("content") or {}).get("parts", []) |
| 61 | + author = event.get("author", "unknown") |
| 62 | + if parts and author != self.author: |
| 63 | + self.finish() |
| 64 | + self.author = author |
| 65 | + print(f"[agent: {author}]", flush=True) |
| 66 | + for part in parts: |
| 67 | + if "functionCall" in part or "functionResponse" in part: |
| 68 | + self.finish() |
| 69 | + key = "functionCall" if "functionCall" in part else "functionResponse" |
| 70 | + label = "tool" if key == "functionCall" else "tool result" |
| 71 | + print( |
| 72 | + f"[{label}]", json.dumps(part[key], ensure_ascii=False), flush=True |
| 73 | + ) |
| 74 | + elif part.get("text"): |
| 75 | + if part.get("thought"): |
| 76 | + label = "progress" |
| 77 | + elif event.get("partial"): |
| 78 | + label = "delta" |
| 79 | + else: |
| 80 | + label = "answer" |
| 81 | + final = True |
| 82 | + if label in {"progress", "delta"}: |
| 83 | + if self.segment != label: |
| 84 | + self.finish() |
| 85 | + print(f"[{label}] ", end="", flush=True) |
| 86 | + self.segment = label |
| 87 | + print(part["text"], end="", flush=True) |
| 88 | + else: |
| 89 | + self.finish() |
| 90 | + print(f"[{label}] {part['text']}", flush=True) |
| 91 | + return final |
| 92 | + |
| 93 | + |
| 94 | +async def run(args): |
| 95 | + # Match the agent's execution/readiness budget, allowing quiet long-running tasks. |
| 96 | + async with httpx.AsyncClient( |
| 97 | + base_url=args.url.rstrip("/"), |
| 98 | + timeout=httpx.Timeout(args.timeout, connect=10), |
| 99 | + trust_env=False, |
| 100 | + ) as client: |
| 101 | + session_id = await session_id_for( |
| 102 | + client, args.app_name, args.user_id, args.session_id |
| 103 | + ) |
| 104 | + print(f"session_id={session_id}", flush=True) |
| 105 | + print("Reuse this ID with --session-id for the next turn.", flush=True) |
| 106 | + final = False |
| 107 | + display_event = EventPrinter() |
| 108 | + try: |
| 109 | + async for event in stream_events( |
| 110 | + client, |
| 111 | + app_name=args.app_name, |
| 112 | + user_id=args.user_id, |
| 113 | + session_id=session_id, |
| 114 | + task=args.task, |
| 115 | + ): |
| 116 | + final = display_event(event) or final |
| 117 | + finally: |
| 118 | + display_event.finish() |
| 119 | + if not final: |
| 120 | + raise RuntimeError( |
| 121 | + "Stream ended without a final answer; execution is unconfirmed. Do not automatically resubmit." |
| 122 | + ) |
| 123 | + |
| 124 | + |
| 125 | +def main(): |
| 126 | + parser = argparse.ArgumentParser(description=__doc__) |
| 127 | + parser.add_argument("task", help="Text task to execute in the sandbox") |
| 128 | + parser.add_argument("--url", default="http://127.0.0.1:8000") |
| 129 | + parser.add_argument("--app-name", default="remote_sandbox_demo") |
| 130 | + parser.add_argument("--user-id", default="demo_user") |
| 131 | + parser.add_argument("--session-id", help="Resume an existing session for this user") |
| 132 | + parser.add_argument("--timeout", type=float, default=1500) |
| 133 | + args = parser.parse_args() |
| 134 | + if not args.task.strip(): |
| 135 | + parser.error("task must contain text") |
| 136 | + if args.timeout <= 0: |
| 137 | + parser.error("timeout must be positive") |
| 138 | + try: |
| 139 | + asyncio.run(run(args)) |
| 140 | + except (httpx.HTTPError, RuntimeError, ValueError) as exc: |
| 141 | + print(f"Request failed: {exc}", file=sys.stderr) |
| 142 | + raise SystemExit(1) from None |
| 143 | + except KeyboardInterrupt: |
| 144 | + raise SystemExit(130) from None |
| 145 | + |
| 146 | + |
| 147 | +if __name__ == "__main__": |
| 148 | + main() |
0 commit comments