(app_conversation_id: str)
| 336 | |
| 337 | |
| 338 | def fetch_app_server_events(app_conversation_id: str) -> list[dict[str, Any]]: |
| 339 | payload = request_json( |
| 340 | OPENHANDS_BASE_URL, |
| 341 | f'/api/v1/conversation/{urllib.parse.quote(app_conversation_id)}/events/search?limit={EVENT_SEARCH_LIMIT}', |
| 342 | headers={'Authorization': openhands_headers()['Authorization']}, |
| 343 | ) |
| 344 | if isinstance(payload, dict): |
| 345 | items = payload.get('items') |
| 346 | return validate_event_search_results(items) if isinstance(items, list) else [] |
| 347 | if isinstance(payload, list): |
| 348 | return validate_event_search_results(payload) |
| 349 | return [] |
| 350 | |
| 351 | |
| 352 | def fetch_agent_server_events( |
no test coverage detected