Setting the file. One moment. foundry-agent/create/scripts/smoke_test_voice.py
Python·438 lines·15 KB
import
os
16from pathlib import Path
17import shutil
18import subprocess
19import sys
20import time
21import uuid
22
23GREETING_GRACE_SECONDS = 1.0
24DEFAULT_PROMPT = "In one short sentence, introduce yourself."
25REQUEST_METADATA_KEY = "foundry_voice_smoke_test_request"
26
27
28def positive_integer(value: str) -> int:
29 try:
30 parsed = int(value)
31 except ValueError as exc:
32 raise argparse.ArgumentTypeError("must be a positive integer") from exc
33 if parsed <= 0:
34 raise argparse.ArgumentTypeError("must be a positive integer")
35 return parsed
36
37
38def parse_args() -> argparse.Namespace:
39 parser = argparse.ArgumentParser(
40 description="Smoke-test a deployed Microsoft Foundry Voice Agent."
41 )
42 parser.add_argument(
43 "--endpoint",
44 required=True,
45 help="Foundry project endpoint or full managed Voice WebSocket URL",
46 )
47 parser.add_argument("--agent-name", required=True, help="Deployed Voice Agent name")
48 parser.add_argument("--text", default=DEFAULT_PROMPT, help="Prompt text to send")
49 parser.add_argument(
50 "--expected-marker",
51 default="",
52 help="Substring that must appear in the requested turn transcript",
53 )
54 parser.add_argument(
55 "--timeout",
56 type=positive_integer,
57 default=180,
58 help="Positive timeout in seconds (default: 180)",
59 )
60 parser.add_argument(
61 "--output-audio",
62 default="",
63 help="Optional path for decoded PCM audio bytes",
64 )
65 parser.add_argument(
66 "--dry-run",
67 action="store_true",
68 help="Print the resolved execution plan without connecting",
69 )
70 args = parser.parse_args()
71
72 if not args.endpoint.strip():
73 parser.error("--endpoint must not be empty")
74 if not args.agent_name.strip():
75 parser.error("--agent-name must not be empty")
76 if not args.text.strip():
77 parser.error("--text must not be empty")
78 if args.expected_marker is not None and not args.expected_marker.strip():
79 args.expected_marker = ""
80
81 return args
82
83
84def build_ws_url(endpoint: str, agent_name: str) -> str:
85 trimmed = endpoint.rstrip("/")
86 if trimmed.startswith("https://"):
87 ws_base = f"wss://{trimmed[len('https://'):]}"
88 elif trimmed.startswith("wss://"):
89 ws_base = trimmed
90 else:
91 raise ValueError("--endpoint must start with https:// or wss://")
92
93 if "/protocols/voice" in ws_base:
94 return ws_base
95 return (
96 f"{ws_base}/agents/{agent_name}/endpoint/protocols/voice"
97 "?api-version=v1"
98 )
99
100
101def acquire_token() -> str:
102 errors: list[str] = []
103 az_bin = shutil.which("az.cmd" if os.name == "nt" else "az")
104 if az_bin:
105 try:
106 token = subprocess.check_output(
107 [
108 az_bin,
109 "account",
110 "get-access-token",
111 "--resource",
112 "https://ai.azure.com",
113 "--query",
114 "accessToken",
115 "-o",
116 "tsv",
117 ],
118 text=True,
119 stderr=subprocess.STDOUT,
120 ).strip()
121 if token:
122 return token
123 errors.append("Azure CLI returned an empty token")
124 except (OSError, subprocess.CalledProcessError) as exc:
125 output = getattr(exc, "output", "")
126 errors.append(f"Azure CLI token acquisition failed: {str(output).strip() or exc}")
127 else:
128 errors.append("Azure CLI executable was not found")
129
130 try:
131 from azure.identity import DefaultAzureCredential
132
133 return DefaultAzureCredential().get_token(
134 "https://ai.azure.com/.default"
135 ).token
136 except Exception as exc:
137 errors.append(f"DefaultAzureCredential failed: {exc}")
138
139 raise RuntimeError("; ".join(errors))
140
141
142async def send_user_turn(ws: object, text: str, request_marker: str) -> None:
143 await ws.send(
144 json.dumps(
145 {
146 "type": "conversation.item.create",
147 "item": {
148 "type": "message",
149 "role": "user",
150 "content": [{"type": "input_text", "text": text}],
151 },
152 }
153 )
154 )
155 await ws.send(
156 json.dumps(
157 {
158 "type": "response.create",
159 "response": {
160 "metadata": {REQUEST_METADATA_KEY: request_marker},
161 },
162 }
163 )
164 )
165
166
167def response_id(event: dict[str, object]) -> str:
168 direct_id = event.get("response_id")
169 if isinstance(direct_id, str):
170 return direct_id
171 response = event.get("response")
172 if isinstance(response, dict):
173 nested_id = response.get("id")
174 if isinstance(nested_id, str):
175 return nested_id
176 return ""
177
178
179def response_has_request_marker(
180 event: dict[str, object], request_marker: str
181) -> bool:
182 response = event.get("response")
183 if not isinstance(response, dict):
184 return False
185 metadata = response.get("metadata")
186 return (
187 isinstance(metadata, dict)
188 and metadata.get(REQUEST_METADATA_KEY) == request_marker
189 )
190
191
192async def run_smoke_test(args: argparse.Namespace, ws_url_base: str) -> int:
193 try:
194 import websockets
195 except ImportError as exc:
196 print(
197 f"ERROR: Missing WebSocket client dependency ({exc}). "
198 "Install with: python -m pip install websockets",
199 file=sys.stderr,
200 )
201 return 1
202
203 try:
204 token = acquire_token()
205 except Exception as exc:
206 print(
207 "ERROR: Failed to acquire an Entra token for https://ai.azure.com: "
208 f"{exc}",
209 file=sys.stderr,
210 )
211 return 1
212
213 separator = "&" if "?" in ws_url_base else "?"
214 uri = (
215 ws_url_base
216 if "agent_session_id=" in ws_url_base
217 else f"{ws_url_base}{separator}agent_session_id={uuid.uuid4()}"
218 )
219 headers = {
220 "Authorization": f"Bearer {token}",
221 "Foundry-Features": "VoiceAgents=V1Preview",
222 }
223
224 events: list[str] = []
225 transcript: list[str] = []
226 final_transcript = ""
227 audio_base64_characters = 0
228 decoded_audio = bytearray()
229 session_updated = False
230 greeting_in_progress = False
231 grace_deadline: float | None = None
232 sent_turn = False
233 request_marker = uuid.uuid4().hex
234 requested_response_id = ""
235 completed_status: str | None = None
236
237 try:
238 try:
239 ws_context = websockets.connect(
240 uri,
241 additional_headers=headers,
242 open_timeout=60,
243 close_timeout=10,
244 max_size=16 * 1024 * 1024,
245 )
246 except TypeError:
247 ws_context = websockets.connect(
248 uri,
249 extra_headers=headers,
250 open_timeout=60,
251 close_timeout=10,
252 max_size=16 * 1024 * 1024,
253 )
254
255 async with ws_context as ws:
256 while True:
257 if not sent_turn:
258 if (
259 session_updated
260 and not greeting_in_progress
261 and grace_deadline is not None
262 ):
263 remaining_grace = grace_deadline - time.monotonic()
264 if remaining_grace <= 0:
265 await send_user_turn(ws, args.text, request_marker)
266 sent_turn = True
267 continue
268 receive_timeout = min(float(args.timeout), remaining_grace)
269 else:
270 receive_timeout = float(args.timeout)
271
272 try:
273 raw_message = await asyncio.wait_for(
274 ws.recv(), timeout=receive_timeout
275 )
276 except (TimeoutError, asyncio.TimeoutError):
277 if session_updated and not greeting_in_progress:
278 await send_user_turn(ws, args.text, request_marker)
279 sent_turn = True
280 continue
281 raise
282
283 event = json.loads(raw_message)
284 event_type = event.get("type", "")
285 events.append(event_type)
286
287 if event_type == "error":
288 print(
289 "ERROR: Voice session returned an error event: "
290 f"{json.dumps(event)}",
291 file=sys.stderr,
292 )
293 return 1
294 if event_type == "session.updated":
295 session_updated = True
296 if not greeting_in_progress:
297 grace_deadline = (
298 time.monotonic() + GREETING_GRACE_SECONDS
299 )
300 elif event_type == "response.created" or event_type.startswith(
301 "response.output_"
302 ):
303 greeting_in_progress = True
304 grace_deadline = None
305 elif event_type == "response.done" and greeting_in_progress:
306 greeting_in_progress = False
307 grace_deadline = None
308 await send_user_turn(ws, args.text, request_marker)
309 sent_turn = True
310 continue
311
312 raw_message = await asyncio.wait_for(
313 ws.recv(), timeout=float(args.timeout)
314 )
315 event = json.loads(raw_message)
316 event_type = event.get("type", "")
317 events.append(event_type)
318
319 if event_type == "error":
320 print(
321 "ERROR: Voice session returned an error event: "
322 f"{json.dumps(event)}",
323 file=sys.stderr,
324 )
325 return 1
326 if event_type == "response.created":
327 if response_has_request_marker(event, request_marker):
328 requested_response_id = response_id(event)
329 if not requested_response_id:
330 print(
331 "ERROR: Requested Voice Agent response did not "
332 "include a response ID.",
333 file=sys.stderr,
334 )
335 return 1
336 continue
337 if (
338 not requested_response_id
339 or response_id(event) != requested_response_id
340 ):
341 continue
342
343 if event_type == "response.output_audio_transcript.delta":
344 transcript.append(event.get("delta", ""))
345 elif event_type == "response.output_audio_transcript.done":
346 final_transcript = event.get("transcript", "")
347 elif event_type == "response.output_audio.delta":
348 encoded_delta = event.get("delta", "")
349 audio_base64_characters += len(encoded_delta)
350 if args.output_audio and encoded_delta:
351 decoded_audio.extend(base64.b64decode(encoded_delta))
352 elif event_type == "response.done":
353 completed_status = event.get("response", {}).get("status")
354 break
355 except (TimeoutError, asyncio.TimeoutError):
356 print(
357 f"ERROR: Timed out after {args.timeout}s waiting for Voice Agent "
358 f"'{args.agent_name}'.",
359 file=sys.stderr,
360 )
361 return 1
362 except Exception as exc:
363 print(
364 f"ERROR: Voice WebSocket smoke test failed for "
365 f"'{args.agent_name}': {exc}",
366 file=sys.stderr,
367 )
368 return 1
369
370 text = (final_transcript or "".join(transcript)).strip()
371 result = {
372 "agent": args.agent_name,
373 "prompt_text": args.text,
374 "response_id": requested_response_id,
375 "status": completed_status,
376 "transcript": text,
377 "audio_base64_characters": audio_base64_characters,
378 "event_types": events,
379 }
380 if args.expected_marker:
381 result["expected_marker"] = args.expected_marker
382
383 if completed_status != "completed" or not text or audio_base64_characters <= 0:
384 print(
385 f"ERROR: Incomplete Voice Agent response: {json.dumps(result)}",
386 file=sys.stderr,
387 )
388 return 1
389
390 if args.expected_marker and args.expected_marker not in text:
391 print(
392 "ERROR: Voice Agent transcript did not contain expected marker "
393 f"{args.expected_marker!r}: {json.dumps(result)}",
394 file=sys.stderr,
395 )
396 return 1
397
398 if args.output_audio:
399 try:
400 Path(args.output_audio).write_bytes(decoded_audio)
401 except OSError as exc:
402 print(
403 f"ERROR: Failed to write audio output '{args.output_audio}': {exc}",
404 file=sys.stderr,
405 )
406 return 1
407
408 print(json.dumps(result, indent=2))
409 return 0
410
411
412def main() -> int:
413 args = parse_args()
414 try:
415 ws_url = build_ws_url(args.endpoint, args.agent_name)
416 except ValueError as exc:
417 print(f"ERROR: {exc}", file=sys.stderr)
418 return 2
419
420 if args.dry_run:
421 separator = "&" if "?" in ws_url else "?"
422 print("DRY-RUN: Voice Agent WebSocket Smoke Test")
423 print(f" Project Endpoint : {args.endpoint}")
424 print(f" Agent Name : {args.agent_name}")
425 print(f" WebSocket URL : {ws_url}{separator}agent_session_id=<uuid>")
426 print(" Headers : Foundry-Features: VoiceAgents=V1Preview")
427 print(f" Prompt Text : {args.text}")
428 print(f" Expected Marker : {args.expected_marker or '(none)'}")
429 print(f" Timeout (s) : {args.timeout}")
430 print(f" Output Audio : {args.output_audio or '(none)'}")
431 print(" Auth Resource : https://ai.azure.com (Azure CLI or DefaultAzureCredential)")
432 return 0
433
434 return asyncio.run(run_smoke_test(args, ws_url))
435
436
437if __name__ == "__main__":
438 sys.exit(main())