from __future__ import annotations import functools import typing as t from pathlib import Path import typing_extensions as te from pydantic import BaseModel as PydanticBaseModel from composio_client import omit from composio.client import HttpClient from composio.client.types import ( Tool, tool_execute_params, tool_proxy_params, tool_proxy_response, ) from composio.core.models._files import FileHelper from composio.core.models.base import Resource from composio.core.models.custom_tool_types import InlineCustomToolsWirePayload from composio.core.models.inline_custom_tools_payload import ( inline_custom_tools_execute_experimental, ) from composio.core.provider import TTool, TToolCollection from composio.core.provider.agentic import AgenticProvider, AgenticProviderExecuteFn from composio.core.provider.base import ExecuteToolFn from composio.core.provider.base import BaseProvider from composio.core.provider.none_agentic import NonAgenticProvider from composio.core.types import ToolkitVersionParam from composio.exceptions import InvalidParams, ToolVersionRequiredError from composio.utils.pydantic import none_to_omit from composio.utils.toolkit_version import get_toolkit_version from composio.utils.upload_dir_allowlist import resolve_effective_upload_allowlist from ._modifiers import ( Modifiers, ToolExecuteParams, after_execute, apply_modifier_by_type, before_execute, before_file_upload, merge_before_file_upload, schema_modifier, ) TOOL_ROUTER_SESSION_TOOLS_PAGE_LIMIT = 500 def _needs_serialization(obj: t.Any) -> bool: """Check if an object contains any Pydantic model instances.""" if isinstance(obj, PydanticBaseModel): return True if isinstance(obj, dict): return any(_needs_serialization(v) for v in obj.values()) if isinstance(obj, list): return any(_needs_serialization(item) for item in obj) return False def _serialize_value(obj: t.Any) -> t.Any: """Recursively convert Pydantic models to JSON-safe primitives.""" if isinstance(obj, PydanticBaseModel): return obj.model_dump(mode="json") if isinstance(obj, dict): return {k: _serialize_value(v) for k, v in obj.items()} if isinstance(obj, list): return [_serialize_value(item) for item in obj] return obj def _serialize_arguments(arguments: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]: """Serialize any Pydantic model instances in tool arguments to JSON-safe dicts. Prevents JSON serialization errors when arguments contain complex types (e.g., RootModel from discriminated union schemas). Returns the original dict unchanged when no Pydantic models are present. """ if not _needs_serialization(arguments): return arguments return {k: _serialize_value(v) for k, v in arguments.items()} class ToolExecutionResponse(te.TypedDict): data: t.Dict error: t.Optional[str] successful: bool class Tools(Resource, t.Generic[TTool, TToolCollection]): """ Tools class definition This class is used to manage tools in the Composio SDK. It provides methods to list, get, and execute tools. Generic Parameters: TTool: The individual tool type returned by the provider (e.g., ChatCompletionToolParam for OpenAI). TToolCollection: The collection type returned by get() (e.g., list[ChatCompletionToolParam]). The return type of get() is automatically inferred from the provider's generic parameters. This works for both built-in providers (OpenAI, Anthropic, etc.) and custom providers. """ provider: BaseProvider[TTool, TToolCollection] def __init__( self, client: HttpClient, provider: BaseProvider[TTool, TToolCollection], file_download_dir: t.Optional[str] = None, toolkit_versions: t.Optional[ToolkitVersionParam] = None, dangerously_allow_auto_upload_download_files: bool = False, sensitive_file_upload_protection: bool = True, file_upload_path_deny_segments: t.Optional[t.Sequence[str]] = None, file_upload_dirs: t.Union[t.Sequence[str], t.Literal[False], None] = None, ): """ Initialize the tools resource. :param client: The client to use for the tools resource. :param provider: The provider to use for the tools resource. :param file_download_dir: Output directory for downloadable files :param toolkit_versions: The versions of the toolkits to use. Defaults to 'latest' if not provided. :param dangerously_allow_auto_upload_download_files: Opt-in for automatic file upload/download. Defaults to False. :param sensitive_file_upload_protection: When True, block local paths on the built-in sensitive-path denylist before upload. :param file_upload_path_deny_segments: Extra path segment names to merge with the built-in denylist. :param file_upload_dirs: Allowlist of directories from which local files may be auto-uploaded. Only consulted when ``dangerously_allow_auto_upload_download_files=True``. See Composio docs for details. """ self._auto_upload_download_files = bool( dangerously_allow_auto_upload_download_files ) # Resolve allowlist once, but only if auto-upload is enabled. When it's # disabled there's no auto-upload code path, so we pass ``None`` and # FileHelper won't run the allowlist check either way. Manual APIs (if # ever added in the future) should also pass ``None``. resolved_allowlist: t.Optional[t.List[Path]] = ( resolve_effective_upload_allowlist(file_upload_dirs) if self._auto_upload_download_files else None ) self._client = client self._tool_schemas: t.Dict[str, Tool] = {} self._file_helper = FileHelper( client=self._client, outdir=file_download_dir, sensitive_file_upload_protection=sensitive_file_upload_protection, file_upload_path_deny_segments=file_upload_path_deny_segments, file_upload_allowlist=resolved_allowlist, ) self._toolkit_versions = toolkit_versions self.provider = provider self.provider.set_execute_tool_fn( t.cast( ExecuteToolFn, functools.partial( self.execute, # Dangerously skip version check for provider controlled tool execution dangerously_skip_version_check=True, ), ), ) def get_raw_composio_tool_by_slug(self, slug: str) -> Tool: """ Returns schema for the given tool slug. """ return t.cast( Tool, self._client.tools.retrieve( tool_slug=slug, toolkit_versions=none_to_omit(self._toolkit_versions), ), ) def get_raw_composio_tools( self, tools: t.Optional[list[str]] = None, search: t.Optional[str] = None, toolkits: t.Optional[list[str]] = None, scopes: t.Optional[t.List[str]] = None, limit: t.Optional[int] = None, ) -> list[Tool]: """ Get a list of tool schemas based on the provided filters. """ if tools is None and search is None and toolkits is None: raise InvalidParams( "Either `tools`, `search`, or `toolkits` must be provided" ) tools_list = [] if tools is not None: if len(tools): tools_list.extend( self._client.tools.list( tool_slugs=",".join(tools), toolkit_versions=none_to_omit(self._toolkit_versions), ).items ) # Search tools by toolkit slugs and search term if toolkits is not None or search is not None: tools_list.extend( self._client.tools.list( toolkit_slug=none_to_omit(",".join(toolkits) if toolkits else None), search=none_to_omit(search), scopes=scopes, limit=limit, toolkit_versions=none_to_omit(self._toolkit_versions), ).items ) return tools_list def get_raw_tool_router_meta_tools( self, session_id: str, modifiers: t.Optional["Modifiers"] = None, ) -> list[Tool]: """ Fetches the tools exposed by a tool router session. This method fetches helper/meta tools and any preloaded app tools from the Composio API and transforms them to the expected format. It provides access to the underlying tool data without provider-specific wrapping. :param session_id: The session ID to get the meta tools for :param modifiers: Optional modifiers to apply to the tool schemas :return: The list of meta tools Example: ```python from composio import Composio composio = Composio() tools_model = composio.tools # Get meta tools for a session meta_tools = tools_model.get_raw_tool_router_meta_tools("session_123") print(meta_tools) # Get meta tools with schema modifiers from composio.core.models import schema_modifier @schema_modifier def modify_schema(tool: str, toolkit: str, schema): # Customize the schema schema.description = f"Modified: {schema.description}" return schema meta_tools = tools_model.get_raw_tool_router_meta_tools( "session_123", modifiers=[modify_schema] ) ``` """ # Fetch session tools from the API tools_list: t.List[Tool] = [] cursor: t.Optional[str] = None while True: tools_response = self._client.tool_router.session.tools( session_id=session_id, cursor=none_to_omit(cursor), limit=TOOL_ROUTER_SESSION_TOOLS_PAGE_LIMIT, ) # Cast to Tool type - session.tools returns compatible Item type from different response schema tools_list.extend(t.cast(Tool, item) for item in tools_response.items) cursor = getattr(tools_response, "next_cursor", None) if not cursor: break # Apply schema modifiers if provided if modifiers is not None: from composio.core.models._modifiers import apply_modifier_by_type tools_list = [ t.cast( Tool, apply_modifier_by_type( modifiers=modifiers, toolkit=tool.toolkit.slug, tool=tool.slug, type="schema", schema=tool, ), ) for tool in tools_list ] self._tool_schemas.update( {tool.slug: tool.model_copy(deep=True) for tool in tools_list} ) return tools_list def _get( self, user_id: str, tools: t.Optional[list[str]] = None, search: t.Optional[str] = None, toolkits: t.Optional[list[str]] = None, scopes: t.Optional[t.List[str]] = None, modifiers: t.Optional[Modifiers] = None, limit: t.Optional[int] = None, ) -> TToolCollection: """Get a list of tools based on the provided filters.""" tools_list = self.get_raw_composio_tools( tools=tools, search=search, toolkits=toolkits, scopes=scopes, limit=limit, ) if modifiers is not None: tools_list = [ apply_modifier_by_type( modifiers=modifiers, toolkit=tool.toolkit.slug, tool=tool.slug, type="schema", schema=tool, ) for tool in tools_list ] self._tool_schemas.update( {tool.slug: tool.model_copy(deep=True) for tool in tools_list} ) # Always enhance schema descriptions (type hints and required notes) # regardless of dangerously_allow_auto_upload_download_files for tool in tools_list: tool.input_parameters = self._file_helper.enhance_schema_descriptions( schema=tool.input_parameters, ) # Only process file_uploadable schemas when the caller opted in via # `dangerously_allow_auto_upload_download_files=True`. if self._auto_upload_download_files: for tool in tools_list: tool.input_parameters = ( self._file_helper.process_file_uploadable_schema( schema=tool.input_parameters, ) ) if issubclass(type(self.provider), NonAgenticProvider): return t.cast( TToolCollection, t.cast( NonAgenticProvider[TTool, TToolCollection], self.provider ).wrap_tools(tools=tools_list), ) return t.cast( TToolCollection, t.cast(AgenticProvider[TTool, TToolCollection], self.provider).wrap_tools( tools=tools_list, execute_tool=self._wrap_execute_tool( user_id=user_id, modifiers=modifiers, ), ), ) def get( self, user_id: str, *, slug: t.Optional[str] = None, tools: t.Optional[list[str]] = None, search: t.Optional[str] = None, toolkits: t.Optional[list[str]] = None, scopes: t.Optional[t.List[str]] = None, modifiers: t.Optional[Modifiers] = None, limit: t.Optional[int] = None, ) -> TToolCollection: """ Get a tool or list of tools based on the provided arguments. The return type is automatically inferred based on the provider's generic parameters. For example: - OpenAIProvider -> list[ChatCompletionToolParam] - AnthropicProvider -> list[ToolParam] - CustomProvider[MyTool, list[MyTool]] -> list[MyTool] :param user_id: The user ID to get tools for. :param slug: Get a single tool by slug. :param tools: Get tools by a list of tool slugs. :param search: Search tools by search term. :param toolkits: Get tools from specific toolkits. :param scopes: Filter by scopes. :param modifiers: Optional modifiers to apply. :param limit: Limit the number of tools returned. :return: Provider-specific tool collection (TToolCollection). """ if slug is not None: return self._get( user_id=user_id, tools=[slug], modifiers=modifiers, ) return self._get( user_id=user_id, tools=tools, search=search, toolkits=toolkits, scopes=scopes, modifiers=modifiers, limit=limit, ) def _wrap_execute_tool( self, modifiers: t.Optional[Modifiers] = None, user_id: t.Optional[str] = None, ) -> AgenticProviderExecuteFn: """Wrap the execute tool function""" return t.cast( AgenticProviderExecuteFn, functools.partial( self.execute, modifiers=modifiers, user_id=user_id, # Dangerously skip version check for agentic tool execution via providers # This can be safe because most agentic flows users fetch latest version and then execute the tool dangerously_skip_version_check=True, ), ) def _wrap_execute_tool_for_tool_router( self, session_id: str, modifiers: t.Optional[Modifiers] = None, inline_custom_tools_payload: t.Optional[InlineCustomToolsWirePayload] = None, ) -> AgenticProviderExecuteFn: """ Create an execute function for tool router session tools. This method creates a function that executes tools within a tool router session context. It uses the session's execute endpoint which handles authentication and connection management automatically. :param session_id: The session ID :param modifiers: Optional modifiers to apply before and after execution :return: Execute function for tool router """ def execute_tool_fn(slug: str, arguments: t.Dict) -> t.Dict: """ Execute a tool in the tool router session. This function is used by agentic providers to execute tools within a tool router session context. It uses the session's execute endpoint which handles authentication and connection management automatically. :param slug: The tool slug to execute :param arguments: The tool arguments :return: Tool execution response """ tool = self._tool_schemas.get(slug) if tool is None: tool = t.cast( Tool, self._client.tools.retrieve( tool_slug=slug, toolkit_versions=none_to_omit(self._toolkit_versions), ), ) self._tool_schemas[slug] = tool if self._auto_upload_download_files: meta_tk = tool.toolkit.slug if tool.toolkit else "composio" bfu = merge_before_file_upload( modifiers, tool=slug, toolkit=meta_tk, ) arguments = self._file_helper.substitute_file_uploads( tool=tool, request=arguments, before_file_upload=bfu, ) toolkit_slug = tool.toolkit.slug if tool.toolkit else "composio" # Apply before_execute modifiers processed_arguments = arguments if modifiers is not None: params: ToolExecuteParams = { "arguments": arguments, } type_before: t.Literal["before_execute"] = "before_execute" modified_params = apply_modifier_by_type( modifiers=modifiers, toolkit=toolkit_slug, tool=slug, type=type_before, request=params, ) processed_arguments = modified_params.get("arguments", arguments) # Serialize any Pydantic model instances before sending to the API processed_arguments = _serialize_arguments(processed_arguments) response = self._client.tool_router.session.execute( session_id=session_id, tool_slug=slug, arguments=processed_arguments, # Provider-wrapped session tools are agentic calls, so they opt into # direct tool offload when the backend session workbench allows it. enable_auto_workbench_offload=True, experimental=inline_custom_tools_execute_experimental( inline_custom_tools_payload ), ) # Convert response to standard format result: ToolExecutionResponse = { "data": response.data if hasattr(response, "data") else {}, "error": response.error if hasattr(response, "error") else None, "successful": not (hasattr(response, "error") and response.error), } # Apply after_execute modifiers if modifiers is not None: type_after: t.Literal["after_execute"] = "after_execute" result = apply_modifier_by_type( modifiers=modifiers, toolkit=toolkit_slug, tool=slug, type=type_after, response=result, ) return t.cast(t.Dict, result) return t.cast(AgenticProviderExecuteFn, execute_tool_fn) def _execute_tool( self, slug: str, arguments: t.Dict, connected_account_id: t.Optional[str] = None, custom_auth_params: t.Optional[tool_execute_params.CustomAuthParams] = None, custom_connection_data: t.Optional[ tool_execute_params.CustomConnectionData ] = None, user_id: t.Optional[str] = None, text: t.Optional[str] = None, version: t.Optional[str] = None, dangerously_skip_version_check: t.Optional[bool] = None, ) -> ToolExecutionResponse: """Execute a tool""" # Serialize any Pydantic model instances in arguments to plain dicts # before sending to the API (fixes PLEN-1514: RootModel not JSON serializable) arguments = _serialize_arguments(arguments) # Get the tool to determine its toolkit tool = self.get_raw_composio_tool_by_slug(slug) # If version is not explicitly provided, resolve it from instance-level toolkit versions # This matches the TypeScript behavior - always resolve version when None if version is None: toolkit_slug = tool.toolkit.slug if tool.toolkit else "unknown" # Use instance-level toolkit versions configuration version = get_toolkit_version(toolkit_slug, self._toolkit_versions) # Check if the version is 'latest' and dangerously_skip_version_check is not True # If so, raise an error to prevent unexpected behavior if version == "latest" and not dangerously_skip_version_check: raise ToolVersionRequiredError() return t.cast( ToolExecutionResponse, # Disable retries: tool execution is a non-idempotent write, and a # silent retry after a read timeout can duplicate the side effect. self._client.without_retries.tools.execute( tool_slug=slug, arguments=arguments, connected_account_id=none_to_omit(connected_account_id), custom_auth_params=none_to_omit(custom_auth_params), custom_connection_data=none_to_omit(custom_connection_data), user_id=none_to_omit(user_id), text=none_to_omit(text), version=none_to_omit(version), ).model_dump( exclude={ "log_id", "session_info", } ), ) def execute( self, slug: str, arguments: t.Dict, *, connected_account_id: t.Optional[str] = None, custom_auth_params: t.Optional[tool_execute_params.CustomAuthParams] = None, custom_connection_data: t.Optional[ tool_execute_params.CustomConnectionData ] = None, user_id: t.Optional[str] = None, text: t.Optional[str] = None, version: t.Optional[str] = None, dangerously_skip_version_check: t.Optional[bool] = None, modifiers: t.Optional[Modifiers] = None, ) -> ToolExecutionResponse: """ Execute a tool with the provided parameters. This method calls the Composio API to execute the tool and returns the response. :param slug: The slug of the tool to execute. :param arguments: The arguments to pass to the tool. :param connected_account_id: The ID of the connected account to use for the tool. :param custom_auth_params: The custom auth params to use for the tool. :param custom_connection_data: The custom connection data to use for the tool, takes priority over custom_auth_params. :param user_id: The ID of the user to execute the tool for. :param text: The text to pass to the tool. :param version: The version of the tool to execute (overrides the SDK-level toolkit versions for this execution). :param dangerously_skip_version_check: Skip the version check for 'latest' version. This might cause unexpected behavior when new versions are released. :param modifiers: The modifiers to apply to the tool (include ``@before_file_upload`` for file path hooks). :return: The response from the tool. """ tool = self._tool_schemas.get(slug) if tool is None: tool = t.cast( Tool, self._client.tools.retrieve( tool_slug=slug, toolkit_versions=none_to_omit(self._toolkit_versions), ), ) self._tool_schemas[slug] = tool if self._auto_upload_download_files: tk = tool.toolkit.slug if tool.toolkit else "unknown" bfu = merge_before_file_upload( modifiers, tool=slug, toolkit=tk, ) arguments = self._file_helper.substitute_file_uploads( tool=tool, request=arguments, before_file_upload=bfu, ) if modifiers is not None: type_before_exec: t.Literal["before_execute"] = "before_execute" request_params: ToolExecuteParams = { "arguments": arguments, } if connected_account_id is not None: request_params["connected_account_id"] = connected_account_id if custom_auth_params is not None: request_params["custom_auth_params"] = custom_auth_params if custom_connection_data is not None: request_params["custom_connection_data"] = custom_connection_data if version is not None: request_params["version"] = version if text is not None: request_params["text"] = text if user_id is not None: request_params["user_id"] = user_id if dangerously_skip_version_check is not None: request_params["dangerously_skip_version_check"] = ( dangerously_skip_version_check ) processed_params = apply_modifier_by_type( modifiers=modifiers, toolkit=tool.toolkit.slug, tool=slug, type=type_before_exec, request=request_params, ) connected_account_id = processed_params.get( "connected_account_id", connected_account_id ) custom_auth_params = processed_params.get( "custom_auth_params", custom_auth_params ) custom_connection_data = processed_params.get( "custom_connection_data", custom_connection_data ) text = processed_params.get("text", text) version = processed_params.get("version", version) user_id = processed_params.get("user_id", user_id) arguments = processed_params["arguments"] dangerously_skip_version_check = processed_params.get( "dangerously_skip_version_check", dangerously_skip_version_check ) response = self._execute_tool( slug=slug, arguments=arguments, connected_account_id=connected_account_id, custom_auth_params=custom_auth_params, custom_connection_data=custom_connection_data, user_id=user_id, text=text, version=version, dangerously_skip_version_check=dangerously_skip_version_check, ) if self._auto_upload_download_files: response = self._file_helper.substitute_file_downloads( tool=tool, response=response, ) if modifiers is not None: response = apply_modifier_by_type( modifiers=modifiers, toolkit=tool.toolkit.slug, tool=slug, type="after_execute", response=response, ) return response def proxy( self, endpoint: str, method: t.Literal["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD"], body: t.Optional[object] = None, connected_account_id: t.Optional[str] = None, parameters: t.Optional[t.List[tool_proxy_params.Parameter]] = None, custom_connection_data: t.Optional[ tool_proxy_params.CustomConnectionData ] = None, ) -> tool_proxy_response.ToolProxyResponse: """Proxy a tool call to the Composio API""" # Disable retries: a proxied call is a non-idempotent write, and a silent # retry after a read timeout can duplicate the side effect. return self._client.without_retries.tools.proxy( endpoint=endpoint, method=method, body=body if body is not None else omit, connected_account_id=connected_account_id if connected_account_id is not None else omit, parameters=parameters if parameters is not None else omit, custom_connection_data=custom_connection_data if custom_connection_data is not None else omit, ) __all__ = [ "Tools", "ToolExecuteParams", "ToolExecutionResponse", "Modifiers", "after_execute", "before_execute", "before_file_upload", "schema_modifier", ]