Spaces:
Running on Zero
Running on Zero
File size: 2,518 Bytes
513fab2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | #!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Part 491: CLI helper to send a prompt to /api/agent/stream and print SSE events.
Usage: python scripts/chat_test.py "What is 123 * 456?"
"""
import sys
import os
import json
import argparse
from pathlib import Path
# Load .env
env_path = Path(__file__).parent.parent / ".env"
if env_path.exists():
for line in env_path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if line and not line.startswith("#") and "=" in line:
k, v = line.split("=", 1)
os.environ[k.strip()] = v.strip()
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fastapi.testclient import TestClient
from sunnyware.main import app
def main():
parser = argparse.ArgumentParser(description="Test /api/agent/stream")
parser.add_argument("prompt", help="Prompt to send")
parser.add_argument("--session", default="cli-test", help="Session ID")
args = parser.parse_args()
print("Prompt: %s" % args.prompt)
print("Session: %s\n" % args.session)
with TestClient(app) as client:
with client.stream("POST", "/api/agent/stream", json={
"input": args.prompt,
"session_id": args.session,
}) as resp:
if resp.status_code != 200:
print("HTTP %s: %s" % (resp.status_code, resp.read().decode()[:300]))
sys.exit(1)
for line in resp.iter_lines():
if not line:
continue
if isinstance(line, bytes):
line = line.decode("utf-8", errors="replace")
if not line.startswith("data:"):
continue
payload = line[5:].strip()
if not payload:
continue
try:
data = json.loads(payload)
except json.JSONDecodeError:
continue
typ = data.get("type", "")
if typ == "tool_call":
print(" [tool_call] %s(%s)" % (data.get("tool"), data.get("args")))
elif typ == "tool_result":
print(" [tool_result] %s -> %s" % (data.get("tool"), str(data.get("result"))[:100]))
elif typ == "answer":
print(" [answer] %s" % data.get("content", "")[:200])
elif typ == "error":
print(" [error] %s" % data.get("error"))
if __name__ == "__main__":
main()
|