wehub-resource-sync
771 lines
29 KiB
Python
771 lines
29 KiB
Python
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",
|
|
]
|