Tools

tools:
  - id: hold_slot
    type: http
    method: POST
    url: "{{ env.API_BASE_URL }}/slots/hold"
    headers: {Authorization: "Bearer {{ env.ACCESS_TOKEN }}"}
    body: {city: "{{ slots.city }}", date: "{{ slots.date }}"}
    store_response_as: hold      # readable as results.hold.* afterwards
    env_updates: {hold_id: hold_id}   # env key <- dotted path into the response
    run_once: false              # true: at most one call per session
    when: "slots.city"           # expr over state; skip the call when falsy
    timeout: 10

pipelines:
  - id: confirm_and_hold
    steps:
      - tool: hold_slot
        on:
          ok: continue                      # next step, or pipeline success
          http_409: booking.offer_other     # typed status branch
          failed: {retry: 2, on_exhaust: booking.collect}

- id: confirm
  gate: hard
  pipeline: confirm_and_hold
  advance_when:
    - {when: "pipeline.ok", judge: expr, to: booking.close}
    - {when: "pipeline.failed", judge: expr, to: booking.collect}

middleware: {on_status: 401, refresh_with: refresh_auth, then: replay}

tools:
  - id: lookup_customer
    type: python
    args: {phone_number: {type: str, required: true}}
    store_response_as: customer

from superdialog.playbook import Playbook, PlaybookAgent, httpx_http

async def lookup_customer(args, state) -> dict:
    """Look up a customer by phone number."""
    return await crm.get_by_phone(args["phone_number"])

agent = PlaybookAgent(
    playbook=Playbook.load("booking.yaml"),
    talker_llm=talker,
    director_llm=director,
    http=httpx_http,
    python_tools={"lookup_customer": lookup_customer},   # bind by id
)

from superdialog import DialogMachine, PythonTool

agent = DialogMachine(
    "booking.yaml",
    llm="anthropic/claude-haiku-4-5",
    tools=[PythonTool.of(lookup_customer)],   # bridged through the engine
)

from superdialog import MCPTool

tool = MCPTool(
    id="search",
    name="search",
    description="Search the internal knowledge base",
    server="https://mcp.company.io",
)

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)

dm = DialogMachine(flow, llm="...", engine="flow", tools=[lookup_customer])

from superdialog import PythonTool, HttpTool
import os

tool = PythonTool.of(lookup_customer)        # infer id/name/schema

http = HttpTool(
    id="lookup", name="lookup",
    description="Look up a customer by partial Aadhaar",
    url="https://api.company.io/customer/lookup",
    auth={"type": "bearer", "token": os.environ["COMPANY_KEY"]},
)

from superdialog.flow.models import ConversationFlow, FlowNode, Edge
from superdialog.machine.machine import DialogStateMachine
from superdialog.tools import tool

@tool
async def search_kb(query: str) -> dict:
    """Search the knowledge base."""
    return {"results": await kb.search(query)}

async def check_availability(date: str) -> dict:
    """Check available appointment slots for a date."""
    return {"slots": await calendar.get_slots(date)}

flow = ConversationFlow(
    system_prompt="You are an appointment booking assistant.",
    initial_node="collect_info",
    tools=[search_kb],                        # available on every node
    nodes=[
        FlowNode(
            id="collect_info",
            name="Collect Info",
            instruction="Collect patient name and preferred date.",
            tools=[check_availability],       # node-scoped
            edges=[Edge(id="e_confirm", condition="All info collected", target_node_id="confirm")],
        ),
    ],
)

machine = await DialogStateMachine.from_flow(flow, adapter)

from superdialog.machine.models import ToolResult

async def book_appointment(slot_id: str) -> ToolResult:
    """Book the appointment for the given slot."""
    if await calendar.book(slot_id):
        return ToolResult(data={"booked": True}, transition_edge_id="booking_confirmed")
    return ToolResult(data={"booked": False})