# This file was auto-generated by Fern from our API Definition. import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from ..types.action_model import ActionModel from ..types.app_model import AppModel from ..types.mcp_tool import McpTool from ..types.npm_requirement import NpmRequirement from ..types.pip_requirement import PipRequirement from ..types.tool import Tool from ..types.tool_return_message import ToolReturnMessage from .raw_client import AsyncRawToolsClient, RawToolsClient from .types.add_mcp_server_request import AddMcpServerRequest from .types.add_mcp_server_response_item import AddMcpServerResponseItem from .types.connect_mcp_server_request import ConnectMcpServerRequest from .types.delete_mcp_server_response_item import DeleteMcpServerResponseItem from .types.list_mcp_servers_response_value import ListMcpServersResponseValue from .types.streaming_response import StreamingResponse from .types.test_mcp_server_request import TestMcpServerRequest from .types.update_mcp_server_request import UpdateMcpServerRequest from .types.update_mcp_server_response import UpdateMcpServerResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class ToolsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawToolsClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawToolsClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawToolsClient """ return self._raw_client def retrieve(self, tool_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Tool: """ Get a tool by ID Parameters ---------- tool_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.retrieve( tool_id="tool_id", ) """ _response = self._raw_client.retrieve(tool_id, request_options=request_options) return _response.data def delete( self, tool_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Optional[typing.Any]: """ Delete a tool by name Parameters ---------- tool_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Optional[typing.Any] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.delete( tool_id="tool_id", ) """ _response = self._raw_client.delete(tool_id, request_options=request_options) return _response.data def modify( self, tool_id: str, *, description: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, source_code: typing.Optional[str] = OMIT, source_type: typing.Optional[str] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, return_char_limit: typing.Optional[int] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, metadata: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, default_requires_approval: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> Tool: """ Update an existing tool Parameters ---------- tool_id : str description : typing.Optional[str] The description of the tool. tags : typing.Optional[typing.Sequence[str]] Metadata tags. source_code : typing.Optional[str] The source code of the function. source_type : typing.Optional[str] The type of the source code. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. return_char_limit : typing.Optional[int] The maximum number of characters in the response. pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. metadata : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] A dictionary of additional metadata for the tool. default_requires_approval : typing.Optional[bool] Whether or not to require approval before executing this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.modify( tool_id="tool_id", ) """ _response = self._raw_client.modify( tool_id, description=description, tags=tags, source_code=source_code, source_type=source_type, json_schema=json_schema, args_json_schema=args_json_schema, return_char_limit=return_char_limit, pip_requirements=pip_requirements, npm_requirements=npm_requirements, metadata=metadata, default_requires_approval=default_requires_approval, request_options=request_options, ) return _response.data def count( self, *, name: typing.Optional[str] = None, names: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, tool_ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, search: typing.Optional[str] = None, tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, exclude_tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, return_only_letta_tools: typing.Optional[bool] = None, exclude_letta_tools: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> int: """ Get a count of all tools available to agents belonging to the org of the user. Parameters ---------- name : typing.Optional[str] names : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool names tool_ids : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool IDs - accepts repeated params or comma-separated values search : typing.Optional[str] Search tool names (case-insensitive partial match) tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by tool type(s) - accepts repeated params or comma-separated values exclude_tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Tool type(s) to exclude - accepts repeated params or comma-separated values return_only_letta_tools : typing.Optional[bool] Count only tools with tool_type starting with 'letta_' exclude_letta_tools : typing.Optional[bool] Exclude built-in Letta tools from the count request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- int Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.count() """ _response = self._raw_client.count( name=name, names=names, tool_ids=tool_ids, search=search, tool_types=tool_types, exclude_tool_types=exclude_tool_types, return_only_letta_tools=return_only_letta_tools, exclude_letta_tools=exclude_letta_tools, request_options=request_options, ) return _response.data def list( self, *, after: typing.Optional[str] = None, limit: typing.Optional[int] = None, name: typing.Optional[str] = None, names: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, tool_ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, search: typing.Optional[str] = None, tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, exclude_tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, return_only_letta_tools: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.List[Tool]: """ Get a list of all tools available to agents belonging to the org of the user Parameters ---------- after : typing.Optional[str] limit : typing.Optional[int] name : typing.Optional[str] names : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool names tool_ids : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool IDs - accepts repeated params or comma-separated values search : typing.Optional[str] Search tool names (case-insensitive partial match) tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by tool type(s) - accepts repeated params or comma-separated values exclude_tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Tool type(s) to exclude - accepts repeated params or comma-separated values return_only_letta_tools : typing.Optional[bool] Return only tools with tool_type starting with 'letta_' request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[Tool] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.list() """ _response = self._raw_client.list( after=after, limit=limit, name=name, names=names, tool_ids=tool_ids, search=search, tool_types=tool_types, exclude_tool_types=exclude_tool_types, return_only_letta_tools=return_only_letta_tools, request_options=request_options, ) return _response.data def create( self, *, source_code: str, description: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, source_type: typing.Optional[str] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, return_char_limit: typing.Optional[int] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, default_requires_approval: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> Tool: """ Create a new tool Parameters ---------- source_code : str The source code of the function. description : typing.Optional[str] The description of the tool. tags : typing.Optional[typing.Sequence[str]] Metadata tags. source_type : typing.Optional[str] The source type of the function. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. return_char_limit : typing.Optional[int] The maximum number of characters in the response. pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. default_requires_approval : typing.Optional[bool] Whether or not to require approval before executing this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.create( source_code="source_code", ) """ _response = self._raw_client.create( source_code=source_code, description=description, tags=tags, source_type=source_type, json_schema=json_schema, args_json_schema=args_json_schema, return_char_limit=return_char_limit, pip_requirements=pip_requirements, npm_requirements=npm_requirements, default_requires_approval=default_requires_approval, request_options=request_options, ) return _response.data def upsert( self, *, source_code: str, description: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, source_type: typing.Optional[str] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, return_char_limit: typing.Optional[int] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, default_requires_approval: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> Tool: """ Create or update a tool Parameters ---------- source_code : str The source code of the function. description : typing.Optional[str] The description of the tool. tags : typing.Optional[typing.Sequence[str]] Metadata tags. source_type : typing.Optional[str] The source type of the function. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. return_char_limit : typing.Optional[int] The maximum number of characters in the response. pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. default_requires_approval : typing.Optional[bool] Whether or not to require approval before executing this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.upsert( source_code="source_code", ) """ _response = self._raw_client.upsert( source_code=source_code, description=description, tags=tags, source_type=source_type, json_schema=json_schema, args_json_schema=args_json_schema, return_char_limit=return_char_limit, pip_requirements=pip_requirements, npm_requirements=npm_requirements, default_requires_approval=default_requires_approval, request_options=request_options, ) return _response.data def upsert_base_tools(self, *, request_options: typing.Optional[RequestOptions] = None) -> typing.List[Tool]: """ Upsert base tools Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[Tool] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.upsert_base_tools() """ _response = self._raw_client.upsert_base_tools(request_options=request_options) return _response.data def run_tool_from_source( self, *, source_code: str, args: typing.Dict[str, typing.Optional[typing.Any]], env_vars: typing.Optional[typing.Dict[str, str]] = OMIT, name: typing.Optional[str] = OMIT, source_type: typing.Optional[str] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> ToolReturnMessage: """ Attempt to build a tool from source, then run it on the provided arguments Parameters ---------- source_code : str The source code of the function. args : typing.Dict[str, typing.Optional[typing.Any]] The arguments to pass to the tool. env_vars : typing.Optional[typing.Dict[str, str]] The environment variables to pass to the tool. name : typing.Optional[str] The name of the tool to run. source_type : typing.Optional[str] The type of the source code. args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ToolReturnMessage Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.run_tool_from_source( source_code="source_code", args={"key": "value"}, ) """ _response = self._raw_client.run_tool_from_source( source_code=source_code, args=args, env_vars=env_vars, name=name, source_type=source_type, args_json_schema=args_json_schema, json_schema=json_schema, pip_requirements=pip_requirements, npm_requirements=npm_requirements, request_options=request_options, ) return _response.data def list_composio_apps( self, *, user_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[AppModel]: """ Get a list of all Composio apps Parameters ---------- user_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[AppModel] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.list_composio_apps() """ _response = self._raw_client.list_composio_apps(user_id=user_id, request_options=request_options) return _response.data def list_composio_actions_by_app( self, composio_app_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[ActionModel]: """ Get a list of all Composio actions for a specific app Parameters ---------- composio_app_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[ActionModel] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.list_composio_actions_by_app( composio_app_name="composio_app_name", ) """ _response = self._raw_client.list_composio_actions_by_app(composio_app_name, request_options=request_options) return _response.data def add_composio_tool( self, composio_action_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Tool: """ Add a new Composio tool by action name (Composio refers to each tool as an `Action`) Parameters ---------- composio_action_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.add_composio_tool( composio_action_name="composio_action_name", ) """ _response = self._raw_client.add_composio_tool(composio_action_name, request_options=request_options) return _response.data def list_mcp_servers( self, *, user_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> typing.Dict[str, ListMcpServersResponseValue]: """ Get a list of all configured MCP servers Parameters ---------- user_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Dict[str, ListMcpServersResponseValue] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.list_mcp_servers() """ _response = self._raw_client.list_mcp_servers(user_id=user_id, request_options=request_options) return _response.data def add_mcp_server( self, *, request: AddMcpServerRequest, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[AddMcpServerResponseItem]: """ Add a new MCP server to the Letta MCP server config Parameters ---------- request : AddMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[AddMcpServerResponseItem] Successful Response Examples -------- from letta_client import Letta, StdioServerConfig client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.add_mcp_server( request=StdioServerConfig( server_name="server_name", command="command", args=["args"], ), ) """ _response = self._raw_client.add_mcp_server(request=request, request_options=request_options) return _response.data def list_mcp_tools_by_server( self, mcp_server_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[McpTool]: """ Get a list of all tools for a specific MCP server Parameters ---------- mcp_server_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[McpTool] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.list_mcp_tools_by_server( mcp_server_name="mcp_server_name", ) """ _response = self._raw_client.list_mcp_tools_by_server(mcp_server_name, request_options=request_options) return _response.data def add_mcp_tool( self, mcp_server_name: str, mcp_tool_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Tool: """ Register a new MCP tool as a Letta server by MCP server + tool name Parameters ---------- mcp_server_name : str mcp_tool_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.add_mcp_tool( mcp_server_name="mcp_server_name", mcp_tool_name="mcp_tool_name", ) """ _response = self._raw_client.add_mcp_tool(mcp_server_name, mcp_tool_name, request_options=request_options) return _response.data def add_mcp_tool_override( self, mcp_server_name: str, mcp_tool_name: str, overridden_schema:typing.Optional[dict] = None, *, request_options: typing.Optional[RequestOptions] = None ) -> Tool: """ Register a new MCP tool as a Letta server by MCP server + tool name Parameters ---------- mcp_server_name : str mcp_tool_name : str overridden_schema: Optional[dict] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.add_mcp_tool( mcp_server_name="mcp_server_name", mcp_tool_name="mcp_tool_name", ) """ _response = self._raw_client.add_mcp_tool_override(mcp_server_name, mcp_tool_name, overridden_schema=overridden_schema, request_options=request_options) return _response.data def delete_mcp_server( self, mcp_server_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[DeleteMcpServerResponseItem]: """ Delete a MCP server configuration Parameters ---------- mcp_server_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[DeleteMcpServerResponseItem] Successful Response Examples -------- from letta_client import Letta client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.delete_mcp_server( mcp_server_name="mcp_server_name", ) """ _response = self._raw_client.delete_mcp_server(mcp_server_name, request_options=request_options) return _response.data def update_mcp_server( self, mcp_server_name: str, *, request: UpdateMcpServerRequest, request_options: typing.Optional[RequestOptions] = None, ) -> UpdateMcpServerResponse: """ Update an existing MCP server configuration Parameters ---------- mcp_server_name : str request : UpdateMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- UpdateMcpServerResponse Successful Response Examples -------- from letta_client import Letta, UpdateStdioMcpServer client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.update_mcp_server( mcp_server_name="mcp_server_name", request=UpdateStdioMcpServer(), ) """ _response = self._raw_client.update_mcp_server( mcp_server_name, request=request, request_options=request_options ) return _response.data def test_mcp_server( self, *, request: TestMcpServerRequest, request_options: typing.Optional[RequestOptions] = None ) -> typing.Optional[typing.Any]: """ Test connection to an MCP server without adding it. Returns the list of available tools if successful. Parameters ---------- request : TestMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Optional[typing.Any] Successful Response Examples -------- from letta_client import Letta, StdioServerConfig client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) client.tools.test_mcp_server( request=StdioServerConfig( server_name="server_name", command="command", args=["args"], ), ) """ _response = self._raw_client.test_mcp_server(request=request, request_options=request_options) return _response.data def connect_mcp_server( self, *, request: ConnectMcpServerRequest, request_options: typing.Optional[RequestOptions] = None ) -> typing.Iterator[StreamingResponse]: """ Connect to an MCP server with support for OAuth via SSE. Returns a stream of events handling authorization state and exchange if OAuth is required. Parameters ---------- request : ConnectMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Yields ------ typing.Iterator[StreamingResponse] Successful response Examples -------- from letta_client import Letta, StdioServerConfig client = Letta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) response = client.tools.connect_mcp_server( request=StdioServerConfig( server_name="server_name", command="command", args=["args"], ), ) for chunk in response: yield chunk """ with self._raw_client.connect_mcp_server(request=request, request_options=request_options) as r: yield from r.data class AsyncToolsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawToolsClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawToolsClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawToolsClient """ return self._raw_client async def retrieve(self, tool_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Tool: """ Get a tool by ID Parameters ---------- tool_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.retrieve( tool_id="tool_id", ) asyncio.run(main()) """ _response = await self._raw_client.retrieve(tool_id, request_options=request_options) return _response.data async def delete( self, tool_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Optional[typing.Any]: """ Delete a tool by name Parameters ---------- tool_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Optional[typing.Any] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.delete( tool_id="tool_id", ) asyncio.run(main()) """ _response = await self._raw_client.delete(tool_id, request_options=request_options) return _response.data async def modify( self, tool_id: str, *, description: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, source_code: typing.Optional[str] = OMIT, source_type: typing.Optional[str] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, return_char_limit: typing.Optional[int] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, metadata: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, default_requires_approval: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> Tool: """ Update an existing tool Parameters ---------- tool_id : str description : typing.Optional[str] The description of the tool. tags : typing.Optional[typing.Sequence[str]] Metadata tags. source_code : typing.Optional[str] The source code of the function. source_type : typing.Optional[str] The type of the source code. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. return_char_limit : typing.Optional[int] The maximum number of characters in the response. pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. metadata : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] A dictionary of additional metadata for the tool. default_requires_approval : typing.Optional[bool] Whether or not to require approval before executing this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.modify( tool_id="tool_id", ) asyncio.run(main()) """ _response = await self._raw_client.modify( tool_id, description=description, tags=tags, source_code=source_code, source_type=source_type, json_schema=json_schema, args_json_schema=args_json_schema, return_char_limit=return_char_limit, pip_requirements=pip_requirements, npm_requirements=npm_requirements, metadata=metadata, default_requires_approval=default_requires_approval, request_options=request_options, ) return _response.data async def count( self, *, name: typing.Optional[str] = None, names: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, tool_ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, search: typing.Optional[str] = None, tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, exclude_tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, return_only_letta_tools: typing.Optional[bool] = None, exclude_letta_tools: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> int: """ Get a count of all tools available to agents belonging to the org of the user. Parameters ---------- name : typing.Optional[str] names : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool names tool_ids : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool IDs - accepts repeated params or comma-separated values search : typing.Optional[str] Search tool names (case-insensitive partial match) tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by tool type(s) - accepts repeated params or comma-separated values exclude_tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Tool type(s) to exclude - accepts repeated params or comma-separated values return_only_letta_tools : typing.Optional[bool] Count only tools with tool_type starting with 'letta_' exclude_letta_tools : typing.Optional[bool] Exclude built-in Letta tools from the count request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- int Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.count() asyncio.run(main()) """ _response = await self._raw_client.count( name=name, names=names, tool_ids=tool_ids, search=search, tool_types=tool_types, exclude_tool_types=exclude_tool_types, return_only_letta_tools=return_only_letta_tools, exclude_letta_tools=exclude_letta_tools, request_options=request_options, ) return _response.data async def list( self, *, after: typing.Optional[str] = None, limit: typing.Optional[int] = None, name: typing.Optional[str] = None, names: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, tool_ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, search: typing.Optional[str] = None, tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, exclude_tool_types: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, return_only_letta_tools: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.List[Tool]: """ Get a list of all tools available to agents belonging to the org of the user Parameters ---------- after : typing.Optional[str] limit : typing.Optional[int] name : typing.Optional[str] names : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool names tool_ids : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by specific tool IDs - accepts repeated params or comma-separated values search : typing.Optional[str] Search tool names (case-insensitive partial match) tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter by tool type(s) - accepts repeated params or comma-separated values exclude_tool_types : typing.Optional[typing.Union[str, typing.Sequence[str]]] Tool type(s) to exclude - accepts repeated params or comma-separated values return_only_letta_tools : typing.Optional[bool] Return only tools with tool_type starting with 'letta_' request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[Tool] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.list() asyncio.run(main()) """ _response = await self._raw_client.list( after=after, limit=limit, name=name, names=names, tool_ids=tool_ids, search=search, tool_types=tool_types, exclude_tool_types=exclude_tool_types, return_only_letta_tools=return_only_letta_tools, request_options=request_options, ) return _response.data async def create( self, *, source_code: str, description: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, source_type: typing.Optional[str] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, return_char_limit: typing.Optional[int] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, default_requires_approval: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> Tool: """ Create a new tool Parameters ---------- source_code : str The source code of the function. description : typing.Optional[str] The description of the tool. tags : typing.Optional[typing.Sequence[str]] Metadata tags. source_type : typing.Optional[str] The source type of the function. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. return_char_limit : typing.Optional[int] The maximum number of characters in the response. pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. default_requires_approval : typing.Optional[bool] Whether or not to require approval before executing this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.create( source_code="source_code", ) asyncio.run(main()) """ _response = await self._raw_client.create( source_code=source_code, description=description, tags=tags, source_type=source_type, json_schema=json_schema, args_json_schema=args_json_schema, return_char_limit=return_char_limit, pip_requirements=pip_requirements, npm_requirements=npm_requirements, default_requires_approval=default_requires_approval, request_options=request_options, ) return _response.data async def upsert( self, *, source_code: str, description: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, source_type: typing.Optional[str] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, return_char_limit: typing.Optional[int] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, default_requires_approval: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> Tool: """ Create or update a tool Parameters ---------- source_code : str The source code of the function. description : typing.Optional[str] The description of the tool. tags : typing.Optional[typing.Sequence[str]] Metadata tags. source_type : typing.Optional[str] The source type of the function. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. return_char_limit : typing.Optional[int] The maximum number of characters in the response. pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. default_requires_approval : typing.Optional[bool] Whether or not to require approval before executing this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.upsert( source_code="source_code", ) asyncio.run(main()) """ _response = await self._raw_client.upsert( source_code=source_code, description=description, tags=tags, source_type=source_type, json_schema=json_schema, args_json_schema=args_json_schema, return_char_limit=return_char_limit, pip_requirements=pip_requirements, npm_requirements=npm_requirements, default_requires_approval=default_requires_approval, request_options=request_options, ) return _response.data async def upsert_base_tools(self, *, request_options: typing.Optional[RequestOptions] = None) -> typing.List[Tool]: """ Upsert base tools Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[Tool] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.upsert_base_tools() asyncio.run(main()) """ _response = await self._raw_client.upsert_base_tools(request_options=request_options) return _response.data async def run_tool_from_source( self, *, source_code: str, args: typing.Dict[str, typing.Optional[typing.Any]], env_vars: typing.Optional[typing.Dict[str, str]] = OMIT, name: typing.Optional[str] = OMIT, source_type: typing.Optional[str] = OMIT, args_json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, json_schema: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, pip_requirements: typing.Optional[typing.Sequence[PipRequirement]] = OMIT, npm_requirements: typing.Optional[typing.Sequence[NpmRequirement]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> ToolReturnMessage: """ Attempt to build a tool from source, then run it on the provided arguments Parameters ---------- source_code : str The source code of the function. args : typing.Dict[str, typing.Optional[typing.Any]] The arguments to pass to the tool. env_vars : typing.Optional[typing.Dict[str, str]] The environment variables to pass to the tool. name : typing.Optional[str] The name of the tool to run. source_type : typing.Optional[str] The type of the source code. args_json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The args JSON schema of the function. json_schema : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] The JSON schema of the function (auto-generated from source_code if not provided) pip_requirements : typing.Optional[typing.Sequence[PipRequirement]] Optional list of pip packages required by this tool. npm_requirements : typing.Optional[typing.Sequence[NpmRequirement]] Optional list of npm packages required by this tool. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ToolReturnMessage Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.run_tool_from_source( source_code="source_code", args={"key": "value"}, ) asyncio.run(main()) """ _response = await self._raw_client.run_tool_from_source( source_code=source_code, args=args, env_vars=env_vars, name=name, source_type=source_type, args_json_schema=args_json_schema, json_schema=json_schema, pip_requirements=pip_requirements, npm_requirements=npm_requirements, request_options=request_options, ) return _response.data async def list_composio_apps( self, *, user_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[AppModel]: """ Get a list of all Composio apps Parameters ---------- user_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[AppModel] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.list_composio_apps() asyncio.run(main()) """ _response = await self._raw_client.list_composio_apps(user_id=user_id, request_options=request_options) return _response.data async def list_composio_actions_by_app( self, composio_app_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[ActionModel]: """ Get a list of all Composio actions for a specific app Parameters ---------- composio_app_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[ActionModel] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.list_composio_actions_by_app( composio_app_name="composio_app_name", ) asyncio.run(main()) """ _response = await self._raw_client.list_composio_actions_by_app( composio_app_name, request_options=request_options ) return _response.data async def add_composio_tool( self, composio_action_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Tool: """ Add a new Composio tool by action name (Composio refers to each tool as an `Action`) Parameters ---------- composio_action_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.add_composio_tool( composio_action_name="composio_action_name", ) asyncio.run(main()) """ _response = await self._raw_client.add_composio_tool(composio_action_name, request_options=request_options) return _response.data async def list_mcp_servers( self, *, user_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> typing.Dict[str, ListMcpServersResponseValue]: """ Get a list of all configured MCP servers Parameters ---------- user_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Dict[str, ListMcpServersResponseValue] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.list_mcp_servers() asyncio.run(main()) """ _response = await self._raw_client.list_mcp_servers(user_id=user_id, request_options=request_options) return _response.data async def add_mcp_server( self, *, request: AddMcpServerRequest, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[AddMcpServerResponseItem]: """ Add a new MCP server to the Letta MCP server config Parameters ---------- request : AddMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[AddMcpServerResponseItem] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta, StdioServerConfig client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.add_mcp_server( request=StdioServerConfig( server_name="server_name", command="command", args=["args"], ), ) asyncio.run(main()) """ _response = await self._raw_client.add_mcp_server(request=request, request_options=request_options) return _response.data async def list_mcp_tools_by_server( self, mcp_server_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[McpTool]: """ Get a list of all tools for a specific MCP server Parameters ---------- mcp_server_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[McpTool] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.list_mcp_tools_by_server( mcp_server_name="mcp_server_name", ) asyncio.run(main()) """ _response = await self._raw_client.list_mcp_tools_by_server(mcp_server_name, request_options=request_options) return _response.data async def add_mcp_tool( self, mcp_server_name: str, mcp_tool_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Tool: """ Register a new MCP tool as a Letta server by MCP server + tool name Parameters ---------- mcp_server_name : str mcp_tool_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Tool Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.add_mcp_tool( mcp_server_name="mcp_server_name", mcp_tool_name="mcp_tool_name", ) asyncio.run(main()) """ _response = await self._raw_client.add_mcp_tool(mcp_server_name, mcp_tool_name, request_options=request_options) return _response.data async def delete_mcp_server( self, mcp_server_name: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.List[DeleteMcpServerResponseItem]: """ Delete a MCP server configuration Parameters ---------- mcp_server_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[DeleteMcpServerResponseItem] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.delete_mcp_server( mcp_server_name="mcp_server_name", ) asyncio.run(main()) """ _response = await self._raw_client.delete_mcp_server(mcp_server_name, request_options=request_options) return _response.data async def update_mcp_server( self, mcp_server_name: str, *, request: UpdateMcpServerRequest, request_options: typing.Optional[RequestOptions] = None, ) -> UpdateMcpServerResponse: """ Update an existing MCP server configuration Parameters ---------- mcp_server_name : str request : UpdateMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- UpdateMcpServerResponse Successful Response Examples -------- import asyncio from letta_client import AsyncLetta, UpdateStdioMcpServer client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.update_mcp_server( mcp_server_name="mcp_server_name", request=UpdateStdioMcpServer(), ) asyncio.run(main()) """ _response = await self._raw_client.update_mcp_server( mcp_server_name, request=request, request_options=request_options ) return _response.data async def test_mcp_server( self, *, request: TestMcpServerRequest, request_options: typing.Optional[RequestOptions] = None ) -> typing.Optional[typing.Any]: """ Test connection to an MCP server without adding it. Returns the list of available tools if successful. Parameters ---------- request : TestMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Optional[typing.Any] Successful Response Examples -------- import asyncio from letta_client import AsyncLetta, StdioServerConfig client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: await client.tools.test_mcp_server( request=StdioServerConfig( server_name="server_name", command="command", args=["args"], ), ) asyncio.run(main()) """ _response = await self._raw_client.test_mcp_server(request=request, request_options=request_options) return _response.data async def connect_mcp_server( self, *, request: ConnectMcpServerRequest, request_options: typing.Optional[RequestOptions] = None ) -> typing.AsyncIterator[StreamingResponse]: """ Connect to an MCP server with support for OAuth via SSE. Returns a stream of events handling authorization state and exchange if OAuth is required. Parameters ---------- request : ConnectMcpServerRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Yields ------ typing.AsyncIterator[StreamingResponse] Successful response Examples -------- import asyncio from letta_client import AsyncLetta, StdioServerConfig client = AsyncLetta( project="YOUR_PROJECT", token="YOUR_TOKEN", ) async def main() -> None: response = await client.tools.connect_mcp_server( request=StdioServerConfig( server_name="server_name", command="command", args=["args"], ), ) async for chunk in response: yield chunk asyncio.run(main()) """ async with self._raw_client.connect_mcp_server(request=request, request_options=request_options) as r: async for _chunk in r.data: yield _chunk