Tools

Tool system for AI function calling. See the Tool Calling & Policies guide and MCP Tool Provider guide for usage examples.

ToolPolicy

Bases: BaseModel

Per-agent allow/deny rules for tool access.

Rules use :func:fnmatch.fnmatch glob patterns (e.g. "mcp_*", "search_*").

Resolution order:

1. Empty allow and empty deny → permit all (backward compatible).
2. If the tool name matches any deny pattern → blocked.
3. If allow is non-empty and the tool name matches no allow pattern → blocked.
4. Otherwise → permitted.

In short: deny always wins, and a non-empty allow list is a whitelist.

Role overrides ~~~~~~~~~~~~~~

role_overrides maps :class:~roomkit.models.enums.ParticipantRole string values (e.g. "observer", "member") to :class:RoleOverride instances.

Call :meth:resolve with a role to obtain an effective ToolPolicy that merges the base rules with the role-specific override.

resolve

resolve(role=None)

Return an effective ToolPolicy for the given role.

If role is None or has no override entry, returns self unchanged (backward compatible).

is_allowed

is_allowed(tool_name)

Return True if tool_name passes the policy.

as_filter

as_filter()

Return a callable (tool_name) -> bool suitable for :func:filter.

RoleOverride

Bases: BaseModel

Per-role tool policy override.

mode controls how the override combines with the base policy:

• "restrict" (default): deny lists are unioned, allow lists are intersected (a tool must pass both the base and override allow lists).
• "replace": the override completely replaces the base policy.

MCPToolProvider

MCPToolProvider(url, *, transport='streamable_http', tool_filter=None, headers=None)

Discover and invoke tools from an MCP server.

Supports both streamable_http (default) and sse transports.

Usage::

async with MCPToolProvider.from_url("http://localhost:8000/mcp") as mcp:
    tools = mcp.get_tools()          # list[AITool]
    handler = mcp.as_tool_handler()   # ToolHandler for AIChannel

tool_names property

tool_names

Return the names of all discovered tools.

from_url classmethod

from_url(url, *, transport='streamable_http', tool_filter=None, headers=None)

Create an MCPToolProvider for the given URL.

The provider is not connected until used as an async context manager.

Parameters:

Name	Type	Description	Default
url	str	MCP server URL.	required
transport	str	"streamable_http" (default) or "sse".	'streamable_http'
tool_filter	Callable[[str], bool] | None	Optional predicate to include only matching tool names.	None
headers	dict[str, str] | None	Optional HTTP headers sent with every request.	None

Returns:

Type	Description
MCPToolProvider	An MCPToolProvider instance (not yet connected).

get_tools

get_tools()

Return discovered tools as RoomKit AITool instances.

get_tools_as_dicts

get_tools_as_dicts()

Return discovered tools as plain dicts (for binding metadata).

call_tool async

call_tool(name, arguments, *, timeout=30.0)

Call a tool on the MCP server and return the result as a string.

Parameters:

Name	Type	Description	Default
name	str	Tool name.	required
arguments	dict[str, Any]	Tool arguments dict.	required
timeout	float	Maximum seconds to wait for a response.	30.0

Returns:

Type	Description
str	Result string. Single TextContent → plain text; multi-part → JSON array;
str	error results → {"error": "..."}.

as_tool_handler

as_tool_handler()

Return a ToolHandler suitable for AIChannel(tool_handler=...).

Unknown tools (not from this MCP server) return {"error": "Unknown tool: <name>"}, which allows composition via compose_tool_handlers.

compose_tool_handlers

compose_tool_handlers(*handlers)

Chain multiple ToolHandlers so the first one that handles a tool wins.

Each handler is tried in order. If a handler returns a JSON object with {"error": "Unknown tool: ..."} the next handler is tried. The last handler's result is always returned as-is (even if it's an unknown-tool error).

Parameters:

Name	Type	Description	Default
*handlers	ToolHandler	Two or more ToolHandler callables.	()

Returns:

Type	Description
ToolHandler	A single ToolHandler that dispatches to the first matching handler.

Raises:

Type	Description
ValueError	If fewer than two handlers are provided.

Human-in-the-Loop

HumanInputHandler

HumanInputHandler()

Manages pending human input requests.

Core lifecycle::

pending = await handler.create("AskUser", args, room_id="r1", ...)
# → ON_USER_INPUT_REQUIRED hook fires
result  = await handler.wait(pending.pending_id, timeout=300)
# → blocks until resolve() / reject() / timeout

The _on_input_required callback is injected by the framework (via register_channel hook builder) or set by the application directly.

pending property

pending

Active pending requests (read-only snapshot).

create async

create(tool_name, arguments, *, room_id='', tool_call_id='', channel_id='', channel_type=AI)

Register a new pending input request and fire the callback.

If the _on_input_required callback returns False (hook denied), the request is auto-rejected before wait() is called.

wait async

wait(pending_id, *, timeout=300)

Block until the request is resolved, rejected, or times out.

Returns:

Type	Description
str	The result string on resolution.

Raises:

Type	Description
TimeoutError	If the timeout expires.
RuntimeError	If the request was rejected.
ValueError	If pending_id is not found.

resolve

resolve(pending_id, result)

Resolve a pending request with a result.

Returns True if the request was found and resolved.

reject

reject(pending_id, reason='')

Reject a pending request.

Returns True if the request was found and rejected.

HumanInputToolHandler

HumanInputToolHandler(tool_names, timeout=300, handler=None, tool_definitions=None)

ToolHandler wrapper that blocks on human input for specified tools.

Composes with other handlers via :func:~roomkit.tools.compose.compose_tool_handlers. Falls through (returns "Unknown tool" error) for non-matching tool names so the compose chain continues to the next handler.

Pass this to :class:~roomkit.channels.ai.AIChannel via the human_input_handler parameter — the channel auto-composes it and the framework injects the ON_USER_INPUT_REQUIRED hook callback at registration time.

handler property

handler

The underlying :class:HumanInputHandler for resolve/reject access.

tools property

tools

Tool definitions to inject into the AI context.

__call__ async

__call__(name, arguments)

ToolHandler protocol — blocks on matching tools, falls through otherwise.

PendingInput dataclass

PendingInput(pending_id, tool_name, arguments, room_id, tool_call_id, channel_id, status=PENDING, result=None, reject_reason=None, created_at=_utcnow(), _event=Event())

A pending human input request.

Mutable — transitions from PENDING to RESOLVED/REJECTED/TIMED_OUT when the application calls :meth:HumanInputHandler.resolve or :meth:HumanInputHandler.reject.

PendingInputEvent dataclass

PendingInputEvent(pending_id, tool_name, arguments, room_id, tool_call_id, channel_id, channel_type, timestamp=_utcnow())

Event fired through ON_USER_INPUT_REQUIRED hooks.

Carries the pending request details so notification layers (WebSocket, REST, etc.) can inform the user.

pending_id instance-attribute

pending_id

Handler-generated ID for resolving this request.

tool_name instance-attribute

tool_name

Name of the tool that requires human input.

arguments instance-attribute

arguments

Tool arguments (e.g. questions, options).

room_id instance-attribute

room_id

Room where the tool call originated.

tool_call_id instance-attribute

tool_call_id

Provider-assigned tool call ID.

channel_id instance-attribute

channel_id

Channel that triggered the tool call.

channel_type instance-attribute

channel_type

Type of the originating channel.

timestamp class-attribute instance-attribute

timestamp = field(default_factory=_utcnow)

When the pending request was created.

PendingInputStatus

Bases: StrEnum

Status of a pending human input request.