# 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.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from ..core.serialization import convert_and_respect_annotation_metadata from ..errors.bad_request_error import BadRequestError from ..errors.not_found_error import NotFoundError from ..errors.unauthorized_error import UnauthorizedError from ..types.agent_blueprint_public import AgentBlueprintPublic from ..types.agent_blueprint_write import AgentBlueprintWrite from ..types.agent_config_env import AgentConfigEnv from ..types.blueprint_page_history import BlueprintPageHistory # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawAgentConfigsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def create_agent_config( self, *, blueprint: AgentBlueprintWrite, project_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Creates a new optimizer config with initial blueprint, or adds a new blueprint to existing config Parameters ---------- blueprint : AgentBlueprintWrite project_id : typing.Optional[str] Project ID. Either project_id or project_name must be provided project_name : typing.Optional[str] Project name. Either project_id or project_name must be provided id : typing.Optional[str] Agent config ID. Generated automatically if not provided request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/agent-configs/blueprints", method="POST", json={ "project_id": project_id, "project_name": project_name, "id": id, "blueprint": convert_and_respect_annotation_metadata( object_=blueprint, annotation=AgentBlueprintWrite, direction="write" ), }, 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 == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def create_or_update_envs( self, *, project_id: str, envs: typing.Sequence[AgentConfigEnv], request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Creates or updates environment-to-blueprint mappings Parameters ---------- project_id : str envs : typing.Sequence[AgentConfigEnv] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/agent-configs/blueprints/environments", method="POST", json={ "project_id": project_id, "envs": convert_and_respect_annotation_metadata( object_=envs, annotation=typing.Sequence[AgentConfigEnv], direction="write" ), }, 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 == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_blueprint_by_env( self, env_name: str, project_id: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentBlueprintPublic]: """ Retrieves the blueprint associated with a specific environment Parameters ---------- env_name : str project_id : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/environments/{jsonable_encoder(env_name)}/projects/{jsonable_encoder(project_id)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def set_env_by_blueprint_name( self, env_name: str, project_id: str, *, blueprint_name: str, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Sets an environment to point to a blueprint identified by name Parameters ---------- env_name : str project_id : str blueprint_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/environments/{jsonable_encoder(env_name)}/projects/{jsonable_encoder(project_id)}", method="PUT", json={ "blueprint_name": blueprint_name, }, 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 == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete_env( self, env_name: str, project_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Soft-deletes an environment by setting its ended_at timestamp Parameters ---------- env_name : str project_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/environments/{jsonable_encoder(env_name)}/projects/{jsonable_encoder(project_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_blueprint_by_id( self, blueprint_id: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentBlueprintPublic]: """ Retrieves a specific blueprint by its ID Parameters ---------- blueprint_id : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/{jsonable_encoder(blueprint_id)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_blueprint_by_name( self, project_id: str, name: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentBlueprintPublic]: """ Retrieves a specific blueprint by its name within a project Parameters ---------- project_id : str name : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/projects/{jsonable_encoder(project_id)}/names/{jsonable_encoder(name)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_blueprint_history( self, project_id: str, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[BlueprintPageHistory]: """ Retrieves paginated blueprint history for a project Parameters ---------- project_id : str page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[BlueprintPageHistory] History retrieved """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/history/projects/{jsonable_encoder(project_id)}", method="GET", params={ "page": page, "size": size, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( BlueprintPageHistory, parse_obj_as( type_=BlueprintPageHistory, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_delta_by_id( self, blueprint_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[AgentBlueprintPublic]: """ Retrieves only the changes (delta) introduced in a specific blueprint Parameters ---------- blueprint_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBlueprintPublic] Delta retrieved """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/{jsonable_encoder(blueprint_id)}/deltas", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_latest_blueprint( self, project_id: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AgentBlueprintPublic]: """ Retrieves the latest blueprint for a project Parameters ---------- project_id : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/latest/projects/{jsonable_encoder(project_id)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawAgentConfigsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def create_agent_config( self, *, blueprint: AgentBlueprintWrite, project_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Creates a new optimizer config with initial blueprint, or adds a new blueprint to existing config Parameters ---------- blueprint : AgentBlueprintWrite project_id : typing.Optional[str] Project ID. Either project_id or project_name must be provided project_name : typing.Optional[str] Project name. Either project_id or project_name must be provided id : typing.Optional[str] Agent config ID. Generated automatically if not provided request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/agent-configs/blueprints", method="POST", json={ "project_id": project_id, "project_name": project_name, "id": id, "blueprint": convert_and_respect_annotation_metadata( object_=blueprint, annotation=AgentBlueprintWrite, direction="write" ), }, 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 == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def create_or_update_envs( self, *, project_id: str, envs: typing.Sequence[AgentConfigEnv], request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Creates or updates environment-to-blueprint mappings Parameters ---------- project_id : str envs : typing.Sequence[AgentConfigEnv] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/agent-configs/blueprints/environments", method="POST", json={ "project_id": project_id, "envs": convert_and_respect_annotation_metadata( object_=envs, annotation=typing.Sequence[AgentConfigEnv], direction="write" ), }, 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 == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_blueprint_by_env( self, env_name: str, project_id: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentBlueprintPublic]: """ Retrieves the blueprint associated with a specific environment Parameters ---------- env_name : str project_id : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/environments/{jsonable_encoder(env_name)}/projects/{jsonable_encoder(project_id)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def set_env_by_blueprint_name( self, env_name: str, project_id: str, *, blueprint_name: str, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Sets an environment to point to a blueprint identified by name Parameters ---------- env_name : str project_id : str blueprint_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/environments/{jsonable_encoder(env_name)}/projects/{jsonable_encoder(project_id)}", method="PUT", json={ "blueprint_name": blueprint_name, }, 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 == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete_env( self, env_name: str, project_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Soft-deletes an environment by setting its ended_at timestamp Parameters ---------- env_name : str project_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/environments/{jsonable_encoder(env_name)}/projects/{jsonable_encoder(project_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_blueprint_by_id( self, blueprint_id: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentBlueprintPublic]: """ Retrieves a specific blueprint by its ID Parameters ---------- blueprint_id : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/{jsonable_encoder(blueprint_id)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_blueprint_by_name( self, project_id: str, name: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentBlueprintPublic]: """ Retrieves a specific blueprint by its name within a project Parameters ---------- project_id : str name : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/projects/{jsonable_encoder(project_id)}/names/{jsonable_encoder(name)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_blueprint_history( self, project_id: str, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[BlueprintPageHistory]: """ Retrieves paginated blueprint history for a project Parameters ---------- project_id : str page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[BlueprintPageHistory] History retrieved """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/history/projects/{jsonable_encoder(project_id)}", method="GET", params={ "page": page, "size": size, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( BlueprintPageHistory, parse_obj_as( type_=BlueprintPageHistory, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_delta_by_id( self, blueprint_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[AgentBlueprintPublic]: """ Retrieves only the changes (delta) introduced in a specific blueprint Parameters ---------- blueprint_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBlueprintPublic] Delta retrieved """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/{jsonable_encoder(blueprint_id)}/deltas", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_latest_blueprint( self, project_id: str, *, mask_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AgentBlueprintPublic]: """ Retrieves the latest blueprint for a project Parameters ---------- project_id : str mask_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AgentBlueprintPublic] Blueprint retrieved """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/agent-configs/blueprints/latest/projects/{jsonable_encoder(project_id)}", method="GET", params={ "mask_id": mask_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AgentBlueprintPublic, parse_obj_as( type_=AgentBlueprintPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[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) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)