# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ....core.api_error import ApiError from ....core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ....core.http_response import AsyncHttpResponse, HttpResponse from ....core.jsonable_encoder import jsonable_encoder from ....core.request_options import RequestOptions from ....core.serialization import convert_and_respect_annotation_metadata from ....core.unchecked_base_model import construct_type from ....errors.unprocessable_entity_error import UnprocessableEntityError from ....types.adhoc_agent_config_override_for_test_request_model import AdhocAgentConfigOverrideForTestRequestModel from ....types.get_test_invocations_page_response_model import GetTestInvocationsPageResponseModel from ....types.get_test_suite_invocation_response_model import GetTestSuiteInvocationResponseModel from ....types.http_validation_error import HttpValidationError # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawInvocationsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, agent_id: str, page_size: typing.Optional[int] = None, cursor: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[GetTestInvocationsPageResponseModel]: """ Lists all test invocations with pagination support and optional search filtering. Parameters ---------- agent_id : str Filter by agent ID page_size : typing.Optional[int] How many Tests to return at maximum. Can not exceed 100, defaults to 30. cursor : typing.Optional[str] Used for fetching next page. Cursor is returned in the response. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetTestInvocationsPageResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/test-invocations", method="GET", params={ "agent_id": agent_id, "page_size": page_size, "cursor": cursor, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetTestInvocationsPageResponseModel, construct_type( type_=GetTestInvocationsPageResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get( self, test_invocation_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[GetTestSuiteInvocationResponseModel]: """ Gets a test invocation by ID. Parameters ---------- test_invocation_id : str The id of a test invocation. This is returned when tests are run. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetTestSuiteInvocationResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/test-invocations/{jsonable_encoder(test_invocation_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetTestSuiteInvocationResponseModel, construct_type( type_=GetTestSuiteInvocationResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def resubmit( self, test_invocation_id: str, *, test_run_ids: typing.Sequence[str], agent_id: str, agent_config_override: typing.Optional[AdhocAgentConfigOverrideForTestRequestModel] = OMIT, branch_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.Any]: """ Resubmits specific test runs from a test invocation. Parameters ---------- test_invocation_id : str The id of a test invocation. This is returned when tests are run. test_run_ids : typing.Sequence[str] List of test run IDs to resubmit agent_id : str Agent ID to resubmit tests for agent_config_override : typing.Optional[AdhocAgentConfigOverrideForTestRequestModel] Configuration overrides to use for testing. If not provided, the agent's default configuration will be used. branch_id : typing.Optional[str] ID of the branch to run the tests on. If not provided, the tests will be run on the agent default configuration. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/test-invocations/{jsonable_encoder(test_invocation_id)}/resubmit", method="POST", json={ "test_run_ids": test_run_ids, "agent_config_override": convert_and_respect_annotation_metadata( object_=agent_config_override, annotation=AdhocAgentConfigOverrideForTestRequestModel, direction="write", ), "agent_id": agent_id, "branch_id": branch_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if _response is None or not _response.text.strip(): return HttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawInvocationsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, agent_id: str, page_size: typing.Optional[int] = None, cursor: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[GetTestInvocationsPageResponseModel]: """ Lists all test invocations with pagination support and optional search filtering. Parameters ---------- agent_id : str Filter by agent ID page_size : typing.Optional[int] How many Tests to return at maximum. Can not exceed 100, defaults to 30. cursor : typing.Optional[str] Used for fetching next page. Cursor is returned in the response. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetTestInvocationsPageResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/test-invocations", method="GET", params={ "agent_id": agent_id, "page_size": page_size, "cursor": cursor, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetTestInvocationsPageResponseModel, construct_type( type_=GetTestInvocationsPageResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get( self, test_invocation_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[GetTestSuiteInvocationResponseModel]: """ Gets a test invocation by ID. Parameters ---------- test_invocation_id : str The id of a test invocation. This is returned when tests are run. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetTestSuiteInvocationResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/test-invocations/{jsonable_encoder(test_invocation_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetTestSuiteInvocationResponseModel, construct_type( type_=GetTestSuiteInvocationResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def resubmit( self, test_invocation_id: str, *, test_run_ids: typing.Sequence[str], agent_id: str, agent_config_override: typing.Optional[AdhocAgentConfigOverrideForTestRequestModel] = OMIT, branch_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.Any]: """ Resubmits specific test runs from a test invocation. Parameters ---------- test_invocation_id : str The id of a test invocation. This is returned when tests are run. test_run_ids : typing.Sequence[str] List of test run IDs to resubmit agent_id : str Agent ID to resubmit tests for agent_config_override : typing.Optional[AdhocAgentConfigOverrideForTestRequestModel] Configuration overrides to use for testing. If not provided, the agent's default configuration will be used. branch_id : typing.Optional[str] ID of the branch to run the tests on. If not provided, the tests will be run on the agent default configuration. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.Any] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/test-invocations/{jsonable_encoder(test_invocation_id)}/resubmit", method="POST", json={ "test_run_ids": test_run_ids, "agent_config_override": convert_and_respect_annotation_metadata( object_=agent_config_override, annotation=AdhocAgentConfigOverrideForTestRequestModel, direction="write", ), "agent_id": agent_id, "branch_id": branch_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if _response is None or not _response.text.strip(): return AsyncHttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)