# 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.agent_branch_response import AgentBranchResponse from ....types.agent_workflow_request_model import AgentWorkflowRequestModel from ....types.branch_protection_status import BranchProtectionStatus from ....types.create_agent_branch_response_model import CreateAgentBranchResponseModel from ....types.http_validation_error import HttpValidationError from ....types.list_response_agent_branch_summary import ListResponseAgentBranchSummary # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawBranchesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, agent_id: str, *, include_archived: typing.Optional[bool] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ListResponseAgentBranchSummary]: """ Returns a list of branches an agent has Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. include_archived : typing.Optional[bool] Whether archived branches should be included limit : typing.Optional[int] How many results at most should be returned request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ListResponseAgentBranchSummary] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches", method="GET", params={ "include_archived": include_archived, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListResponseAgentBranchSummary, construct_type( type_=ListResponseAgentBranchSummary, # 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 create( self, agent_id: str, *, parent_version_id: str, name: str, description: str, conversation_config: typing.Optional[typing.Dict[str, typing.Any]] = OMIT, platform_settings: typing.Optional[typing.Dict[str, typing.Any]] = OMIT, workflow: typing.Optional[AgentWorkflowRequestModel] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[CreateAgentBranchResponseModel]: """ Create a new branch from a given version of main branch Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. parent_version_id : str ID of the version to branch from name : str Name of the branch. It is unique within the agent. description : str Description for the branch conversation_config : typing.Optional[typing.Dict[str, typing.Any]] Changes to apply to conversation config platform_settings : typing.Optional[typing.Dict[str, typing.Any]] Changes to apply to platform settings workflow : typing.Optional[AgentWorkflowRequestModel] Updated workflow definition request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[CreateAgentBranchResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches", method="POST", json={ "parent_version_id": parent_version_id, "name": name, "description": description, "conversation_config": conversation_config, "platform_settings": platform_settings, "workflow": convert_and_respect_annotation_metadata( object_=workflow, annotation=AgentWorkflowRequestModel, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentBranchResponseModel, construct_type( type_=CreateAgentBranchResponseModel, # 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, branch_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[AgentBranchResponse]: """ Get information about a single agent branch Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. branch_id : str Unique identifier for the branch. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBranchResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches/{jsonable_encoder(branch_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBranchResponse, construct_type( type_=AgentBranchResponse, # 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 update( self, agent_id: str, branch_id: str, *, name: typing.Optional[str] = OMIT, is_archived: typing.Optional[bool] = OMIT, protection_status: typing.Optional[BranchProtectionStatus] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentBranchResponse]: """ Update agent branch properties such as archiving status and protection level Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. branch_id : str Unique identifier for the branch. name : typing.Optional[str] New name for the branch. Must be unique within the agent. is_archived : typing.Optional[bool] Whether the branch should be archived protection_status : typing.Optional[BranchProtectionStatus] The protection level for the branch request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBranchResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches/{jsonable_encoder(branch_id)}", method="PATCH", json={ "name": name, "is_archived": is_archived, "protection_status": protection_status, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBranchResponse, construct_type( type_=AgentBranchResponse, # 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 merge( self, agent_id: str, source_branch_id: str, *, target_branch_id: str, archive_source_branch: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.Any]: """ Merge a branch into a target branch Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. source_branch_id : str Unique identifier for the source branch to merge from. target_branch_id : str The ID of the target branch to merge into (must be the main branch). archive_source_branch : typing.Optional[bool] Whether to archive the source branch after merging request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches/{jsonable_encoder(source_branch_id)}/merge", method="POST", params={ "target_branch_id": target_branch_id, }, json={ "archive_source_branch": archive_source_branch, }, 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 AsyncRawBranchesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, agent_id: str, *, include_archived: typing.Optional[bool] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ListResponseAgentBranchSummary]: """ Returns a list of branches an agent has Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. include_archived : typing.Optional[bool] Whether archived branches should be included limit : typing.Optional[int] How many results at most should be returned request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ListResponseAgentBranchSummary] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches", method="GET", params={ "include_archived": include_archived, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListResponseAgentBranchSummary, construct_type( type_=ListResponseAgentBranchSummary, # 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 create( self, agent_id: str, *, parent_version_id: str, name: str, description: str, conversation_config: typing.Optional[typing.Dict[str, typing.Any]] = OMIT, platform_settings: typing.Optional[typing.Dict[str, typing.Any]] = OMIT, workflow: typing.Optional[AgentWorkflowRequestModel] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[CreateAgentBranchResponseModel]: """ Create a new branch from a given version of main branch Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. parent_version_id : str ID of the version to branch from name : str Name of the branch. It is unique within the agent. description : str Description for the branch conversation_config : typing.Optional[typing.Dict[str, typing.Any]] Changes to apply to conversation config platform_settings : typing.Optional[typing.Dict[str, typing.Any]] Changes to apply to platform settings workflow : typing.Optional[AgentWorkflowRequestModel] Updated workflow definition request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[CreateAgentBranchResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches", method="POST", json={ "parent_version_id": parent_version_id, "name": name, "description": description, "conversation_config": conversation_config, "platform_settings": platform_settings, "workflow": convert_and_respect_annotation_metadata( object_=workflow, annotation=AgentWorkflowRequestModel, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentBranchResponseModel, construct_type( type_=CreateAgentBranchResponseModel, # 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, branch_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[AgentBranchResponse]: """ Get information about a single agent branch Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. branch_id : str Unique identifier for the branch. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBranchResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches/{jsonable_encoder(branch_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBranchResponse, construct_type( type_=AgentBranchResponse, # 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 update( self, agent_id: str, branch_id: str, *, name: typing.Optional[str] = OMIT, is_archived: typing.Optional[bool] = OMIT, protection_status: typing.Optional[BranchProtectionStatus] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentBranchResponse]: """ Update agent branch properties such as archiving status and protection level Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. branch_id : str Unique identifier for the branch. name : typing.Optional[str] New name for the branch. Must be unique within the agent. is_archived : typing.Optional[bool] Whether the branch should be archived protection_status : typing.Optional[BranchProtectionStatus] The protection level for the branch request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBranchResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/agents/{jsonable_encoder(agent_id)}/branches/{jsonable_encoder(branch_id)}", method="PATCH", json={ "name": name, "is_archived": is_archived, "protection_status": protection_status, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBranchResponse, construct_type( type_=AgentBranchResponse, # 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 merge( self, agent_id: str, source_branch_id: str, *, target_branch_id: str, archive_source_branch: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.Any]: """ Merge a branch into a target branch Parameters ---------- agent_id : str The id of an agent. This is returned on agent creation. source_branch_id : str Unique identifier for the source branch to merge from. target_branch_id : str The ID of the target branch to merge into (must be the main branch). archive_source_branch : typing.Optional[bool] Whether to archive the source branch after merging 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/agents/{jsonable_encoder(agent_id)}/branches/{jsonable_encoder(source_branch_id)}/merge", method="POST", params={ "target_branch_id": target_branch_id, }, json={ "archive_source_branch": archive_source_branch, }, 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)