Skip to content

Tools & Registry Reference

This section documents the tool abstraction layer, execution context, tool result payloads, and the tool registry.

BaseTool & Context

edgevoiceai.tools.base

Base abstractions for pluggable tools in the Voice AI Framework.

BaseTool

Bases: ABC

Abstract Base Class for any Voice Agent Tool. Implementations specify schema, trigger description, and async execution logic.

Source code in edgevoiceai/tools/base.py 
 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
class BaseTool(ABC):
    """
    Abstract Base Class for any Voice Agent Tool.
    Implementations specify schema, trigger description, and async execution logic.
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """Unique function declaration name sent to Gemini."""
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        """Tool description in language(s) optimized for LLM trigger."""
        pass

    @property
    @abstractmethod
    def parameters_schema(self) -> Dict[str, Any]:
        """JSON schema parameters definition."""
        pass

    @property
    def prompt_instruction(self) -> Optional[str]:
        """
        Optional system prompt guideline/rule for this tool.
        If provided, the framework automatically composes this rule into the AI's system prompt.
        """
        return None

    @abstractmethod
    async def execute(self, args: Dict[str, Any], context: ToolExecutionContext) -> ToolResult:
        """
        Asynchronously execute tool with provided arguments and session context.
        """
        pass


    def to_function_declaration(self) -> Dict[str, Any]:
        """Converts this tool into a Gemini API function declaration object."""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters_schema,
        }

description abstractmethod property 

description

Tool description in language(s) optimized for LLM trigger.

name abstractmethod property 

name

Unique function declaration name sent to Gemini.

parameters_schema abstractmethod property 

parameters_schema

JSON schema parameters definition.

prompt_instruction property 

prompt_instruction

Optional system prompt guideline/rule for this tool. If provided, the framework automatically composes this rule into the AI's system prompt.

execute abstractmethod async 

execute(args, context)

Asynchronously execute tool with provided arguments and session context.

Source code in edgevoiceai/tools/base.py 
 96
 97
 98
 99
100
101
@abstractmethod
async def execute(self, args: Dict[str, Any], context: ToolExecutionContext) -> ToolResult:
    """
    Asynchronously execute tool with provided arguments and session context.
    """
    pass

to_function_declaration 

to_function_declaration()

Converts this tool into a Gemini API function declaration object.

Source code in edgevoiceai/tools/base.py 
104
105
106
107
108
109
110
def to_function_declaration(self) -> Dict[str, Any]:
    """Converts this tool into a Gemini API function declaration object."""
    return {
        "name": self.name,
        "description": self.description,
        "parameters": self.parameters_schema,
    }

ToolExecutionContext dataclass

Context passed into every tool execution call. Contains session details, turn counter, and optional client message emitter.

Source code in edgevoiceai/tools/base.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class ToolExecutionContext:
    """
    Context passed into every tool execution call.
    Contains session details, turn counter, and optional client message emitter.
    """
    session_id: str
    current_turn: int = 0
    client_emitter: Optional[Callable[[Dict[str, Any]], Coroutine[Any, Any, None]]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    async def emit_to_client(self, event_data: Dict[str, Any]) -> None:
        """Sends an event back to the client interface (e.g., browser or device UI)."""
        if self.client_emitter:
            try:
                await self.client_emitter(event_data)
            except Exception as exc:
                print(f"⚠️ [ToolExecutionContext] Failed to emit client event: {exc}")

emit_to_client async 

emit_to_client(event_data)

Sends an event back to the client interface (e.g., browser or device UI).

Source code in edgevoiceai/tools/base.py 
55
56
57
58
59
60
61
async def emit_to_client(self, event_data: Dict[str, Any]) -> None:
    """Sends an event back to the client interface (e.g., browser or device UI)."""
    if self.client_emitter:
        try:
            await self.client_emitter(event_data)
        except Exception as exc:
            print(f"⚠️ [ToolExecutionContext] Failed to emit client event: {exc}")

ToolResult dataclass

Unified return type for any tool executed within the Voice AI engine.

Source code in edgevoiceai/tools/base.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass
class ToolResult:
    """
    Unified return type for any tool executed within the Voice AI engine.
    """
    status: str = "success"
    message: Optional[str] = None
    data: Dict[str, Any] = field(default_factory=dict)

    # Just-In-Time instructions to guide the voice model's immediate response
    jit_instruction: Optional[str] = None

    # Event data to broadcast to connected client(s) (e.g. browser WebSocket)
    client_event: Optional[Dict[str, Any]] = None

    # Audio file URL to queue for physical playback (e.g. on Raspberry Pi speaker)
    queued_audio_url: Optional[str] = None

    def to_gemini_response_content(self) -> Dict[str, Any]:
        """Formats the result into the dictionary expected by Gemini's toolResponse."""
        content: Dict[str, Any] = {
            "status": self.status,
        }
        if self.message:
            content["message"] = self.message
        if self.jit_instruction:
            content["instruction"] = self.jit_instruction
        if self.data:
            content.update(self.data)
        return content

to_gemini_response_content 

to_gemini_response_content()

Formats the result into the dictionary expected by Gemini's toolResponse.

Source code in edgevoiceai/tools/base.py 
30
31
32
33
34
35
36
37
38
39
40
41
def to_gemini_response_content(self) -> Dict[str, Any]:
    """Formats the result into the dictionary expected by Gemini's toolResponse."""
    content: Dict[str, Any] = {
        "status": self.status,
    }
    if self.message:
        content["message"] = self.message
    if self.jit_instruction:
        content["instruction"] = self.jit_instruction
    if self.data:
        content.update(self.data)
    return content

ToolRegistry

edgevoiceai.tools.registry

Central ToolRegistry for discovering, registering, and dynamically dispatching tool calls.

ToolRegistry

Manages a collection of BaseTool instances and handles dispatching tool calls.

Source code in edgevoiceai/tools/registry.py 
 14
 15
 16
 17
 18
 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 ToolRegistry:
    """
    Manages a collection of BaseTool instances and handles dispatching tool calls.
    """

    def __init__(self, tools: Optional[List[BaseTool]] = None):
        self._tools: Dict[str, BaseTool] = {}
        if tools:
            for tool in tools:
                self.register(tool)

    def register(self, tool: BaseTool) -> None:
        """Registers a tool instance. Overwrites existing registration if name matches."""
        self._tools[tool.name] = tool
        _logger.debug("Registered tool '%s'", tool.name)

    def unregister(self, tool_name: str) -> Optional[BaseTool]:
        """Removes a tool from registry by name."""
        return self._tools.pop(tool_name, None)

    def get(self, tool_name: str) -> Optional[BaseTool]:
        """Returns the registered tool instance, if found."""
        return self._tools.get(tool_name)

    def has_tool(self, tool_name: str) -> bool:
        """Checks if a tool is registered."""
        return tool_name in self._tools

    def list_tools(self) -> List[BaseTool]:
        """Returns a list of all registered tools."""
        return list(self._tools.values())

    def to_gemini_tools_payload(self) -> List[Dict[str, Any]]:
        """
        Builds the Gemini Live `tools` array configuration:
        [{"functionDeclarations": [...]}]
        """
        if not self._tools:
            return []
        return [
            {
                "functionDeclarations": [
                    tool.to_function_declaration() for tool in self._tools.values()
                ]
            }
        ]

    def build_tools_instruction(self, header: str = "[ACTIVE TOOLS & USAGE RULES]") -> str:
        """
        Aggregates prompt instructions from all registered tools into a unified guideline block.
        Ensures the AI assistant strictly understands when and how to call active tools.
        """
        rules: List[str] = []
        for tool in self._tools.values():
            instruction = tool.prompt_instruction
            if instruction and instruction.strip():
                rules.append(instruction.strip())

        if not rules:
            return ""

        return f"{header}\n" + "\n\n".join(rules)


    async def dispatch(
        self,
        tool_name: str,
        args: Dict[str, Any],
        context: ToolExecutionContext,
    ) -> ToolResult:
        """
        Dynamically dispatches a tool call to the registered handler.
        Emits client events if the tool produces any.
        """
        tool = self.get(tool_name)
        if not tool:
            _logger.warning("Attempted to call unregistered tool: %s", tool_name)
            return ToolResult(
                status="error",
                message=f"Tool '{tool_name}' is not registered or supported by this agent.",
            )

        try:
            result = await tool.execute(args, context)

            # Emit client event if produced by tool execution
            if result.client_event and context.client_emitter:
                await context.emit_to_client(result.client_event)

            return result
        except Exception as exc:
            _logger.exception("Error executing tool '%s': %s", tool_name, exc)
            return ToolResult(
                status="error",
                message=f"Error executing tool '{tool_name}': {exc}",
            )

build_tools_instruction 

build_tools_instruction(header='[ACTIVE TOOLS & USAGE RULES]')

Aggregates prompt instructions from all registered tools into a unified guideline block. Ensures the AI assistant strictly understands when and how to call active tools.

Source code in edgevoiceai/tools/registry.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def build_tools_instruction(self, header: str = "[ACTIVE TOOLS & USAGE RULES]") -> str:
    """
    Aggregates prompt instructions from all registered tools into a unified guideline block.
    Ensures the AI assistant strictly understands when and how to call active tools.
    """
    rules: List[str] = []
    for tool in self._tools.values():
        instruction = tool.prompt_instruction
        if instruction and instruction.strip():
            rules.append(instruction.strip())

    if not rules:
        return ""

    return f"{header}\n" + "\n\n".join(rules)

dispatch async 

dispatch(tool_name, args, context)

Dynamically dispatches a tool call to the registered handler. Emits client events if the tool produces any.

Source code in edgevoiceai/tools/registry.py 
 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
async def dispatch(
    self,
    tool_name: str,
    args: Dict[str, Any],
    context: ToolExecutionContext,
) -> ToolResult:
    """
    Dynamically dispatches a tool call to the registered handler.
    Emits client events if the tool produces any.
    """
    tool = self.get(tool_name)
    if not tool:
        _logger.warning("Attempted to call unregistered tool: %s", tool_name)
        return ToolResult(
            status="error",
            message=f"Tool '{tool_name}' is not registered or supported by this agent.",
        )

    try:
        result = await tool.execute(args, context)

        # Emit client event if produced by tool execution
        if result.client_event and context.client_emitter:
            await context.emit_to_client(result.client_event)

        return result
    except Exception as exc:
        _logger.exception("Error executing tool '%s': %s", tool_name, exc)
        return ToolResult(
            status="error",
            message=f"Error executing tool '{tool_name}': {exc}",
        )

get 

get(tool_name)

Returns the registered tool instance, if found.

Source code in edgevoiceai/tools/registry.py 
34
35
36
def get(self, tool_name: str) -> Optional[BaseTool]:
    """Returns the registered tool instance, if found."""
    return self._tools.get(tool_name)

has_tool 

has_tool(tool_name)

Checks if a tool is registered.

Source code in edgevoiceai/tools/registry.py 
38
39
40
def has_tool(self, tool_name: str) -> bool:
    """Checks if a tool is registered."""
    return tool_name in self._tools

list_tools 

list_tools()

Returns a list of all registered tools.

Source code in edgevoiceai/tools/registry.py 
42
43
44
def list_tools(self) -> List[BaseTool]:
    """Returns a list of all registered tools."""
    return list(self._tools.values())

register 

register(tool)

Registers a tool instance. Overwrites existing registration if name matches.

Source code in edgevoiceai/tools/registry.py 
25
26
27
28
def register(self, tool: BaseTool) -> None:
    """Registers a tool instance. Overwrites existing registration if name matches."""
    self._tools[tool.name] = tool
    _logger.debug("Registered tool '%s'", tool.name)

to_gemini_tools_payload 

to_gemini_tools_payload()

Builds the Gemini Live tools array configuration: [{"functionDeclarations": [...]}]

Source code in edgevoiceai/tools/registry.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def to_gemini_tools_payload(self) -> List[Dict[str, Any]]:
    """
    Builds the Gemini Live `tools` array configuration:
    [{"functionDeclarations": [...]}]
    """
    if not self._tools:
        return []
    return [
        {
            "functionDeclarations": [
                tool.to_function_declaration() for tool in self._tools.values()
            ]
        }
    ]

unregister 

unregister(tool_name)

Removes a tool from registry by name.

Source code in edgevoiceai/tools/registry.py 
30
31
32
def unregister(self, tool_name: str) -> Optional[BaseTool]:
    """Removes a tool from registry by name."""
    return self._tools.pop(tool_name, None)