Overview
Tools let your DialogMachine take actions during a conversation — look up a customer record, book a slot, query a database, call an API. When the LLM decides a tool is needed, SuperDialog executes it automatically and feeds the result back into the conversation.
Three tool types, one interface:
| Type | Execution | Best for |
|---|
PythonTool | In-process | Local functions, direct DB calls, any Python code |
HttpTool | HTTP request | External REST APIs, microservices |
MCPTool | MCP protocol | Model Context Protocol servers |
There are three ways to register a Python function as a tool.
The fastest way. Schema is inferred from type hints and description from the docstring.
from superdialog.tools import tool
@tool
async def lookup_customer(customer_id: str) -> dict:
"""Look up customer record by ID."""
return await crm.get(customer_id)
dialog_machine = DialogMachine(flow=flow, llm="...", tools=[lookup_customer])
@tool(name="check_stock", description="Check inventory availability by SKU")
async def check_inventory(sku: str, qty: int) -> dict:
return await inventory_api.check(sku, qty)
async def lookup_customer(customer_id: str) -> dict:
"""Look up customer record by ID."""
return await crm.get(customer_id)
async def check_inventory(sku: str) -> dict:
"""Check inventory levels."""
return await inventory_api.check(sku)
dialog_machine = DialogMachine(
flow=flow,
llm="...",
tools=[lookup_customer, check_inventory], # plain functions, no wrapping
)
3. Function references on the flow model
Attach functions directly to ConversationFlow.tools or FlowNode.tools when building flows in Python. This gives you scoping — flow-level tools are available on every node, node-level tools only on that node.
from superdialog.flow.models import ConversationFlow, FlowNode, Edge
async def search_kb(query: str) -> dict:
"""Search the knowledge base."""
return await kb.search(query)
async def check_availability(date: str) -> dict:
"""Check appointment availability."""
return await calendar.slots(date)
flow = ConversationFlow(
system_prompt="You are an appointment assistant.",
initial_node="greeting",
tools=[search_kb], # flow-level: available on every node
nodes=[
FlowNode(
id="collect",
name="Collect Info",
instruction="Collect patient name and preferred date.",
tools=[check_availability], # node-level: only on this node
edges=[
Edge(id="e_confirm", condition="All info collected", target_node_id="confirm")
],
),
...
],
)
# No tool_handlers dict required — functions are bound directly
machine = await DialogStateMachine.from_flow(flow, adapter)
@tool-decorated functions work here too:
@tool
async def lookup_customer(customer_id: str) -> dict:
"""Look up a customer."""
...
flow = ConversationFlow(
tools=[lookup_customer],
...
)
For explicit control over the tool’s id, name, description, or schema:
from superdialog import PythonTool
# Convenience constructor — infers id, name, description, schema from the function
tool = PythonTool.of(lookup_customer)
# Explicit constructor
tool = PythonTool(
id="crm-lookup",
name="lookup_customer",
description="Fetch full customer record from CRM by customer ID",
fn=lookup_customer,
)
For external REST APIs. SuperDialog POSTs to the URL with the tool’s input as a JSON body.
from superdialog import HttpTool
import os
tool = HttpTool(
id="lookup",
name="lookup",
description="Look up a customer by partial Aadhaar number",
url="https://api.company.io/customer/lookup",
method="POST", # default
auth={"type": "bearer", "token": os.environ["COMPANY_KEY"]},
)
# Bearer token
auth={"type": "bearer", "token": "your-token"}
For servers that implement the Model Context Protocol. Requires pip install superdialog[mcp].
from superdialog import MCPTool
tool = MCPTool(
id="search",
name="search",
description="Search the internal knowledge base",
server="https://mcp.company.io",
)
MCPTool connects lazily on first use via SSE. Auto-discovery of all tools published by an MCP server is planned for a follow-up release.
Deserialize a tool from a plain dict — useful when loading tool specs from JSON config files:
from superdialog import Tool
# HTTP tool from config
tool = Tool.from_dict({
"type": "http",
"id": "lookup",
"name": "lookup",
"description": "Look up customer",
"url": "https://api.company.io/lookup",
})
# Python tool from config (pass handler_registry to resolve function by name)
tool = Tool.from_dict(
{"type": "python", "id": "verify", "name": "verify", "description": "...", "handler_id": "verify_fn"},
handler_registry={"verify_fn": verify_customer},
)
Tool results are automatically merged into node slot values and made available in the conversation context. A tool can also trigger a flow transition by returning a ToolResult with a transition_edge_id:
from superdialog.machine.models import ToolResult
async def book_appointment(slot_id: str) -> ToolResult:
"""Book the appointment for the given slot."""
success = await calendar.book(slot_id)
if success:
return ToolResult(
data={"booked": True, "slot": slot_id},
transition_edge_id="booking_confirmed", # triggers edge transition
)
return ToolResult(data={"booked": False})
import asyncio
from superdialog.flow.models import ConversationFlow, FlowNode, Edge
from superdialog.machine.machine import DialogStateMachine
from superdialog.tools import tool
# Flow-level tool — available on every node
@tool
async def search_kb(query: str) -> dict:
"""Search the knowledge base for answers."""
return {"results": await kb.search(query)}
# Node-level tool — only available on the collect_info node
async def check_availability(date: str) -> dict:
"""Check available appointment slots for a date."""
return {"slots": await calendar.get_slots(date)}
async def verify_insurance(policy_number: str) -> dict:
"""Verify patient insurance coverage."""
return {"verified": await insurance_api.verify(policy_number)}
flow = ConversationFlow(
system_prompt="You are an appointment booking assistant.",
initial_node="greeting",
tools=[search_kb], # available everywhere
nodes=[
FlowNode(
id="greeting",
name="Greeting",
static_text="Hello! I can help you book an appointment.",
edges=[Edge(id="e_start", condition="User ready", target_node_id="collect_info")],
),
FlowNode(
id="collect_info",
name="Collect Info",
instruction="Collect patient name, date, and insurance policy number.",
tools=[check_availability, verify_insurance], # node-scoped
edges=[Edge(id="e_confirm", condition="All info collected", target_node_id="confirm")],
),
FlowNode(
id="confirm",
name="Confirm",
static_text="Your appointment is confirmed!",
is_final=True,
),
],
)
machine = await DialogStateMachine.from_flow(flow, adapter)
