# 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 encode_path_param from ...core.parse_error import ParsingError from ...core.request_options import RequestOptions from ...core.unchecked_base_model import construct_type from ...errors.bad_request_error import BadRequestError from ...types.agent_configuration_v1 import AgentConfigurationV1 from ...types.create_agent_configuration_v1response import CreateAgentConfigurationV1Response from ...types.delete_agent_configuration_v1response import DeleteAgentConfigurationV1Response from ...types.list_agent_configurations_v1response import ListAgentConfigurationsV1Response from pydantic import ValidationError # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawConfigurationsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, project_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ListAgentConfigurationsV1Response]: """ Returns all agent configurations for the specified project. Configurations are returned in their uninterpolated form—template variable placeholders appear as-is rather than with their substituted values. Parameters ---------- project_id : str The unique identifier of the project request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ListAgentConfigurationsV1Response] A list of agent configurations """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListAgentConfigurationsV1Response, construct_type( type_=ListAgentConfigurationsV1Response, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def create( self, project_id: str, *, config: str, metadata: typing.Optional[typing.Dict[str, str]] = OMIT, api_version: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[CreateAgentConfigurationV1Response]: """ Creates a new reusable agent configuration. The `config` field must be a valid JSON string representing the `agent` block of a Settings message. The returned `agent_id` can be passed in place of the full `agent` object in future Settings messages. Parameters ---------- project_id : str The unique identifier of the project config : str A valid JSON string representing the agent block of a Settings message metadata : typing.Optional[typing.Dict[str, str]] A map of arbitrary key-value pairs for labeling or organizing the agent configuration api_version : typing.Optional[int] API version. Defaults to 1 request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[CreateAgentConfigurationV1Response] Agent configuration created successfully """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "config": config, "metadata": metadata, "api_version": api_version, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentConfigurationV1Response, construct_type( type_=CreateAgentConfigurationV1Response, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get( self, project_id: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[AgentConfigurationV1]: """ Returns the specified agent configuration in its uninterpolated form Parameters ---------- project_id : str The unique identifier of the project agent_id : str The unique identifier of the agent configuration request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentConfigurationV1] An agent configuration """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents/{encode_path_param(agent_id)}", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentConfigurationV1, construct_type( type_=AgentConfigurationV1, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def update( self, project_id: str, agent_id: str, *, metadata: typing.Dict[str, str], request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentConfigurationV1]: """ Updates the metadata associated with an agent configuration. The config itself is immutable—to change the configuration, delete the existing agent and create a new one. Parameters ---------- project_id : str The unique identifier of the project agent_id : str The unique identifier of the agent configuration metadata : typing.Dict[str, str] A map of string key-value pairs to associate with this agent configuration request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentConfigurationV1] Agent configuration updated """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents/{encode_path_param(agent_id)}", base_url=self._client_wrapper.get_environment().base, method="PUT", json={ "metadata": metadata, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentConfigurationV1, construct_type( type_=AgentConfigurationV1, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete( self, project_id: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteAgentConfigurationV1Response]: """ Deletes the specified agent configuration. Deleting an agent configuration can cause a production outage if your service references this agent UUID. Migrate all active sessions to a new configuration before deleting. Parameters ---------- project_id : str The unique identifier of the project agent_id : str The unique identifier of the agent configuration request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteAgentConfigurationV1Response] Agent configuration deleted """ _response = self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents/{encode_path_param(agent_id)}", base_url=self._client_wrapper.get_environment().base, method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteAgentConfigurationV1Response, construct_type( type_=DeleteAgentConfigurationV1Response, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawConfigurationsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, project_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ListAgentConfigurationsV1Response]: """ Returns all agent configurations for the specified project. Configurations are returned in their uninterpolated form—template variable placeholders appear as-is rather than with their substituted values. Parameters ---------- project_id : str The unique identifier of the project request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ListAgentConfigurationsV1Response] A list of agent configurations """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListAgentConfigurationsV1Response, construct_type( type_=ListAgentConfigurationsV1Response, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def create( self, project_id: str, *, config: str, metadata: typing.Optional[typing.Dict[str, str]] = OMIT, api_version: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[CreateAgentConfigurationV1Response]: """ Creates a new reusable agent configuration. The `config` field must be a valid JSON string representing the `agent` block of a Settings message. The returned `agent_id` can be passed in place of the full `agent` object in future Settings messages. Parameters ---------- project_id : str The unique identifier of the project config : str A valid JSON string representing the agent block of a Settings message metadata : typing.Optional[typing.Dict[str, str]] A map of arbitrary key-value pairs for labeling or organizing the agent configuration api_version : typing.Optional[int] API version. Defaults to 1 request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[CreateAgentConfigurationV1Response] Agent configuration created successfully """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "config": config, "metadata": metadata, "api_version": api_version, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateAgentConfigurationV1Response, construct_type( type_=CreateAgentConfigurationV1Response, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get( self, project_id: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[AgentConfigurationV1]: """ Returns the specified agent configuration in its uninterpolated form Parameters ---------- project_id : str The unique identifier of the project agent_id : str The unique identifier of the agent configuration request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentConfigurationV1] An agent configuration """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents/{encode_path_param(agent_id)}", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentConfigurationV1, construct_type( type_=AgentConfigurationV1, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def update( self, project_id: str, agent_id: str, *, metadata: typing.Dict[str, str], request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentConfigurationV1]: """ Updates the metadata associated with an agent configuration. The config itself is immutable—to change the configuration, delete the existing agent and create a new one. Parameters ---------- project_id : str The unique identifier of the project agent_id : str The unique identifier of the agent configuration metadata : typing.Dict[str, str] A map of string key-value pairs to associate with this agent configuration request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentConfigurationV1] Agent configuration updated """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents/{encode_path_param(agent_id)}", base_url=self._client_wrapper.get_environment().base, method="PUT", json={ "metadata": metadata, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentConfigurationV1, construct_type( type_=AgentConfigurationV1, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete( self, project_id: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteAgentConfigurationV1Response]: """ Deletes the specified agent configuration. Deleting an agent configuration can cause a production outage if your service references this agent UUID. Migrate all active sessions to a new configuration before deleting. Parameters ---------- project_id : str The unique identifier of the project agent_id : str The unique identifier of the agent configuration request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteAgentConfigurationV1Response] Agent configuration deleted """ _response = await self._client_wrapper.httpx_client.request( f"v1/projects/{encode_path_param(project_id)}/agents/{encode_path_param(agent_id)}", base_url=self._client_wrapper.get_environment().base, method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteAgentConfigurationV1Response, construct_type( type_=DeleteAgentConfigurationV1Response, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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) except ValidationError as e: raise ParsingError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)