Internal: create and validate a CustomTool.
(
slug: str,
*,
name: str,
description: str,
input_params: t.Type[BaseModel],
execute: CustomToolExecuteFn,
extends_toolkit: t.Optional[str] = None,
output_params: t.Optional[t.Type[BaseModel]] = None,
preload: t.Optional[bool] = None,
)
| 157 | |
| 158 | |
| 159 | def _create_tool( |
| 160 | slug: str, |
| 161 | *, |
| 162 | name: str, |
| 163 | description: str, |
| 164 | input_params: t.Type[BaseModel], |
| 165 | execute: CustomToolExecuteFn, |
| 166 | extends_toolkit: t.Optional[str] = None, |
| 167 | output_params: t.Optional[t.Type[BaseModel]] = None, |
| 168 | preload: t.Optional[bool] = None, |
| 169 | ) -> CustomTool: |
| 170 | """Internal: create and validate a CustomTool.""" |
| 171 | context = "experimental.tool" |
| 172 | |
| 173 | _validate_slug(slug, context) |
| 174 | |
| 175 | if not name: |
| 176 | raise ValidationError(f"{context}: name is required") |
| 177 | if not description: |
| 178 | raise ValidationError(f"{context}: description is required") |
| 179 | |
| 180 | if not isinstance(input_params, type) or not issubclass(input_params, BaseModel): |
| 181 | raise ValidationError( |
| 182 | f"{context}: input_params must be a Pydantic BaseModel subclass. " |
| 183 | f"Tool input parameters are always an object with named properties." |
| 184 | ) |
| 185 | |
| 186 | try: |
| 187 | from pydantic import RootModel |
| 188 | |
| 189 | if issubclass(input_params, RootModel): |
| 190 | raise ValidationError( |
| 191 | f"{context}: input_params must be a regular BaseModel with named fields, " |
| 192 | f"not a RootModel. Tool input parameters are always an object with " |
| 193 | f"named properties." |
| 194 | ) |
| 195 | except ImportError: |
| 196 | pass |
| 197 | |
| 198 | if not callable(execute): |
| 199 | raise ValidationError(f"{context}: execute must be a callable") |
| 200 | |
| 201 | if asyncio.iscoroutinefunction(execute): |
| 202 | raise ValidationError( |
| 203 | f"{context}: execute must be a synchronous function, not async. " |
| 204 | f"The Composio Python SDK is synchronous — use a regular " |
| 205 | f"'def fn(input, ctx)' instead of 'async def'." |
| 206 | ) |
| 207 | |
| 208 | _validate_slug_length(slug, extends_toolkit, context) |
| 209 | |
| 210 | input_schema = _get_input_json_schema(input_params) |
| 211 | |
| 212 | output_schema: t.Optional[t.Dict[str, t.Any]] = None |
| 213 | if output_params is not None: |
| 214 | if not isinstance(output_params, type) or not issubclass( |
| 215 | output_params, BaseModel |
| 216 | ): |
no test coverage detected
searching dependent graphs…