Setting the file. One moment. foundry-agent/create/scripts/smoke_test_gpt_live.py
Python·250 lines·9 KB
json
15import sys
16from urllib.error import HTTPError, URLError
17from urllib.parse import quote, urlsplit, urlunsplit
18from urllib.request import Request, urlopen
19
20from smoke_test_voice import acquire_token, positive_integer
21
22API_VERSION = "v1"
23PREVIEW_HEADER = "VoiceAgents=V1Preview"
24
25
26class SmokeError(Exception):
27 pass
28
29
30@dataclass
31class Evidence:
32 event_types: list[str] = field(default_factory=list)
33 closed: bool = False
34 close_reason: str = ""
35
36
37def parse_args() -> argparse.Namespace:
38 parser = argparse.ArgumentParser(
39 description="Smoke-test a deployed Microsoft Foundry GPT-Live Voice Agent."
40 )
41 parser.add_argument("--endpoint", required=True, help="Foundry project endpoint")
42 parser.add_argument("--agent-name", required=True, help="GPT-Live agent name")
43 parser.add_argument("--timeout", type=positive_integer, default=180)
44 parser.add_argument("--dry-run", action="store_true")
45 args = parser.parse_args()
46 if not args.endpoint.strip() or not args.agent_name.strip():
47 parser.error("--endpoint and --agent-name must not be empty")
48 return args
49
50
51def normalize_endpoint(value: str) -> str:
52 parsed = urlsplit(value.strip())
53 path = parsed.path.rstrip("/")
54 project = path.removeprefix("/api/projects/")
55 if (
56 parsed.scheme != "https"
57 or not parsed.hostname
58 or parsed.username is not None
59 or parsed.password is not None
60 or parsed.query
61 or parsed.fragment
62 or not path.startswith("/api/projects/")
63 or not project
64 or "/" in project
65 ):
66 raise SmokeError(
67 "--endpoint must be an HTTPS Foundry project endpoint ending in "
68 "/api/projects/<project>, without credentials, query, or fragment"
69 )
70 return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
71
72
73def websocket_url(endpoint: str, agent: str) -> str:
74 parsed = urlsplit(endpoint)
75 path = (
76 f"{parsed.path}/agents/{quote(agent, safe='')}"
77 "/endpoint/protocols/voice/live/sessions"
78 )
79 return urlunsplit(("wss", parsed.netloc, path, f"api-version={API_VERSION}", ""))
80
81
82def fetch_json(url: str, token: str) -> dict[str, object]:
83 request = Request(
84 url,
85 headers={
86 "Authorization": "{} {}".format("Bearer", token),
87 "Foundry-Features": PREVIEW_HEADER,
88 },
89 )
90 try:
91 with urlopen(request, timeout=30) as response:
92 return json.loads(response.read().decode("utf-8"))
93 except HTTPError as exc:
94 body = exc.read().decode("utf-8", errors="replace")
95 raise SmokeError(f"HTTP {exc.code}: {body}") from exc
96 except (URLError, OSError, json.JSONDecodeError) as exc:
97 raise SmokeError(f"Foundry request failed: {exc}") from exc
98
99
100def resolve_agent(endpoint: str, agent: str, token: str) -> tuple[str, str, str]:
101 agent_url = f"{endpoint}/agents/{quote(agent, safe='')}"
102 payload = fetch_json(f"{agent_url}?api-version={API_VERSION}", token)
103 versions = payload.get("versions")
104 latest = versions.get("latest") if isinstance(versions, dict) else None
105 definition = latest.get("definition") if isinstance(latest, dict) else None
106 if not isinstance(definition, dict):
107 raise SmokeError("Agent response did not include versions.latest.definition")
108 if definition.get("sub_protocol") != "live" or definition.get(
109 "model_type"
110 ) != "self_deployed":
111 raise SmokeError(
112 "Agent must use sub_protocol=live and model_type=self_deployed"
113 )
114 model = definition.get("model")
115 if not isinstance(model, str) or not model.strip():
116 raise SmokeError("Agent definition must reference a model deployment")
117 version = str(latest.get("version", "")).strip()
118 if not version:
119 raise SmokeError("Agent response did not include the latest version")
120 return agent_url, version, model.strip()
121
122
123async def receive(events: object, evidence: Evidence) -> None:
124 async for event in events:
125 kind = event.type
126 evidence.event_types.append(kind)
127 if kind == "error":
128 raise SmokeError(f"GPT-Live error: {event.error.message}")
129 if kind == "session.closed":
130 evidence.close_reason = event.reason
131 if event.reason != "close_requested":
132 raise SmokeError(
133 "GPT-Live session closed unexpectedly: "
134 f"{event.reason}"
135 )
136 evidence.closed = True
137 return
138 raise SmokeError("WebSocket ended without session.closed")
139
140
141async def run_smoke(args: argparse.Namespace, endpoint: str) -> int:
142 try:
143 import openai
144 from openai import AsyncOpenAI
145 except ImportError as exc:
146 raise SmokeError(
147 f"Missing GPT-Live SDK dependency ({exc}). Install with: "
148 'python -m pip install "openai[realtime]==3.19.2"'
149 ) from exc
150 if openai.__version__ != "3.19.2":
151 raise SmokeError(
152 "GPT-Live smoke testing requires openai[realtime]==3.19.2"
153 )
154
155 token = acquire_token()
156 agent_url, version, model = resolve_agent(endpoint, args.agent_name, token)
157 evidence = Evidence()
158
159 async with asyncio.timeout(args.timeout):
160 async with AsyncOpenAI(
161 api_key=token,
162 base_url=f"{agent_url}/endpoint/protocols/voice",
163 max_retries=0,
164 ) as client:
165 async with client.live.connect(
166 extra_headers={"Foundry-Features": PREVIEW_HEADER},
167 extra_query={
168 "api-version": API_VERSION,
169 "x-agent-version-override": version,
170 },
171 max_retries=0,
172 websocket_connection_options={
173 "open_timeout": 20,
174 "close_timeout": 5,
175 "max_size": 1024 * 1024,
176 },
177 ) as connection:
178 events = connection.__aiter__()
179 async for event in events:
180 evidence.event_types.append(event.type)
181 if event.type == "error":
182 raise SmokeError(f"GPT-Live startup error: {event.error.message}")
183 if event.type == "session.started":
184 break
185 if event.type == "session.closed":
186 raise SmokeError("Session closed before session.started")
187 else:
188 raise SmokeError("WebSocket ended before session.started")
189
190 receiver = asyncio.create_task(receive(events, evidence))
191 try:
192 await connection.session.close()
193 await asyncio.gather(receiver)
194 if not evidence.closed:
195 raise SmokeError("Session did not emit session.closed")
196 finally:
197 if not receiver.done():
198 receiver.cancel()
199 await asyncio.gather(receiver, return_exceptions=True)
200
201 print(
202 json.dumps(
203 {
204 "agent": args.agent_name,
205 "agent_version": version,
206 "model_deployment": model,
207 "session_started": True,
208 "session_closed": evidence.closed,
209 "close_reason": evidence.close_reason,
210 "event_types": evidence.event_types,
211 },
212 indent=2,
213 )
214 )
215 return 0
216
217
218def main() -> int:
219 if sys.version_info < (3, 11):
220 print("ERROR: Python 3.11 or later is required.", file=sys.stderr)
221 return 1
222 args = parse_args()
223 try:
224 endpoint = normalize_endpoint(args.endpoint)
225 if args.dry_run:
226 print("DRY-RUN: GPT-Live Voice Agent Smoke Test")
227 print(f" Project Endpoint : {endpoint}")
228 print(f" Agent Name : {args.agent_name}")
229 print(f" WebSocket URL : {websocket_url(endpoint, args.agent_name)}")
230 print(f" Timeout (s) : {args.timeout}")
231 print(f" Header : Foundry-Features: {PREVIEW_HEADER}")
232 print(" Auth Resource : https://ai.azure.com")
233 return 0
234 return asyncio.run(run_smoke(args, endpoint))
235 except TimeoutError:
236 print(f"ERROR: Timed out after {args.timeout}s.", file=sys.stderr)
237 return 1
238 except SmokeError as exc:
239 print(f"ERROR: {exc}", file=sys.stderr)
240 return 1
241 except Exception as exc:
242 print(
243 f"ERROR: Unexpected GPT-Live failure ({type(exc).__name__}): {exc}",
244 file=sys.stderr,
245 )
246 return 1
247
248
249if __name__ == "__main__":
250 sys.exit(main())