composiohq__composio/python/composio/core/models/tools.py

from __future__ import annotations

import functools
import typing as t
from collections.abc import Mapping
from pathlib import Path

import typing_extensions as te
from composio_client import APIStatusError, omit
from pydantic import BaseModel as PydanticBaseModel

from composio.client import HttpClient
from composio.client.types import (
    Tool,
    ToolkitMinimal,
    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 BaseProvider, ExecuteToolFn
from composio.core.provider.none_agentic import NonAgenticProvider
from composio.core.types import ToolkitVersionParam
from composio.exceptions import (
    InvalidParams,
    ToolNotFoundError,
    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 _normalize_tool(tool: PydanticBaseModel | Mapping[str, object]) -> Tool:
    """Normalize generated-client responses to the SDK's tool model shape."""
    normalized: PydanticBaseModel
    if isinstance(tool, Mapping):
        normalized = Tool.model_construct(_fields_set=set(tool), **dict(tool))
    else:
        normalized = tool

    toolkit = getattr(normalized, "toolkit", None)
    if isinstance(toolkit, Mapping):
        normalized_toolkit = ToolkitMinimal.model_construct(
            _fields_set=set(toolkit), **dict(toolkit)
        )
        normalized = normalized.model_copy(update={"toolkit": normalized_toolkit})

    return t.cast(Tool, normalized)


def _toolkit_slug(tool: Tool, fallback: str) -> str:
    """Resolve untrusted toolkit metadata without assuming a generated shape."""
    toolkit = getattr(tool, "toolkit", None)
    slug = (
        toolkit.get("slug")
        if isinstance(toolkit, Mapping)
        else getattr(toolkit, "slug", None)
    )
    return slug if isinstance(slug, str) and slug else fallback


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.

        :raises ToolNotFoundError: when the backend reports the slug as unknown
            (404, or 400 for a malformed slug). Any other client error, such as
            an invalid API key, is re-raised unchanged.
        """
        try:
            response = self._client.tools.retrieve(
                tool_slug=slug,
                toolkit_versions=none_to_omit(self._toolkit_versions),
            )
        except APIStatusError as error:
            if error.status_code in (400, 404):
                raise ToolNotFoundError(f"Tool with slug {slug} not found") from error
            raise
        return _normalize_tool(response)

    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 [_normalize_tool(tool) for tool in 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,
            )
            tools_list.extend(_normalize_tool(item) for item in tools_response.items)

            cursor = getattr(tools_response, "next_cursor", None)
            if not cursor:
                break

        self._tool_schemas.update(
            {tool.slug: tool.model_copy(deep=True) for tool in tools_list}
        )

        tools_list = [tool.model_copy(deep=True) for tool in tools_list]

        # 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=_toolkit_slug(tool, "unknown"),
                        tool=tool.slug,
                        type="schema",
                        schema=tool,
                    ),
                )
                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."""
        raw_tools = self.get_raw_composio_tools(
            tools=tools,
            search=search,
            toolkits=toolkits,
            scopes=scopes,
            limit=limit,
        )
        self._tool_schemas.update(
            {tool.slug: tool.model_copy(deep=True) for tool in raw_tools}
        )

        tools_list = [tool.model_copy(deep=True) for tool in raw_tools]
        if modifiers is not None:
            tools_list = [
                apply_modifier_by_type(
                    modifiers=modifiers,
                    toolkit=_toolkit_slug(tool, "unknown"),
                    tool=tool.slug,
                    type="schema",
                    schema=tool,
                )
                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 = self.get_raw_composio_tool_by_slug(slug)
                self._tool_schemas[slug] = tool

            if self._auto_upload_download_files:
                meta_tk = _toolkit_slug(tool, "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,
                )
            else:
                arguments = self._file_helper.drop_empty_file_uploads(
                    tool=tool, request=arguments
                )

            toolkit_slug = _toolkit_slug(tool, "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,
        tool: Tool,
        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)

        # 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 = _toolkit_slug(tool, "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 = self.get_raw_composio_tool_by_slug(slug)
            self._tool_schemas[slug] = tool

        if self._auto_upload_download_files:
            tk = _toolkit_slug(tool, "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,
            )
        else:
            # Auto-upload is opt-in, but "no file" must still be sent as an
            # omitted key rather than ``""`` (issue #4233). Explicit ``None``
            # remains valid for nullable file inputs.
            arguments = self._file_helper.drop_empty_file_uploads(
                tool=tool, request=arguments
            )

        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=_toolkit_slug(tool, "unknown"),
                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,
            tool=tool,
            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=_toolkit_slug(tool, "unknown"),
                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",
]