# 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.agent_platform_settings_request_model import AgentPlatformSettingsRequestModel from ...types.agent_simulated_chat_test_response_model import AgentSimulatedChatTestResponseModel from ...types.agent_sort_by import AgentSortBy from ...types.agent_workflow_request_model import AgentWorkflowRequestModel from ...types.conversation_simulation_specification import ConversationSimulationSpecification from ...types.conversational_config import ConversationalConfig from ...types.create_agent_response_model import CreateAgentResponseModel from ...types.get_agent_response_model import GetAgentResponseModel from ...types.get_agents_page_response_model import GetAgentsPageResponseModel from ...types.get_test_suite_invocation_response_model import GetTestSuiteInvocationResponseModel from ...types.http_validation_error import HttpValidationError from ...types.prompt_evaluation_criteria import PromptEvaluationCriteria from ...types.single_test_run_request_model import SingleTestRunRequestModel from ...types.sort_direction import SortDirection # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawAgentsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def create( self, *, conversation_config: ConversationalConfig, platform_settings: typing.Optional[AgentPlatformSettingsRequestModel] = OMIT, workflow: typing.Optional[AgentWorkflowRequestModel] = OMIT, name: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[CreateAgentResponseModel]: """ Create an agent from a config object Parameters ---------- conversation_config : ConversationalConfig Conversation configuration for an agent platform_settings : typing.Optional[AgentPlatformSettingsRequestModel] Platform settings for the agent are all settings that aren't related to the conversation orchestration and content. workflow : typing.Optional[AgentWorkflowRequestModel] Workflow for the agent. This is used to define the flow of the conversation and how the agent interacts with tools. name : typing.Optional[str] A name to make the agent easier to find tags : typing.Optional[typing.Sequence[str]] Tags to help classify and filter the agent request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[CreateAgentResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/agents/create", method="POST", json={ "conversation_config": convert_and_respect_annotation_metadata( object_=conversation_config, annotation=ConversationalConfig, direction="write" ), "platform_settings": convert_and_respect_annotation_metadata( object_=platform_settings, annotation=AgentPlatformSettingsRequestModel, direction="write" ), "workflow": convert_and_respect_annotation_metadata( object_=workflow, annotation=AgentWorkflowRequestModel, direction="write" ), "name": name, "tags": tags, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentResponseModel, construct_type( type_=CreateAgentResponseModel, # 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, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[GetAgentResponseModel]: """ Retrieve config for an agent Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetAgentResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetAgentResponseModel, construct_type( type_=GetAgentResponseModel, # 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 delete(self, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]: """ Delete an agent Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) 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 update( self, agent_id: str, *, conversation_config: typing.Optional[ConversationalConfig] = OMIT, platform_settings: typing.Optional[AgentPlatformSettingsRequestModel] = OMIT, workflow: typing.Optional[AgentWorkflowRequestModel] = OMIT, name: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[GetAgentResponseModel]: """ Patches an Agent settings Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. conversation_config : typing.Optional[ConversationalConfig] Conversation configuration for an agent platform_settings : typing.Optional[AgentPlatformSettingsRequestModel] Platform settings for the agent are all settings that aren't related to the conversation orchestration and content. workflow : typing.Optional[AgentWorkflowRequestModel] Workflow for the agent. This is used to define the flow of the conversation and how the agent interacts with tools. name : typing.Optional[str] A name to make the agent easier to find tags : typing.Optional[typing.Sequence[str]] Tags to help classify and filter the agent request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetAgentResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}", method="PATCH", json={ "conversation_config": convert_and_respect_annotation_metadata( object_=conversation_config, annotation=ConversationalConfig, direction="write" ), "platform_settings": convert_and_respect_annotation_metadata( object_=platform_settings, annotation=AgentPlatformSettingsRequestModel, direction="write" ), "workflow": convert_and_respect_annotation_metadata( object_=workflow, annotation=AgentWorkflowRequestModel, direction="write" ), "name": name, "tags": tags, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetAgentResponseModel, construct_type( type_=GetAgentResponseModel, # 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 list( self, *, page_size: typing.Optional[int] = None, search: typing.Optional[str] = None, archived: typing.Optional[bool] = None, sort_direction: typing.Optional[SortDirection] = None, sort_by: typing.Optional[AgentSortBy] = None, cursor: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[GetAgentsPageResponseModel]: """ Returns a list of your agents and their metadata. Parameters ---------- page_size : typing.Optional[int] How many Agents to return at maximum. Can not exceed 100, defaults to 30. search : typing.Optional[str] Search by agents name. archived : typing.Optional[bool] Filter agents by archived status sort_direction : typing.Optional[SortDirection] The direction to sort the results sort_by : typing.Optional[AgentSortBy] The field to sort the results by 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[GetAgentsPageResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/agents", method="GET", params={ "page_size": page_size, "search": search, "archived": archived, "sort_direction": sort_direction, "sort_by": sort_by, "cursor": cursor, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetAgentsPageResponseModel, construct_type( type_=GetAgentsPageResponseModel, # 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 duplicate( self, agent_id: str, *, name: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[CreateAgentResponseModel]: """ Create a new agent by duplicating an existing one Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. name : typing.Optional[str] A name to make the agent easier to find request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[CreateAgentResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/duplicate", method="POST", json={ "name": name, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentResponseModel, construct_type( type_=CreateAgentResponseModel, # 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 simulate_conversation( self, agent_id: str, *, simulation_specification: ConversationSimulationSpecification, extra_evaluation_criteria: typing.Optional[typing.Sequence[PromptEvaluationCriteria]] = OMIT, new_turns_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentSimulatedChatTestResponseModel]: """ Run a conversation between the agent and a simulated user. Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. simulation_specification : ConversationSimulationSpecification A specification detailing how the conversation should be simulated extra_evaluation_criteria : typing.Optional[typing.Sequence[PromptEvaluationCriteria]] A list of evaluation criteria to test new_turns_limit : typing.Optional[int] Maximum number of new turns to generate in the conversation simulation request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentSimulatedChatTestResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/simulate-conversation", method="POST", json={ "simulation_specification": convert_and_respect_annotation_metadata( object_=simulation_specification, annotation=ConversationSimulationSpecification, direction="write" ), "extra_evaluation_criteria": convert_and_respect_annotation_metadata( object_=extra_evaluation_criteria, annotation=typing.Sequence[PromptEvaluationCriteria], direction="write", ), "new_turns_limit": new_turns_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentSimulatedChatTestResponseModel, construct_type( type_=AgentSimulatedChatTestResponseModel, # 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 simulate_conversation_stream( self, agent_id: str, *, simulation_specification: ConversationSimulationSpecification, extra_evaluation_criteria: typing.Optional[typing.Sequence[PromptEvaluationCriteria]] = OMIT, new_turns_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Run a conversation between the agent and a simulated user and stream back the response. Response is streamed back as partial lists of messages that should be concatenated and once the conversation has complete a single final message with the conversation analysis will be sent. Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. simulation_specification : ConversationSimulationSpecification A specification detailing how the conversation should be simulated extra_evaluation_criteria : typing.Optional[typing.Sequence[PromptEvaluationCriteria]] A list of evaluation criteria to test new_turns_limit : typing.Optional[int] Maximum number of new turns to generate in the conversation simulation request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/simulate-conversation/stream", method="POST", json={ "simulation_specification": convert_and_respect_annotation_metadata( object_=simulation_specification, annotation=ConversationSimulationSpecification, direction="write" ), "extra_evaluation_criteria": convert_and_respect_annotation_metadata( object_=extra_evaluation_criteria, annotation=typing.Sequence[PromptEvaluationCriteria], direction="write", ), "new_turns_limit": new_turns_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) 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 run_tests( self, agent_id: str, *, tests: typing.Sequence[SingleTestRunRequestModel], agent_config_override: typing.Optional[AdhocAgentConfigOverrideForTestRequestModel] = OMIT, branch_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[GetTestSuiteInvocationResponseModel]: """ Run selected tests on the agent with provided configuration. If the agent configuration is provided, it will be used to override default agent configuration. Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. tests : typing.Sequence[SingleTestRunRequestModel] List of tests to run on the agent 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[GetTestSuiteInvocationResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/run-tests", method="POST", json={ "tests": convert_and_respect_annotation_metadata( object_=tests, annotation=typing.Sequence[SingleTestRunRequestModel], direction="write" ), "agent_config_override": convert_and_respect_annotation_metadata( object_=agent_config_override, annotation=AdhocAgentConfigOverrideForTestRequestModel, direction="write", ), "branch_id": branch_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) 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) class AsyncRawAgentsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def create( self, *, conversation_config: ConversationalConfig, platform_settings: typing.Optional[AgentPlatformSettingsRequestModel] = OMIT, workflow: typing.Optional[AgentWorkflowRequestModel] = OMIT, name: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[CreateAgentResponseModel]: """ Create an agent from a config object Parameters ---------- conversation_config : ConversationalConfig Conversation configuration for an agent platform_settings : typing.Optional[AgentPlatformSettingsRequestModel] Platform settings for the agent are all settings that aren't related to the conversation orchestration and content. workflow : typing.Optional[AgentWorkflowRequestModel] Workflow for the agent. This is used to define the flow of the conversation and how the agent interacts with tools. name : typing.Optional[str] A name to make the agent easier to find tags : typing.Optional[typing.Sequence[str]] Tags to help classify and filter the agent request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[CreateAgentResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/agents/create", method="POST", json={ "conversation_config": convert_and_respect_annotation_metadata( object_=conversation_config, annotation=ConversationalConfig, direction="write" ), "platform_settings": convert_and_respect_annotation_metadata( object_=platform_settings, annotation=AgentPlatformSettingsRequestModel, direction="write" ), "workflow": convert_and_respect_annotation_metadata( object_=workflow, annotation=AgentWorkflowRequestModel, direction="write" ), "name": name, "tags": tags, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentResponseModel, construct_type( type_=CreateAgentResponseModel, # 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, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[GetAgentResponseModel]: """ Retrieve config for an agent Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetAgentResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetAgentResponseModel, construct_type( type_=GetAgentResponseModel, # 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 delete( self, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete an agent Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) 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 update( self, agent_id: str, *, conversation_config: typing.Optional[ConversationalConfig] = OMIT, platform_settings: typing.Optional[AgentPlatformSettingsRequestModel] = OMIT, workflow: typing.Optional[AgentWorkflowRequestModel] = OMIT, name: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[GetAgentResponseModel]: """ Patches an Agent settings Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. conversation_config : typing.Optional[ConversationalConfig] Conversation configuration for an agent platform_settings : typing.Optional[AgentPlatformSettingsRequestModel] Platform settings for the agent are all settings that aren't related to the conversation orchestration and content. workflow : typing.Optional[AgentWorkflowRequestModel] Workflow for the agent. This is used to define the flow of the conversation and how the agent interacts with tools. name : typing.Optional[str] A name to make the agent easier to find tags : typing.Optional[typing.Sequence[str]] Tags to help classify and filter the agent request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetAgentResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}", method="PATCH", json={ "conversation_config": convert_and_respect_annotation_metadata( object_=conversation_config, annotation=ConversationalConfig, direction="write" ), "platform_settings": convert_and_respect_annotation_metadata( object_=platform_settings, annotation=AgentPlatformSettingsRequestModel, direction="write" ), "workflow": convert_and_respect_annotation_metadata( object_=workflow, annotation=AgentWorkflowRequestModel, direction="write" ), "name": name, "tags": tags, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetAgentResponseModel, construct_type( type_=GetAgentResponseModel, # 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 list( self, *, page_size: typing.Optional[int] = None, search: typing.Optional[str] = None, archived: typing.Optional[bool] = None, sort_direction: typing.Optional[SortDirection] = None, sort_by: typing.Optional[AgentSortBy] = None, cursor: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[GetAgentsPageResponseModel]: """ Returns a list of your agents and their metadata. Parameters ---------- page_size : typing.Optional[int] How many Agents to return at maximum. Can not exceed 100, defaults to 30. search : typing.Optional[str] Search by agents name. archived : typing.Optional[bool] Filter agents by archived status sort_direction : typing.Optional[SortDirection] The direction to sort the results sort_by : typing.Optional[AgentSortBy] The field to sort the results by 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[GetAgentsPageResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/agents", method="GET", params={ "page_size": page_size, "search": search, "archived": archived, "sort_direction": sort_direction, "sort_by": sort_by, "cursor": cursor, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetAgentsPageResponseModel, construct_type( type_=GetAgentsPageResponseModel, # 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 duplicate( self, agent_id: str, *, name: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[CreateAgentResponseModel]: """ Create a new agent by duplicating an existing one Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. name : typing.Optional[str] A name to make the agent easier to find request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[CreateAgentResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/duplicate", method="POST", json={ "name": name, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentResponseModel, construct_type( type_=CreateAgentResponseModel, # 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 simulate_conversation( self, agent_id: str, *, simulation_specification: ConversationSimulationSpecification, extra_evaluation_criteria: typing.Optional[typing.Sequence[PromptEvaluationCriteria]] = OMIT, new_turns_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentSimulatedChatTestResponseModel]: """ Run a conversation between the agent and a simulated user. Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. simulation_specification : ConversationSimulationSpecification A specification detailing how the conversation should be simulated extra_evaluation_criteria : typing.Optional[typing.Sequence[PromptEvaluationCriteria]] A list of evaluation criteria to test new_turns_limit : typing.Optional[int] Maximum number of new turns to generate in the conversation simulation request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentSimulatedChatTestResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/simulate-conversation", method="POST", json={ "simulation_specification": convert_and_respect_annotation_metadata( object_=simulation_specification, annotation=ConversationSimulationSpecification, direction="write" ), "extra_evaluation_criteria": convert_and_respect_annotation_metadata( object_=extra_evaluation_criteria, annotation=typing.Sequence[PromptEvaluationCriteria], direction="write", ), "new_turns_limit": new_turns_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentSimulatedChatTestResponseModel, construct_type( type_=AgentSimulatedChatTestResponseModel, # 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 simulate_conversation_stream( self, agent_id: str, *, simulation_specification: ConversationSimulationSpecification, extra_evaluation_criteria: typing.Optional[typing.Sequence[PromptEvaluationCriteria]] = OMIT, new_turns_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Run a conversation between the agent and a simulated user and stream back the response. Response is streamed back as partial lists of messages that should be concatenated and once the conversation has complete a single final message with the conversation analysis will be sent. Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. simulation_specification : ConversationSimulationSpecification A specification detailing how the conversation should be simulated extra_evaluation_criteria : typing.Optional[typing.Sequence[PromptEvaluationCriteria]] A list of evaluation criteria to test new_turns_limit : typing.Optional[int] Maximum number of new turns to generate in the conversation simulation request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/simulate-conversation/stream", method="POST", json={ "simulation_specification": convert_and_respect_annotation_metadata( object_=simulation_specification, annotation=ConversationSimulationSpecification, direction="write" ), "extra_evaluation_criteria": convert_and_respect_annotation_metadata( object_=extra_evaluation_criteria, annotation=typing.Sequence[PromptEvaluationCriteria], direction="write", ), "new_turns_limit": new_turns_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) 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 run_tests( self, agent_id: str, *, tests: typing.Sequence[SingleTestRunRequestModel], agent_config_override: typing.Optional[AdhocAgentConfigOverrideForTestRequestModel] = OMIT, branch_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[GetTestSuiteInvocationResponseModel]: """ Run selected tests on the agent with provided configuration. If the agent configuration is provided, it will be used to override default agent configuration. Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. tests : typing.Sequence[SingleTestRunRequestModel] List of tests to run on the agent 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[GetTestSuiteInvocationResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/run-tests", method="POST", json={ "tests": convert_and_respect_annotation_metadata( object_=tests, annotation=typing.Sequence[SingleTestRunRequestModel], direction="write" ), "agent_config_override": convert_and_respect_annotation_metadata( object_=agent_config_override, annotation=AdhocAgentConfigOverrideForTestRequestModel, direction="write", ), "branch_id": branch_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) 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)