(
app_conversation_id: str, agent_server_url: str, session_api_key: str
)
| 350 | |
| 351 | |
| 352 | def fetch_agent_server_events( |
| 353 | app_conversation_id: str, agent_server_url: str, session_api_key: str |
| 354 | ) -> list[dict[str, Any]]: |
| 355 | payload = request_json( |
| 356 | agent_server_url, |
| 357 | f'/api/conversations/{urllib.parse.quote(app_conversation_id)}/events/search?limit={EVENT_SEARCH_LIMIT}', |
| 358 | headers={'X-Session-API-Key': session_api_key}, |
| 359 | ) |
| 360 | if isinstance(payload, dict): |
| 361 | items = payload.get('items') |
| 362 | return validate_event_search_results(items) if isinstance(items, list) else [] |
| 363 | if isinstance(payload, list): |
| 364 | return validate_event_search_results(payload) |
| 365 | return [] |
| 366 | |
| 367 | |
| 368 | def fetch_agent_server_final_response( |
no test coverage detected