# 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.http_validation_error import HttpValidationError from ...types.mcp_approval_policy import McpApprovalPolicy from ...types.mcp_server_config_input import McpServerConfigInput from ...types.mcp_server_response_model import McpServerResponseModel from ...types.mcp_servers_response_model import McpServersResponseModel from ...types.tool_call_sound_behavior import ToolCallSoundBehavior from ...types.tool_call_sound_type import ToolCallSoundType from ...types.tool_execution_mode import ToolExecutionMode from .types.mcp_server_config_update_request_model_request_headers_value import ( McpServerConfigUpdateRequestModelRequestHeadersValue, ) # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawMcpServersClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list(self, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[McpServersResponseModel]: """ Retrieve all MCP server configurations available in the workspace. Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[McpServersResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/mcp-servers", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServersResponseModel, construct_type( type_=McpServersResponseModel, # 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, *, config: McpServerConfigInput, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[McpServerResponseModel]: """ Create a new MCP server configuration in the workspace. Parameters ---------- config : McpServerConfigInput Configuration details for the MCP Server. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[McpServerResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/mcp-servers", method="POST", json={ "config": convert_and_respect_annotation_metadata( object_=config, annotation=McpServerConfigInput, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServerResponseModel, construct_type( type_=McpServerResponseModel, # 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, mcp_server_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[McpServerResponseModel]: """ Retrieve a specific MCP server configuration from the workspace. Parameters ---------- mcp_server_id : str ID of the MCP Server. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[McpServerResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/mcp-servers/{jsonable_encoder(mcp_server_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServerResponseModel, construct_type( type_=McpServerResponseModel, # 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, mcp_server_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.Any]: """ Delete a specific MCP server configuration from the workspace. Parameters ---------- mcp_server_id : str ID of the MCP Server. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/mcp-servers/{jsonable_encoder(mcp_server_id)}", method="DELETE", request_options=request_options, ) 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) def update( self, mcp_server_id: str, *, approval_policy: typing.Optional[McpApprovalPolicy] = OMIT, force_pre_tool_speech: typing.Optional[bool] = OMIT, disable_interruptions: typing.Optional[bool] = OMIT, tool_call_sound: typing.Optional[ToolCallSoundType] = OMIT, tool_call_sound_behavior: typing.Optional[ToolCallSoundBehavior] = OMIT, execution_mode: typing.Optional[ToolExecutionMode] = OMIT, request_headers: typing.Optional[ typing.Dict[str, typing.Optional[McpServerConfigUpdateRequestModelRequestHeadersValue]] ] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[McpServerResponseModel]: """ Update the configuration settings for an MCP server. Parameters ---------- mcp_server_id : str ID of the MCP Server. approval_policy : typing.Optional[McpApprovalPolicy] The approval mode to set for the MCP server force_pre_tool_speech : typing.Optional[bool] If set, overrides the server's force_pre_tool_speech setting for this tool disable_interruptions : typing.Optional[bool] If set, overrides the server's disable_interruptions setting for this tool tool_call_sound : typing.Optional[ToolCallSoundType] Predefined tool call sound type to play during tool execution for all tools from this MCP server tool_call_sound_behavior : typing.Optional[ToolCallSoundBehavior] Determines when the tool call sound should play for all tools from this MCP server execution_mode : typing.Optional[ToolExecutionMode] If set, overrides the server's execution_mode setting for this tool request_headers : typing.Optional[typing.Dict[str, typing.Optional[McpServerConfigUpdateRequestModelRequestHeadersValue]]] The headers to include in requests to the MCP server request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[McpServerResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/mcp-servers/{jsonable_encoder(mcp_server_id)}", method="PATCH", json={ "approval_policy": approval_policy, "force_pre_tool_speech": force_pre_tool_speech, "disable_interruptions": disable_interruptions, "tool_call_sound": tool_call_sound, "tool_call_sound_behavior": tool_call_sound_behavior, "execution_mode": execution_mode, "request_headers": convert_and_respect_annotation_metadata( object_=request_headers, annotation=typing.Dict[str, typing.Optional[McpServerConfigUpdateRequestModelRequestHeadersValue]], direction="write", ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServerResponseModel, construct_type( type_=McpServerResponseModel, # 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 AsyncRawMcpServersClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[McpServersResponseModel]: """ Retrieve all MCP server configurations available in the workspace. Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[McpServersResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/mcp-servers", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServersResponseModel, construct_type( type_=McpServersResponseModel, # 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, *, config: McpServerConfigInput, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[McpServerResponseModel]: """ Create a new MCP server configuration in the workspace. Parameters ---------- config : McpServerConfigInput Configuration details for the MCP Server. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[McpServerResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/mcp-servers", method="POST", json={ "config": convert_and_respect_annotation_metadata( object_=config, annotation=McpServerConfigInput, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServerResponseModel, construct_type( type_=McpServerResponseModel, # 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, mcp_server_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[McpServerResponseModel]: """ Retrieve a specific MCP server configuration from the workspace. Parameters ---------- mcp_server_id : str ID of the MCP Server. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[McpServerResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/mcp-servers/{jsonable_encoder(mcp_server_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServerResponseModel, construct_type( type_=McpServerResponseModel, # 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, mcp_server_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.Any]: """ Delete a specific MCP server configuration from the workspace. Parameters ---------- mcp_server_id : str ID of the MCP Server. 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/mcp-servers/{jsonable_encoder(mcp_server_id)}", method="DELETE", request_options=request_options, ) 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) async def update( self, mcp_server_id: str, *, approval_policy: typing.Optional[McpApprovalPolicy] = OMIT, force_pre_tool_speech: typing.Optional[bool] = OMIT, disable_interruptions: typing.Optional[bool] = OMIT, tool_call_sound: typing.Optional[ToolCallSoundType] = OMIT, tool_call_sound_behavior: typing.Optional[ToolCallSoundBehavior] = OMIT, execution_mode: typing.Optional[ToolExecutionMode] = OMIT, request_headers: typing.Optional[ typing.Dict[str, typing.Optional[McpServerConfigUpdateRequestModelRequestHeadersValue]] ] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[McpServerResponseModel]: """ Update the configuration settings for an MCP server. Parameters ---------- mcp_server_id : str ID of the MCP Server. approval_policy : typing.Optional[McpApprovalPolicy] The approval mode to set for the MCP server force_pre_tool_speech : typing.Optional[bool] If set, overrides the server's force_pre_tool_speech setting for this tool disable_interruptions : typing.Optional[bool] If set, overrides the server's disable_interruptions setting for this tool tool_call_sound : typing.Optional[ToolCallSoundType] Predefined tool call sound type to play during tool execution for all tools from this MCP server tool_call_sound_behavior : typing.Optional[ToolCallSoundBehavior] Determines when the tool call sound should play for all tools from this MCP server execution_mode : typing.Optional[ToolExecutionMode] If set, overrides the server's execution_mode setting for this tool request_headers : typing.Optional[typing.Dict[str, typing.Optional[McpServerConfigUpdateRequestModelRequestHeadersValue]]] The headers to include in requests to the MCP server request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[McpServerResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/mcp-servers/{jsonable_encoder(mcp_server_id)}", method="PATCH", json={ "approval_policy": approval_policy, "force_pre_tool_speech": force_pre_tool_speech, "disable_interruptions": disable_interruptions, "tool_call_sound": tool_call_sound, "tool_call_sound_behavior": tool_call_sound_behavior, "execution_mode": execution_mode, "request_headers": convert_and_respect_annotation_metadata( object_=request_headers, annotation=typing.Dict[str, typing.Optional[McpServerConfigUpdateRequestModelRequestHeadersValue]], direction="write", ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( McpServerResponseModel, construct_type( type_=McpServerResponseModel, # 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)