Skip to content

Tools

Tool (Protocol)

Tool

Bases: Protocol

Protocol for tool implementations.

Tools are pluggable components that agents can use to interact with external systems (databases, APIs, caches, etc.).

Using Protocol so that third-party tool implementations can satisfy this interface without depending on AgenticAPI.

Source code in src/agenticapi/runtime/tools/base.py 
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
@runtime_checkable
class Tool(Protocol):
    """Protocol for tool implementations.

    Tools are pluggable components that agents can use to interact
    with external systems (databases, APIs, caches, etc.).

    Using Protocol so that third-party tool implementations can
    satisfy this interface without depending on AgenticAPI.
    """

    @property
    def definition(self) -> ToolDefinition:
        """Return the tool's metadata definition."""
        ...

    async def invoke(self, **kwargs: Any) -> Any:
        """Invoke the tool with the given keyword arguments.

        Args:
            **kwargs: Tool-specific parameters.

        Returns:
            The tool's result (type depends on the tool).
        """
        ...

definition property 

definition: ToolDefinition

Return the tool's metadata definition.

invoke async 

invoke(**kwargs: Any) -> Any

Invoke the tool with the given keyword arguments.

Parameters:

Name Type Description Default
**kwargs Any

Tool-specific parameters.

 {}

Returns:

Type Description
Any

The tool's result (type depends on the tool).
Source code in src/agenticapi/runtime/tools/base.py 
61
62
63
64
65
66
67
68
69
70
async def invoke(self, **kwargs: Any) -> Any:
    """Invoke the tool with the given keyword arguments.

    Args:
        **kwargs: Tool-specific parameters.

    Returns:
        The tool's result (type depends on the tool).
    """
    ...

ToolDefinition

ToolDefinition dataclass

Metadata describing a tool's interface and capabilities.

Attributes:

Name Type Description
name str

Unique identifier for the tool.
description str

Human-readable description of what the tool does.
capabilities list[ToolCapability]

List of capabilities this tool provides.
parameters_schema dict[str, Any]

JSON Schema describing the tool's parameters.
Source code in src/agenticapi/runtime/tools/base.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True, slots=True)
class ToolDefinition:
    """Metadata describing a tool's interface and capabilities.

    Attributes:
        name: Unique identifier for the tool.
        description: Human-readable description of what the tool does.
        capabilities: List of capabilities this tool provides.
        parameters_schema: JSON Schema describing the tool's parameters.
    """

    name: str
    description: str
    capabilities: list[ToolCapability]
    parameters_schema: dict[str, Any] = field(default_factory=dict)

ToolCapability

ToolCapability

Bases: StrEnum

Capabilities that a tool can provide.

Used for policy evaluation to determine what operations a tool is permitted to perform.

Source code in src/agenticapi/runtime/tools/base.py 
14
15
16
17
18
19
20
21
22
23
24
25
class ToolCapability(StrEnum):
    """Capabilities that a tool can provide.

    Used for policy evaluation to determine what operations
    a tool is permitted to perform.
    """

    READ = "read"
    WRITE = "write"
    AGGREGATE = "aggregate"
    SEARCH = "search"
    EXECUTE = "execute"

ToolRegistry

ToolRegistry

Registry for managing available tools.

Manages tool registration, lookup by name, and provides tool definitions for code generation prompts.

Example

registry = ToolRegistry() registry.register(database_tool) tool = registry.get("database") definitions = registry.get_definitions()

Source code in src/agenticapi/runtime/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
class ToolRegistry:
    """Registry for managing available tools.

    Manages tool registration, lookup by name, and provides
    tool definitions for code generation prompts.

    Example:
        registry = ToolRegistry()
        registry.register(database_tool)
        tool = registry.get("database")
        definitions = registry.get_definitions()
    """

    def __init__(self, tools: list[Tool | Callable[..., Any]] | None = None) -> None:
        """Initialize the registry with optional initial tools.

        Args:
            tools: Optional list of tools to register immediately.
                Each entry may be either a :class:`Tool` instance or a
                plain function — plain functions are auto-wrapped via
                :func:`agenticapi.runtime.tools.tool` so you can mix
                both styles freely.
        """
        self._tools: dict[str, Tool] = {}
        if tools:
            for tool in tools:
                self.register(tool)

    def register(self, tool: Tool | Callable[..., Any]) -> None:
        """Register a tool in the registry.

        Args:
            tool: The tool to register. May be either a :class:`Tool`
                instance or a plain (sync or async) function. Plain
                functions are automatically wrapped via the
                :func:`agenticapi.runtime.tools.tool` decorator so the
                FastAPI-style ``register(my_func)`` shortcut works.

        Raises:
            ToolError: If a tool with the same name is already registered.
        """
        # Auto-wrap plain functions for the FastAPI-style ergonomic
        # ``registry.register(my_func)`` form. We detect "is this a
        # Tool already" by checking for the protocol's ``definition``
        # attribute, which is more reliable than ``isinstance(...)``
        # against a runtime-checkable Protocol.
        if not (hasattr(tool, "definition") and hasattr(tool, "invoke")):
            if not callable(tool):
                raise ToolError(f"Cannot register non-callable, non-Tool object: {tool!r}")
            from agenticapi.runtime.tools.decorator import tool as _tool_decorator

            tool = _tool_decorator(tool)

        # By this point ``tool`` definitely satisfies the protocol.
        name = tool.definition.name
        if name in self._tools:
            raise ToolError(f"Tool '{name}' is already registered")
        self._tools[name] = tool
        logger.info(
            "tool_registered",
            tool_name=name,
            capabilities=[c.value for c in tool.definition.capabilities],
        )

    def get(self, name: str) -> Tool:
        """Look up a tool by name.

        Args:
            name: The name of the tool to retrieve.

        Returns:
            The registered tool.

        Raises:
            ToolError: If no tool with the given name is registered.
        """
        tool = self._tools.get(name)
        if tool is None:
            available = list(self._tools.keys())
            raise ToolError(f"Tool '{name}' not found. Available tools: {available}")
        return tool

    def list_tools(self) -> list[ToolDefinition]:
        """Return definitions for all registered tools.

        Returns:
            List of ToolDefinition objects for all registered tools.
        """
        return [tool.definition for tool in self._tools.values()]

    def get_definitions(self) -> list[ToolDefinition]:
        """Return definitions for all registered tools.

        Alias for list_tools() for API consistency.

        Returns:
            List of ToolDefinition objects for all registered tools.
        """
        return self.list_tools()

    def __len__(self) -> int:
        return len(self._tools)

    def __contains__(self, name: str) -> bool:
        return name in self._tools

__init__ 

__init__(
    tools: list[Tool | Callable[..., Any]] | None = None,
) -> None

Initialize the registry with optional initial tools.

Parameters:

Name Type Description Default
tools list[Tool | Callable[..., Any]] | None

Optional list of tools to register immediately. Each entry may be either a :class:Tool instance or a plain function — plain functions are auto-wrapped via :func:agenticapi.runtime.tools.tool so you can mix both styles freely.

 None
Source code in src/agenticapi/runtime/tools/registry.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self, tools: list[Tool | Callable[..., Any]] | None = None) -> None:
    """Initialize the registry with optional initial tools.

    Args:
        tools: Optional list of tools to register immediately.
            Each entry may be either a :class:`Tool` instance or a
            plain function — plain functions are auto-wrapped via
            :func:`agenticapi.runtime.tools.tool` so you can mix
            both styles freely.
    """
    self._tools: dict[str, Tool] = {}
    if tools:
        for tool in tools:
            self.register(tool)

register 

register(tool: Tool | Callable[..., Any]) -> None

Register a tool in the registry.

Parameters:

Name Type Description Default
tool Tool | Callable[..., Any]

The tool to register. May be either a :class:Tool instance or a plain (sync or async) function. Plain functions are automatically wrapped via the :func:agenticapi.runtime.tools.tool decorator so the FastAPI-style register(my_func) shortcut works.

 required

Raises:

Type Description
ToolError

If a tool with the same name is already registered.
Source code in src/agenticapi/runtime/tools/registry.py 
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
def register(self, tool: Tool | Callable[..., Any]) -> None:
    """Register a tool in the registry.

    Args:
        tool: The tool to register. May be either a :class:`Tool`
            instance or a plain (sync or async) function. Plain
            functions are automatically wrapped via the
            :func:`agenticapi.runtime.tools.tool` decorator so the
            FastAPI-style ``register(my_func)`` shortcut works.

    Raises:
        ToolError: If a tool with the same name is already registered.
    """
    # Auto-wrap plain functions for the FastAPI-style ergonomic
    # ``registry.register(my_func)`` form. We detect "is this a
    # Tool already" by checking for the protocol's ``definition``
    # attribute, which is more reliable than ``isinstance(...)``
    # against a runtime-checkable Protocol.
    if not (hasattr(tool, "definition") and hasattr(tool, "invoke")):
        if not callable(tool):
            raise ToolError(f"Cannot register non-callable, non-Tool object: {tool!r}")
        from agenticapi.runtime.tools.decorator import tool as _tool_decorator

        tool = _tool_decorator(tool)

    # By this point ``tool`` definitely satisfies the protocol.
    name = tool.definition.name
    if name in self._tools:
        raise ToolError(f"Tool '{name}' is already registered")
    self._tools[name] = tool
    logger.info(
        "tool_registered",
        tool_name=name,
        capabilities=[c.value for c in tool.definition.capabilities],
    )

get 

get(name: str) -> Tool

Look up a tool by name.

Parameters:

Name Type Description Default
name str

The name of the tool to retrieve.

 required

Returns:

Type Description
Tool

The registered tool.

Raises:

Type Description
ToolError

If no tool with the given name is registered.
Source code in src/agenticapi/runtime/tools/registry.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get(self, name: str) -> Tool:
    """Look up a tool by name.

    Args:
        name: The name of the tool to retrieve.

    Returns:
        The registered tool.

    Raises:
        ToolError: If no tool with the given name is registered.
    """
    tool = self._tools.get(name)
    if tool is None:
        available = list(self._tools.keys())
        raise ToolError(f"Tool '{name}' not found. Available tools: {available}")
    return tool

list_tools 

list_tools() -> list[ToolDefinition]

Return definitions for all registered tools.

Returns:

Type Description
list[ToolDefinition]

List of ToolDefinition objects for all registered tools.
Source code in src/agenticapi/runtime/tools/registry.py 
105
106
107
108
109
110
111
def list_tools(self) -> list[ToolDefinition]:
    """Return definitions for all registered tools.

    Returns:
        List of ToolDefinition objects for all registered tools.
    """
    return [tool.definition for tool in self._tools.values()]

get_definitions 

get_definitions() -> list[ToolDefinition]

Return definitions for all registered tools.

Alias for list_tools() for API consistency.

Returns:

Type Description
list[ToolDefinition]

List of ToolDefinition objects for all registered tools.
Source code in src/agenticapi/runtime/tools/registry.py 
113
114
115
116
117
118
119
120
121
def get_definitions(self) -> list[ToolDefinition]:
    """Return definitions for all registered tools.

    Alias for list_tools() for API consistency.

    Returns:
        List of ToolDefinition objects for all registered tools.
    """
    return self.list_tools()

@tool decorator

FastAPI-style decorator that turns a plain Python function into a registered tool with an auto-generated JSON Schema. See the Tool Decorator guide for usage patterns.

tool 

tool(func: F) -> Tool

tool(
    *,
    name: str | None = None,
    description: str | None = None,
    capabilities: list[ToolCapability] | None = None,
) -> Callable[[F], Tool]

tool(
    func: Callable[..., Any] | None = None,
    /,
    *,
    name: str | None = None,
    description: str | None = None,
    capabilities: list[ToolCapability] | None = None,
) -> Any

Decorate a function so it satisfies the :class:Tool protocol.

Parameters:

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

The function to decorate. Supplied automatically by the @tool form; supply explicitly when calling tool(my_func, ...).

 None
name str | None

Optional override for the tool name. Defaults to the function's __name__.

 None
description str | None

Optional override for the tool description. Defaults to the first non-empty line of the function's docstring, or a generic placeholder.

 None
capabilities list[ToolCapability] | None

Optional list of :class:ToolCapability values. When omitted, capabilities are inferred from the function name (e.g. functions starting with delete_ get WRITE; everything else defaults to READ).

 None

Returns:

Type Description
Any

An object satisfying the :class:Tool protocol that is also
Any

callable as the original function.
Source code in src/agenticapi/runtime/tools/decorator.py 
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
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def tool(
    func: Callable[..., Any] | None = None,
    /,
    *,
    name: str | None = None,
    description: str | None = None,
    capabilities: list[ToolCapability] | None = None,
) -> Any:
    """Decorate a function so it satisfies the :class:`Tool` protocol.

    Args:
        func: The function to decorate. Supplied automatically by the
            ``@tool`` form; supply explicitly when calling
            ``tool(my_func, ...)``.
        name: Optional override for the tool name. Defaults to the
            function's ``__name__``.
        description: Optional override for the tool description.
            Defaults to the first non-empty line of the function's
            docstring, or a generic placeholder.
        capabilities: Optional list of :class:`ToolCapability` values.
            When omitted, capabilities are inferred from the function
            name (e.g. functions starting with ``delete_`` get
            ``WRITE``; everything else defaults to ``READ``).

    Returns:
        An object satisfying the :class:`Tool` protocol that is also
        callable as the original function.
    """

    def _decorate(target: Callable[..., Any]) -> Tool:
        resolved_name = name or target.__name__
        resolved_description = _derive_description(target, description)
        resolved_capabilities = capabilities or _derive_capabilities(target)
        try:
            return_annotation = get_type_hints(target).get("return", None)
        except Exception:
            return_annotation = None
        decorated = _DecoratedTool(
            func=target,
            name=resolved_name,
            description=resolved_description,
            capabilities=resolved_capabilities,
            parameters_schema=_build_parameters_schema(target),
            return_annotation=return_annotation,
        )
        return cast("Tool", decorated)

    if func is not None:
        # Used as ``@tool`` (no parentheses).
        return _decorate(func)
    # Used as ``@tool(...)``.
    return _decorate

DatabaseTool

DatabaseTool

A tool that executes database queries via an async callable.

Wraps a user-provided async function for executing queries, enforcing read-only mode when configured.

Example

async def execute_query(query: str, params: dict | None = None): return await db.fetch_all(query, params or {})

tool = DatabaseTool(execute_fn=execute_query, read_only=True) result = await tool.invoke(query="SELECT COUNT(*) FROM orders")

Source code in src/agenticapi/runtime/tools/database.py 
 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
class DatabaseTool:
    """A tool that executes database queries via an async callable.

    Wraps a user-provided async function for executing queries,
    enforcing read-only mode when configured.

    Example:
        async def execute_query(query: str, params: dict | None = None):
            return await db.fetch_all(query, params or {})

        tool = DatabaseTool(execute_fn=execute_query, read_only=True)
        result = await tool.invoke(query="SELECT COUNT(*) FROM orders")
    """

    # SQL keywords that indicate write operations
    _WRITE_KEYWORDS: frozenset[str] = frozenset(
        {
            "INSERT",
            "UPDATE",
            "DELETE",
            "DROP",
            "CREATE",
            "ALTER",
            "TRUNCATE",
            "REPLACE",
            "MERGE",
            "UPSERT",
        }
    )

    def __init__(
        self,
        *,
        name: str = "database",
        description: str = "Execute SQL queries against a database",
        execute_fn: AsyncExecuteFn | None = None,
        read_only: bool = True,
    ) -> None:
        """Initialize the database tool.

        Args:
            name: The name for this tool instance.
            description: Human-readable description.
            execute_fn: Async callable that executes queries.
                        Signature: async (query: str, params: dict | None) -> Any
            read_only: If True, reject write operations.
        """
        self._name = name
        self._description = description
        self._execute_fn = execute_fn
        self._read_only = read_only

    @property
    def definition(self) -> ToolDefinition:
        """Return the tool's metadata definition."""
        capabilities = [ToolCapability.READ, ToolCapability.AGGREGATE, ToolCapability.SEARCH]
        if not self._read_only:
            capabilities.append(ToolCapability.WRITE)

        return ToolDefinition(
            name=self._name,
            description=self._description,
            capabilities=capabilities,
            parameters_schema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SQL query to execute"},
                    "params": {
                        "type": "object",
                        "description": "Query parameters for parameterized queries",
                        "default": None,
                    },
                },
                "required": ["query"],
            },
        )

    async def invoke(self, *, query: str, params: dict[str, Any] | None = None) -> Any:
        """Execute a database query.

        Args:
            query: The SQL query string to execute.
            params: Optional parameters for parameterized queries.

        Returns:
            The query result from the execute function.

        Raises:
            ToolError: If the tool has no execute function configured,
                       or if a write operation is attempted in read-only mode.
        """
        if self._execute_fn is None:
            raise ToolError(f"DatabaseTool '{self._name}' has no execute function configured")

        if self._read_only and self._is_write_query(query):
            raise ToolError(
                f"DatabaseTool '{self._name}' is read-only. Write operations are not permitted. Query: {query[:100]}"
            )

        logger.info(
            "database_tool_invoke",
            tool_name=self._name,
            query_preview=query[:100],
            read_only=self._read_only,
        )

        try:
            return await self._execute_fn(query, params)
        except ToolError:
            raise
        except Exception as exc:
            logger.error("database_tool_error", tool_name=self._name, error=str(exc))
            raise ToolError(f"Database query execution failed: {exc}") from exc

    @classmethod
    def _is_write_query(cls, query: str) -> bool:
        """Check if a query is a write operation.

        Strips SQL comments before checking to prevent bypasses like:
            "-- comment\\nDELETE FROM users"

        Args:
            query: The SQL query to check.

        Returns:
            True if the query appears to be a write operation.
        """
        # Remove line comments (-- ...) and block comments (/* ... */)
        cleaned = re.sub(r"--[^\n]*", "", query)
        cleaned = re.sub(r"/\*.*?\*/", "", cleaned, flags=re.DOTALL)
        stripped = cleaned.strip().upper()
        return any(stripped.startswith(keyword) for keyword in cls._WRITE_KEYWORDS)

definition property 

definition: ToolDefinition

Return the tool's metadata definition.

__init__ 

__init__(
    *,
    name: str = "database",
    description: str = "Execute SQL queries against a database",
    execute_fn: AsyncExecuteFn | None = None,
    read_only: bool = True,
) -> None

Initialize the database tool.

Parameters:

Name Type Description Default
name str

The name for this tool instance.

 'database'
description str

Human-readable description.

 'Execute SQL queries against a database'
execute_fn AsyncExecuteFn | None

Async callable that executes queries. Signature: async (query: str, params: dict | None) -> Any

 None
read_only bool

If True, reject write operations.

 True
Source code in src/agenticapi/runtime/tools/database.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    *,
    name: str = "database",
    description: str = "Execute SQL queries against a database",
    execute_fn: AsyncExecuteFn | None = None,
    read_only: bool = True,
) -> None:
    """Initialize the database tool.

    Args:
        name: The name for this tool instance.
        description: Human-readable description.
        execute_fn: Async callable that executes queries.
                    Signature: async (query: str, params: dict | None) -> Any
        read_only: If True, reject write operations.
    """
    self._name = name
    self._description = description
    self._execute_fn = execute_fn
    self._read_only = read_only

invoke async 

invoke(
    *, query: str, params: dict[str, Any] | None = None
) -> Any

Execute a database query.

Parameters:

Name Type Description Default
query str

The SQL query string to execute.

 required
params dict[str, Any] | None

Optional parameters for parameterized queries.

 None

Returns:

Type Description
Any

The query result from the execute function.

Raises:

Type Description
ToolError

If the tool has no execute function configured, or if a write operation is attempted in read-only mode.
Source code in src/agenticapi/runtime/tools/database.py 
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
async def invoke(self, *, query: str, params: dict[str, Any] | None = None) -> Any:
    """Execute a database query.

    Args:
        query: The SQL query string to execute.
        params: Optional parameters for parameterized queries.

    Returns:
        The query result from the execute function.

    Raises:
        ToolError: If the tool has no execute function configured,
                   or if a write operation is attempted in read-only mode.
    """
    if self._execute_fn is None:
        raise ToolError(f"DatabaseTool '{self._name}' has no execute function configured")

    if self._read_only and self._is_write_query(query):
        raise ToolError(
            f"DatabaseTool '{self._name}' is read-only. Write operations are not permitted. Query: {query[:100]}"
        )

    logger.info(
        "database_tool_invoke",
        tool_name=self._name,
        query_preview=query[:100],
        read_only=self._read_only,
    )

    try:
        return await self._execute_fn(query, params)
    except ToolError:
        raise
    except Exception as exc:
        logger.error("database_tool_error", tool_name=self._name, error=str(exc))
        raise ToolError(f"Database query execution failed: {exc}") from exc

CacheTool

CacheTool

An in-memory cache tool with TTL support.

Provides get, set, delete, and exists operations on a key-value store with per-entry expiration.

Example

tool = CacheTool(default_ttl_seconds=300) await tool.invoke(action="set", key="user:1", value={"name": "Alice"}) result = await tool.invoke(action="get", key="user:1")

Source code in src/agenticapi/runtime/tools/cache.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
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
class CacheTool:
    """An in-memory cache tool with TTL support.

    Provides get, set, delete, and exists operations on a
    key-value store with per-entry expiration.

    Example:
        tool = CacheTool(default_ttl_seconds=300)
        await tool.invoke(action="set", key="user:1", value={"name": "Alice"})
        result = await tool.invoke(action="get", key="user:1")
    """

    _ALLOWED_ACTIONS: frozenset[str] = frozenset({"get", "set", "delete", "exists"})

    def __init__(
        self,
        *,
        name: str = "cache",
        description: str = "In-memory key-value cache with TTL",
        max_size: int = 1000,
        default_ttl_seconds: float = 300.0,
    ) -> None:
        """Initialize the cache tool.

        Args:
            name: The name for this tool instance.
            description: Human-readable description.
            max_size: Maximum number of entries before eviction.
            default_ttl_seconds: Default TTL for entries in seconds.
        """
        self._name = name
        self._description = description
        self._max_size = max_size
        self._default_ttl = default_ttl_seconds
        self._store: dict[str, tuple[Any, float]] = {}  # key -> (value, expires_at)

    @property
    def definition(self) -> ToolDefinition:
        """Return the tool's metadata definition."""
        return ToolDefinition(
            name=self._name,
            description=self._description,
            capabilities=[ToolCapability.READ, ToolCapability.WRITE],
            parameters_schema={
                "type": "object",
                "properties": {
                    "action": {
                        "type": "string",
                        "enum": ["get", "set", "delete", "exists"],
                        "description": "Cache operation to perform",
                    },
                    "key": {"type": "string", "description": "Cache key"},
                    "value": {"description": "Value to cache (for set)", "default": None},
                    "ttl": {
                        "type": "number",
                        "description": "TTL in seconds (for set)",
                        "default": None,
                    },
                },
                "required": ["action", "key"],
            },
        )

    async def invoke(
        self,
        *,
        action: str,
        key: str,
        value: Any = None,
        ttl: float | None = None,
    ) -> Any:
        """Perform a cache operation.

        Args:
            action: One of "get", "set", "delete", "exists".
            key: The cache key.
            value: The value to store (required for "set").
            ttl: TTL in seconds (defaults to default_ttl_seconds).

        Returns:
            The cached value for "get", True/False for "exists",
            None for "set" and "delete".

        Raises:
            ToolError: If the action is invalid.
        """
        if action not in self._ALLOWED_ACTIONS:
            raise ToolError(f"Invalid cache action '{action}'. Allowed: {sorted(self._ALLOWED_ACTIONS)}")

        logger.info("cache_tool_invoke", tool_name=self._name, action=action, key=key)

        if action == "get":
            return self._get(key)
        elif action == "set":
            self._set(key, value, ttl)
            return None
        elif action == "delete":
            self._delete(key)
            return None
        elif action == "exists":
            return self._exists(key)
        return None

    def _get(self, key: str) -> Any:
        """Get a value from the cache."""
        entry = self._store.get(key)
        if entry is None:
            return None

        value, expires_at = entry
        if time.monotonic() > expires_at:
            del self._store[key]
            return None

        return value

    def _set(self, key: str, value: Any, ttl: float | None) -> None:
        """Set a value in the cache."""
        # Evict oldest if at capacity
        if len(self._store) >= self._max_size and key not in self._store:
            oldest_key = next(iter(self._store))
            del self._store[oldest_key]

        actual_ttl = ttl if ttl is not None else self._default_ttl
        expires_at = time.monotonic() + actual_ttl
        self._store[key] = (value, expires_at)

    def _delete(self, key: str) -> None:
        """Delete a key from the cache."""
        self._store.pop(key, None)

    def _exists(self, key: str) -> bool:
        """Check if a key exists and is not expired."""
        return self._get(key) is not None

definition property 

definition: ToolDefinition

Return the tool's metadata definition.

__init__ 

__init__(
    *,
    name: str = "cache",
    description: str = "In-memory key-value cache with TTL",
    max_size: int = 1000,
    default_ttl_seconds: float = 300.0,
) -> None

Initialize the cache tool.

Parameters:

Name Type Description Default
name str

The name for this tool instance.

 'cache'
description str

Human-readable description.

 'In-memory key-value cache with TTL'
max_size int

Maximum number of entries before eviction.

 1000
default_ttl_seconds float

Default TTL for entries in seconds.

 300.0
Source code in src/agenticapi/runtime/tools/cache.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    *,
    name: str = "cache",
    description: str = "In-memory key-value cache with TTL",
    max_size: int = 1000,
    default_ttl_seconds: float = 300.0,
) -> None:
    """Initialize the cache tool.

    Args:
        name: The name for this tool instance.
        description: Human-readable description.
        max_size: Maximum number of entries before eviction.
        default_ttl_seconds: Default TTL for entries in seconds.
    """
    self._name = name
    self._description = description
    self._max_size = max_size
    self._default_ttl = default_ttl_seconds
    self._store: dict[str, tuple[Any, float]] = {}  # key -> (value, expires_at)

invoke async 

invoke(
    *,
    action: str,
    key: str,
    value: Any = None,
    ttl: float | None = None,
) -> Any

Perform a cache operation.

Parameters:

Name Type Description Default
action str

One of "get", "set", "delete", "exists".

 required
key str

The cache key.

 required
value Any

The value to store (required for "set").

 None
ttl float | None

TTL in seconds (defaults to default_ttl_seconds).

 None

Returns:

Type Description
Any

The cached value for "get", True/False for "exists",
Any

None for "set" and "delete".

Raises:

Type Description
ToolError

If the action is invalid.
Source code in src/agenticapi/runtime/tools/cache.py 
 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
async def invoke(
    self,
    *,
    action: str,
    key: str,
    value: Any = None,
    ttl: float | None = None,
) -> Any:
    """Perform a cache operation.

    Args:
        action: One of "get", "set", "delete", "exists".
        key: The cache key.
        value: The value to store (required for "set").
        ttl: TTL in seconds (defaults to default_ttl_seconds).

    Returns:
        The cached value for "get", True/False for "exists",
        None for "set" and "delete".

    Raises:
        ToolError: If the action is invalid.
    """
    if action not in self._ALLOWED_ACTIONS:
        raise ToolError(f"Invalid cache action '{action}'. Allowed: {sorted(self._ALLOWED_ACTIONS)}")

    logger.info("cache_tool_invoke", tool_name=self._name, action=action, key=key)

    if action == "get":
        return self._get(key)
    elif action == "set":
        self._set(key, value, ttl)
        return None
    elif action == "delete":
        self._delete(key)
        return None
    elif action == "exists":
        return self._exists(key)
    return None

HttpClientTool

HttpClientTool

A tool for making HTTP requests.

Wraps httpx.AsyncClient with optional host allowlisting. Supports GET, POST, PUT, PATCH, DELETE methods.

Example

tool = HttpClientTool(allowed_hosts=["api.example.com"]) result = await tool.invoke(method="GET", url="https://api.example.com/data")

Source code in src/agenticapi/runtime/tools/http_client.py 
 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
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
class HttpClientTool:
    """A tool for making HTTP requests.

    Wraps httpx.AsyncClient with optional host allowlisting.
    Supports GET, POST, PUT, PATCH, DELETE methods.

    Example:
        tool = HttpClientTool(allowed_hosts=["api.example.com"])
        result = await tool.invoke(method="GET", url="https://api.example.com/data")
    """

    _ALLOWED_METHODS: frozenset[str] = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"})

    def __init__(
        self,
        *,
        name: str = "http_client",
        description: str = "Make HTTP requests to external APIs",
        allowed_hosts: list[str] | None = None,
        timeout: float = 30.0,
        default_headers: dict[str, str] | None = None,
    ) -> None:
        """Initialize the HTTP client tool.

        Args:
            name: The name for this tool instance.
            description: Human-readable description.
            allowed_hosts: If set, only requests to these hosts are permitted.
            timeout: Default request timeout in seconds.
            default_headers: Default headers to include in every request.
        """
        self._name = name
        self._description = description
        self._allowed_hosts = set(allowed_hosts) if allowed_hosts else None
        self._timeout = timeout
        self._default_headers = default_headers or {}

    @property
    def definition(self) -> ToolDefinition:
        """Return the tool's metadata definition."""
        return ToolDefinition(
            name=self._name,
            description=self._description,
            capabilities=[ToolCapability.READ, ToolCapability.EXECUTE],
            parameters_schema={
                "type": "object",
                "properties": {
                    "method": {
                        "type": "string",
                        "description": "HTTP method (GET, POST, PUT, PATCH, DELETE)",
                    },
                    "url": {"type": "string", "description": "The URL to request"},
                    "headers": {
                        "type": "object",
                        "description": "Additional HTTP headers",
                        "default": None,
                    },
                    "body": {
                        "description": "Request body (for POST/PUT/PATCH)",
                        "default": None,
                    },
                },
                "required": ["method", "url"],
            },
        )

    async def invoke(
        self,
        *,
        method: str,
        url: str,
        headers: dict[str, str] | None = None,
        body: Any = None,
    ) -> dict[str, Any]:
        """Make an HTTP request.

        Args:
            method: HTTP method (GET, POST, etc.).
            url: The URL to request.
            headers: Optional additional headers.
            body: Optional request body (JSON-serializable).

        Returns:
            Dict with status, headers, and body.

        Raises:
            ToolError: If the host is not allowed, method is invalid,
                       or the request fails.
        """
        method_upper = method.upper()
        if method_upper not in self._ALLOWED_METHODS:
            raise ToolError(f"HTTP method '{method}' is not allowed")

        # Validate host
        parsed = urlparse(url)
        if not parsed.hostname:
            raise ToolError(f"Invalid URL: {url}")

        if self._allowed_hosts is not None and parsed.hostname not in self._allowed_hosts:
            raise ToolError(
                f"Host '{parsed.hostname}' is not in the allowed hosts list. Allowed: {sorted(self._allowed_hosts)}"
            )

        logger.info(
            "http_client_tool_invoke",
            tool_name=self._name,
            method=method_upper,
            url=url[:200],
        )

        try:
            import httpx

            merged_headers = {**self._default_headers, **(headers or {})}
            async with httpx.AsyncClient(timeout=self._timeout) as client:
                response = await client.request(
                    method=method_upper,
                    url=url,
                    headers=merged_headers,
                    json=body if body is not None else None,
                )

            return {
                "status": response.status_code,
                "headers": dict(response.headers),
                "body": response.text,
            }
        except ToolError:
            raise
        except Exception as exc:
            logger.error("http_client_tool_error", tool_name=self._name, error=str(exc))
            raise ToolError(f"HTTP request failed: {exc}") from exc

definition property 

definition: ToolDefinition

Return the tool's metadata definition.

__init__ 

__init__(
    *,
    name: str = "http_client",
    description: str = "Make HTTP requests to external APIs",
    allowed_hosts: list[str] | None = None,
    timeout: float = 30.0,
    default_headers: dict[str, str] | None = None,
) -> None

Initialize the HTTP client tool.

Parameters:

Name Type Description Default
name str

The name for this tool instance.

 'http_client'
description str

Human-readable description.

 'Make HTTP requests to external APIs'
allowed_hosts list[str] | None

If set, only requests to these hosts are permitted.

 None
timeout float

Default request timeout in seconds.

 30.0
default_headers dict[str, str] | None

Default headers to include in every request.

 None
Source code in src/agenticapi/runtime/tools/http_client.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    *,
    name: str = "http_client",
    description: str = "Make HTTP requests to external APIs",
    allowed_hosts: list[str] | None = None,
    timeout: float = 30.0,
    default_headers: dict[str, str] | None = None,
) -> None:
    """Initialize the HTTP client tool.

    Args:
        name: The name for this tool instance.
        description: Human-readable description.
        allowed_hosts: If set, only requests to these hosts are permitted.
        timeout: Default request timeout in seconds.
        default_headers: Default headers to include in every request.
    """
    self._name = name
    self._description = description
    self._allowed_hosts = set(allowed_hosts) if allowed_hosts else None
    self._timeout = timeout
    self._default_headers = default_headers or {}

invoke async 

invoke(
    *,
    method: str,
    url: str,
    headers: dict[str, str] | None = None,
    body: Any = None,
) -> dict[str, Any]

Make an HTTP request.

Parameters:

Name Type Description Default
method str

HTTP method (GET, POST, etc.).

 required
url str

The URL to request.

 required
headers dict[str, str] | None

Optional additional headers.

 None
body Any

Optional request body (JSON-serializable).

 None

Returns:

Type Description
dict[str, Any]

Dict with status, headers, and body.

Raises:

Type Description
ToolError

If the host is not allowed, method is invalid, or the request fails.
Source code in src/agenticapi/runtime/tools/http_client.py 
 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
async def invoke(
    self,
    *,
    method: str,
    url: str,
    headers: dict[str, str] | None = None,
    body: Any = None,
) -> dict[str, Any]:
    """Make an HTTP request.

    Args:
        method: HTTP method (GET, POST, etc.).
        url: The URL to request.
        headers: Optional additional headers.
        body: Optional request body (JSON-serializable).

    Returns:
        Dict with status, headers, and body.

    Raises:
        ToolError: If the host is not allowed, method is invalid,
                   or the request fails.
    """
    method_upper = method.upper()
    if method_upper not in self._ALLOWED_METHODS:
        raise ToolError(f"HTTP method '{method}' is not allowed")

    # Validate host
    parsed = urlparse(url)
    if not parsed.hostname:
        raise ToolError(f"Invalid URL: {url}")

    if self._allowed_hosts is not None and parsed.hostname not in self._allowed_hosts:
        raise ToolError(
            f"Host '{parsed.hostname}' is not in the allowed hosts list. Allowed: {sorted(self._allowed_hosts)}"
        )

    logger.info(
        "http_client_tool_invoke",
        tool_name=self._name,
        method=method_upper,
        url=url[:200],
    )

    try:
        import httpx

        merged_headers = {**self._default_headers, **(headers or {})}
        async with httpx.AsyncClient(timeout=self._timeout) as client:
            response = await client.request(
                method=method_upper,
                url=url,
                headers=merged_headers,
                json=body if body is not None else None,
            )

        return {
            "status": response.status_code,
            "headers": dict(response.headers),
            "body": response.text,
        }
    except ToolError:
        raise
    except Exception as exc:
        logger.error("http_client_tool_error", tool_name=self._name, error=str(exc))
        raise ToolError(f"HTTP request failed: {exc}") from exc

QueueTool

QueueTool

An in-memory async queue tool for message passing.

Provides enqueue, dequeue, peek, and size operations on named queues.

Example

tool = QueueTool() await tool.invoke(action="enqueue", queue_name="tasks", message={"id": 1}) msg = await tool.invoke(action="dequeue", queue_name="tasks")

Source code in src/agenticapi/runtime/tools/queue.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
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
class QueueTool:
    """An in-memory async queue tool for message passing.

    Provides enqueue, dequeue, peek, and size operations
    on named queues.

    Example:
        tool = QueueTool()
        await tool.invoke(action="enqueue", queue_name="tasks", message={"id": 1})
        msg = await tool.invoke(action="dequeue", queue_name="tasks")
    """

    _ALLOWED_ACTIONS: frozenset[str] = frozenset({"enqueue", "dequeue", "peek", "size"})

    def __init__(
        self,
        *,
        name: str = "queue",
        description: str = "In-memory async message queue",
        max_size: int = 0,
    ) -> None:
        """Initialize the queue tool.

        Args:
            name: The name for this tool instance.
            description: Human-readable description.
            max_size: Maximum queue size (0 for unlimited).
        """
        self._name = name
        self._description = description
        self._max_size = max_size
        self._queues: dict[str, asyncio.Queue[Any]] = {}

    @property
    def definition(self) -> ToolDefinition:
        """Return the tool's metadata definition."""
        return ToolDefinition(
            name=self._name,
            description=self._description,
            capabilities=[ToolCapability.READ, ToolCapability.WRITE],
            parameters_schema={
                "type": "object",
                "properties": {
                    "action": {
                        "type": "string",
                        "enum": ["enqueue", "dequeue", "peek", "size"],
                        "description": "Queue operation to perform",
                    },
                    "queue_name": {"type": "string", "description": "Name of the queue"},
                    "message": {"description": "Message to enqueue", "default": None},
                    "timeout": {
                        "type": "number",
                        "description": "Timeout for dequeue in seconds",
                        "default": None,
                    },
                },
                "required": ["action", "queue_name"],
            },
        )

    def _get_queue(self, queue_name: str) -> asyncio.Queue[Any]:
        """Get or create a named queue."""
        if queue_name not in self._queues:
            self._queues[queue_name] = asyncio.Queue(maxsize=self._max_size)
        return self._queues[queue_name]

    async def invoke(
        self,
        *,
        action: str,
        queue_name: str,
        message: Any = None,
        timeout: float | None = None,
    ) -> Any:
        """Perform a queue operation.

        Args:
            action: One of "enqueue", "dequeue", "peek", "size".
            queue_name: Name of the queue to operate on.
            message: Message to enqueue (required for "enqueue").
            timeout: Timeout for dequeue in seconds.

        Returns:
            The message for "dequeue"/"peek", int for "size", None for "enqueue".

        Raises:
            ToolError: If the action is invalid or the operation fails.
        """
        if action not in self._ALLOWED_ACTIONS:
            raise ToolError(f"Invalid queue action '{action}'. Allowed: {sorted(self._ALLOWED_ACTIONS)}")

        logger.info("queue_tool_invoke", tool_name=self._name, action=action, queue_name=queue_name)

        queue = self._get_queue(queue_name)

        if action == "enqueue":
            try:
                queue.put_nowait(message)
            except asyncio.QueueFull as exc:
                raise ToolError(f"Queue '{queue_name}' is full (max_size={self._max_size})") from exc
            return None
        elif action == "dequeue":
            try:
                if timeout is not None:
                    return await asyncio.wait_for(queue.get(), timeout=timeout)
                return queue.get_nowait()
            except asyncio.QueueEmpty:
                return None
            except TimeoutError:
                return None
        elif action == "peek":
            if queue.empty():
                return None
            # Note: peek is not atomic. In concurrent scenarios, the peeked
            # item may be consumed by another coroutine between get and put.
            # For Phase 1, this is acceptable. Phase 2 should use a deque.
            try:
                item = queue.get_nowait()
            except asyncio.QueueEmpty:
                return None
            await queue.put(item)
            return item
        elif action == "size":
            return queue.qsize()
        return None

definition property 

definition: ToolDefinition

Return the tool's metadata definition.

__init__ 

__init__(
    *,
    name: str = "queue",
    description: str = "In-memory async message queue",
    max_size: int = 0,
) -> None

Initialize the queue tool.

Parameters:

Name Type Description Default
name str

The name for this tool instance.

 'queue'
description str

Human-readable description.

 'In-memory async message queue'
max_size int

Maximum queue size (0 for unlimited).

 0
Source code in src/agenticapi/runtime/tools/queue.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    *,
    name: str = "queue",
    description: str = "In-memory async message queue",
    max_size: int = 0,
) -> None:
    """Initialize the queue tool.

    Args:
        name: The name for this tool instance.
        description: Human-readable description.
        max_size: Maximum queue size (0 for unlimited).
    """
    self._name = name
    self._description = description
    self._max_size = max_size
    self._queues: dict[str, asyncio.Queue[Any]] = {}

invoke async 

invoke(
    *,
    action: str,
    queue_name: str,
    message: Any = None,
    timeout: float | None = None,
) -> Any

Perform a queue operation.

Parameters:

Name Type Description Default
action str

One of "enqueue", "dequeue", "peek", "size".

 required
queue_name str

Name of the queue to operate on.

 required
message Any

Message to enqueue (required for "enqueue").

 None
timeout float | None

Timeout for dequeue in seconds.

 None

Returns:

Type Description
Any

The message for "dequeue"/"peek", int for "size", None for "enqueue".

Raises:

Type Description
ToolError

If the action is invalid or the operation fails.
Source code in src/agenticapi/runtime/tools/queue.py 
 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
async def invoke(
    self,
    *,
    action: str,
    queue_name: str,
    message: Any = None,
    timeout: float | None = None,
) -> Any:
    """Perform a queue operation.

    Args:
        action: One of "enqueue", "dequeue", "peek", "size".
        queue_name: Name of the queue to operate on.
        message: Message to enqueue (required for "enqueue").
        timeout: Timeout for dequeue in seconds.

    Returns:
        The message for "dequeue"/"peek", int for "size", None for "enqueue".

    Raises:
        ToolError: If the action is invalid or the operation fails.
    """
    if action not in self._ALLOWED_ACTIONS:
        raise ToolError(f"Invalid queue action '{action}'. Allowed: {sorted(self._ALLOWED_ACTIONS)}")

    logger.info("queue_tool_invoke", tool_name=self._name, action=action, queue_name=queue_name)

    queue = self._get_queue(queue_name)

    if action == "enqueue":
        try:
            queue.put_nowait(message)
        except asyncio.QueueFull as exc:
            raise ToolError(f"Queue '{queue_name}' is full (max_size={self._max_size})") from exc
        return None
    elif action == "dequeue":
        try:
            if timeout is not None:
                return await asyncio.wait_for(queue.get(), timeout=timeout)
            return queue.get_nowait()
        except asyncio.QueueEmpty:
            return None
        except TimeoutError:
            return None
    elif action == "peek":
        if queue.empty():
            return None
        # Note: peek is not atomic. In concurrent scenarios, the peeked
        # item may be consumed by another coroutine between get and put.
        # For Phase 1, this is acceptable. Phase 2 should use a deque.
        try:
            item = queue.get_nowait()
        except asyncio.QueueEmpty:
            return None
        await queue.put(item)
        return item
    elif action == "size":
        return queue.qsize()
    return None