Skip to content

Tools API

Pass Python callables to Agent(tools=[...]). ToolRegistry extracts schemas and supports explicit name, description, and parameter overrides. Nonempty explicit description strings are preserved verbatim, including paragraphs and hard line breaks. None or an empty string uses the first docstring paragraph; a function without a docstring falls back to Call <tool name>. Parameter schemas are unchanged by description overrides. Its register() method returns a ToolDefinition, not a wrapper to pass back into Agent(tools=...). See Tool Usage.

Schema extraction is not runtime input validation or a security sandbox. Validate inputs inside the callable, constrain side effects, and inspect ToolResultEvent.error for execution failures. A custom executor must use a registry matching the agent's tools.

nagents.ToolRegistry

Registry for callable tools with automatic schema extraction.

Automatically extracts JSON Schema from function type hints and docstrings.

Example

registry = ToolRegistry()

def get_weather(city: str, unit: str = "celsius") -> str: '''Get the current weather for a city.''' return f"Weather in {city}: 20{unit[0].upper()}"

registry.register(get_weather) tools = registry.get_all()

Source code in src/nagents/tools/registry.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class ToolRegistry:
    """
    Registry for callable tools with automatic schema extraction.

    Automatically extracts JSON Schema from function type hints and docstrings.

    Example:
        registry = ToolRegistry()

        def get_weather(city: str, unit: str = "celsius") -> str:
            '''Get the current weather for a city.'''
            return f"Weather in {city}: 20{unit[0].upper()}"

        registry.register(get_weather)
        tools = registry.get_all()
    """

    def __init__(self) -> None:
        self._tools: dict[str, ToolDefinition] = {}

    def clear(self) -> None:
        """Clear all registered tools."""
        self._tools.clear()
        logger.debug("Cleared all tools from registry")

    def register(
        self,
        func: Callable[..., Any],
        name: str | None = None,
        description: str | None = None,
        parameters: JsonSchema | None = None,
    ) -> ToolDefinition:
        """
        Register a function as a tool.

        Args:
            func: The function to register
            name: Optional override for tool name (defaults to function name)
            description: Nonempty override preserved verbatim. None or an empty
                         string uses the first docstring paragraph.
            parameters: Optional JSON Schema for parameters. When provided,
                         skips auto-extraction from type hints.

        Returns:
            The created ToolDefinition
        """
        tool_name = name or func.__name__

        # Use provided schema or extract from type hints
        param_schema = parameters if parameters is not None else self._extract_parameters(func)

        # Explicit descriptions are model-facing instructions supplied by the
        # caller (including MCP/Designer metadata), not docstrings to summarize.
        # Keep the existing empty-string fallback while preserving all explicit
        # paragraphs, hard line breaks and Markdown whitespace.
        tool_description = description
        if not tool_description:
            docstring = func.__doc__ or f"Call {tool_name}"
            tool_description = docstring.strip().split("\n\n")[0].strip()

        tool_def = ToolDefinition(
            name=tool_name,
            description=tool_description,
            parameters=param_schema,
            func=func,
        )

        self._tools[tool_name] = tool_def
        logger.debug(f"Registered tool: {tool_name}")
        return tool_def

    def _extract_parameters(self, func: Callable[..., Any]) -> JsonSchema:
        """
        Extract JSON Schema from function type hints.

        Args:
            func: The function to extract parameters from

        Returns:
            JSON Schema dict for function parameters
        """
        try:
            hints = get_type_hints(func)
        except Exception:
            hints = {}

        sig = inspect.signature(func)

        properties: dict[str, JsonSchemaProperty] = {}
        required: list[str] = []

        for param_name, param in sig.parameters.items():
            if param_name in ("self", "cls"):
                continue

            param_type = hints.get(param_name, str)
            properties[param_name] = self._type_to_schema(param_type)

            # Add description from docstring if available
            param_doc = self._extract_param_doc(func, param_name)
            if param_doc:
                properties[param_name]["description"] = param_doc

            # Required if no default value
            if param.default is inspect.Parameter.empty:
                required.append(param_name)

        schema: JsonSchema = {
            "type": "object",
            "properties": properties,
        }

        if required:
            schema["required"] = required

        return schema

    def _type_to_schema(self, t: Any) -> JsonSchemaProperty:
        """
        Convert Python type to JSON Schema.

        Args:
            t: Python type annotation

        Returns:
            JSON Schema dict
        """
        origin = get_origin(t)
        args = get_args(t)

        # Handle Optional[X] and PEP 604 X | None -> X
        if origin is Union or origin is types.UnionType:
            # Filter out NoneType
            non_none_args = [a for a in args if a is not type(None)]
            if len(non_none_args) == 1:
                return self._type_to_schema(non_none_args[0])
            # Union of multiple types - use first one
            if non_none_args:
                return self._type_to_schema(non_none_args[0])

        # Handle list[X]
        if origin is list:
            item_type = args[0] if args else str
            return JsonSchemaProperty(
                type="array",
                items=self._type_to_schema(item_type),
            )

        # Handle dict[K, V]
        if origin is dict:
            return JsonSchemaProperty(type="object")

        # Basic types
        type_mapping: dict[type, JsonSchemaProperty] = {
            str: JsonSchemaProperty(type="string"),
            int: JsonSchemaProperty(type="integer"),
            float: JsonSchemaProperty(type="number"),
            bool: JsonSchemaProperty(type="boolean"),
            type(None): JsonSchemaProperty(type="null"),
        }

        if t in type_mapping:
            return type_mapping[t]

        # Default to string
        return JsonSchemaProperty(type="string")

    def _extract_param_doc(self, func: Callable[..., Any], param_name: str) -> str | None:
        """
        Extract parameter description from function docstring.

        Supports Google-style and Sphinx-style docstrings.

        Args:
            func: The function
            param_name: Name of the parameter

        Returns:
            Parameter description if found
        """
        if not func.__doc__:
            return None

        doc = func.__doc__

        # Try Google-style: "param_name: description" or "param_name (type): description"
        patterns = [
            rf"{param_name}\s*(?:\([^)]+\))?\s*:\s*(.+?)(?:\n|$)",  # Google style
            rf":param\s+{param_name}\s*:\s*(.+?)(?:\n|$)",  # Sphinx style
        ]

        for pattern in patterns:
            match = re.search(pattern, doc, re.IGNORECASE)
            if match:
                return match.group(1).strip()

        return None

    def get(self, name: str) -> ToolDefinition | None:
        """Get a tool by name."""
        return self._tools.get(name)

    def get_all(self) -> list[ToolDefinition]:
        """Get all registered tools."""
        return list(self._tools.values())

    def unregister(self, name: str) -> None:
        """Remove a tool by name; removing an absent tool is a no-op."""
        self._tools.pop(name, None)

    def has_tools(self) -> bool:
        """Check if any tools are registered."""
        return len(self._tools) > 0

    def names(self) -> list[str]:
        """Get all registered tool names."""
        return list(self._tools.keys())

register(func, name=None, description=None, parameters=None)

Register a function as a tool.

Parameters:

Name Type Description Default
func Callable[..., Any]

The function to register

 required
name str | None

Optional override for tool name (defaults to function name)

 None
description str | None

Nonempty override preserved verbatim. None or an empty string uses the first docstring paragraph.

 None
parameters JsonSchema | None

Optional JSON Schema for parameters. When provided, skips auto-extraction from type hints.

 None

Returns:

Type Description
ToolDefinition

The created ToolDefinition
Source code in src/nagents/tools/registry.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def register(
    self,
    func: Callable[..., Any],
    name: str | None = None,
    description: str | None = None,
    parameters: JsonSchema | None = None,
) -> ToolDefinition:
    """
    Register a function as a tool.

    Args:
        func: The function to register
        name: Optional override for tool name (defaults to function name)
        description: Nonempty override preserved verbatim. None or an empty
                     string uses the first docstring paragraph.
        parameters: Optional JSON Schema for parameters. When provided,
                     skips auto-extraction from type hints.

    Returns:
        The created ToolDefinition
    """
    tool_name = name or func.__name__

    # Use provided schema or extract from type hints
    param_schema = parameters if parameters is not None else self._extract_parameters(func)

    # Explicit descriptions are model-facing instructions supplied by the
    # caller (including MCP/Designer metadata), not docstrings to summarize.
    # Keep the existing empty-string fallback while preserving all explicit
    # paragraphs, hard line breaks and Markdown whitespace.
    tool_description = description
    if not tool_description:
        docstring = func.__doc__ or f"Call {tool_name}"
        tool_description = docstring.strip().split("\n\n")[0].strip()

    tool_def = ToolDefinition(
        name=tool_name,
        description=tool_description,
        parameters=param_schema,
        func=func,
    )

    self._tools[tool_name] = tool_def
    logger.debug(f"Registered tool: {tool_name}")
    return tool_def

get(name)

Get a tool by name.

Source code in src/nagents/tools/registry.py 
221
222
223
def get(self, name: str) -> ToolDefinition | None:
    """Get a tool by name."""
    return self._tools.get(name)

get_all()

Get all registered tools.

Source code in src/nagents/tools/registry.py 
225
226
227
def get_all(self) -> list[ToolDefinition]:
    """Get all registered tools."""
    return list(self._tools.values())

names()

Get all registered tool names.

Source code in src/nagents/tools/registry.py 
237
238
239
def names(self) -> list[str]:
    """Get all registered tool names."""
    return list(self._tools.keys())

has_tools()

Check if any tools are registered.

Source code in src/nagents/tools/registry.py 
233
234
235
def has_tools(self) -> bool:
    """Check if any tools are registered."""
    return len(self._tools) > 0

clear()

Clear all registered tools.

Source code in src/nagents/tools/registry.py 
43
44
45
46
def clear(self) -> None:
    """Clear all registered tools."""
    self._tools.clear()
    logger.debug("Cleared all tools from registry")

nagents.ToolExecutor

Executes tool calls.

Handles both sync and async functions: - Async functions are awaited directly - Sync functions are run in a thread pool via asyncio.to_thread()

Example

executor = ToolExecutor(registry) result_event = await executor.execute(tool_call)

Source code in src/nagents/tools/executor.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class ToolExecutor:
    """
    Executes tool calls.

    Handles both sync and async functions:
    - Async functions are awaited directly
    - Sync functions are run in a thread pool via asyncio.to_thread()

    Example:
        executor = ToolExecutor(registry)
        result_event = await executor.execute(tool_call)
    """

    def __init__(self, registry: ToolRegistry):
        """
        Initialize the executor.

        Args:
            registry: Tool registry to look up functions
        """
        self._registry = registry

    async def execute(self, tool_call: ToolCall) -> ToolResultEvent:
        """
        Execute a tool call and return result event.

        Sync functions are automatically run in a thread pool to avoid
        blocking the event loop.

        Args:
            tool_call: The tool call to execute

        Returns:
            ToolResultEvent with result or error
        """
        return await self._execute_definition(tool_call, self._registry.get(tool_call.name))

    async def _execute_definition(self, tool_call: ToolCall, tool: ToolDefinition | None) -> ToolResultEvent:
        """Execute the exact resolved definition; do not look it up again after approval."""
        start = time.monotonic()

        if not tool or not tool.func:
            available_tools = [definition.name for definition in self._registry.get_all()]
            if available_tools:
                tools_list = ", ".join(available_tools)
                error_msg = f"Tool '{tool_call.name}' does not exist. Available tools: {tools_list}"
            else:
                error_msg = f"Tool '{tool_call.name}' does not exist. No tools are available."
            return ToolResultEvent(
                id=tool_call.id,
                name=tool_call.name,
                error=error_msg,
                duration_ms=(time.monotonic() - start) * 1000,
            )

        try:
            func = tool.func
            kwargs = tool_call.arguments

            # Call the function (handle both sync and async)
            if inspect.iscoroutinefunction(func):
                # Async function - await directly
                result = await func(**kwargs)
            else:
                # Sync function - run in thread pool to avoid blocking
                result = await asyncio.to_thread(functools.partial(func, **kwargs))

            duration = (time.monotonic() - start) * 1000
            logger.debug(f"Tool {tool_call.name} executed in {duration:.2f}ms")

            return ToolResultEvent(
                id=tool_call.id,
                name=tool_call.name,
                result=result,
                duration_ms=duration,
            )

        except Exception as e:
            duration = (time.monotonic() - start) * 1000
            if tool_call.name.startswith("mcp__"):
                # MCP error text is untrusted response data and may contain credentials.
                logger.error("MCP tool invocation failed: tool=%s error_type=%s", tool_call.name, type(e).__name__)
            else:
                logger.exception(f"Tool {tool_call.name} failed")

            return ToolResultEvent(
                id=tool_call.id,
                name=tool_call.name,
                error=str(e),
                duration_ms=duration,
            )

__init__(registry)

Initialize the executor.

Parameters:

Name Type Description Default
registry ToolRegistry

Tool registry to look up functions

 required
Source code in src/nagents/tools/executor.py 
32
33
34
35
36
37
38
39
def __init__(self, registry: ToolRegistry):
    """
    Initialize the executor.

    Args:
        registry: Tool registry to look up functions
    """
    self._registry = registry

execute(tool_call) async

Execute a tool call and return result event.

Sync functions are automatically run in a thread pool to avoid blocking the event loop.

Parameters:

Name Type Description Default
tool_call ToolCall

The tool call to execute

 required

Returns:

Type Description
ToolResultEvent

ToolResultEvent with result or error
Source code in src/nagents/tools/executor.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
async def execute(self, tool_call: ToolCall) -> ToolResultEvent:
    """
    Execute a tool call and return result event.

    Sync functions are automatically run in a thread pool to avoid
    blocking the event loop.

    Args:
        tool_call: The tool call to execute

    Returns:
        ToolResultEvent with result or error
    """
    return await self._execute_definition(tool_call, self._registry.get(tool_call.name))

The underlying definition and schema types are in the Types API.