Let KugelAudio run the back office
List your tools and let a model KugelAudio runs decide when to use them, so your code only runs each tool.
On this page
Set delegation.type to responses. KugelAudio then runs the model that works on a delegation,
and your client runs its tools. Everything the model writes is spoken word for word, so it can
also ask the caller something and carry on with the answer.
Run a call with a tool
Save this next to talk.py from the quickstart:
import asyncio
import json
import os
from openai import AsyncOpenAI
from talk import RATE, Speaker, Transcript, stream_microphone
def find_order(number: str) -> dict[str, str]:
"""Your system: replace this with a real lookup."""
return {"number": number, "status": "shipped", "arrives": "tomorrow"}
def run_tool(name: str, arguments: str) -> str:
"""The service asks for a tool, and your code runs it and returns text."""
if name == "find_order":
return json.dumps(find_order(**json.loads(arguments)))
return json.dumps({"error": f"unknown tool {name}"})
async def main() -> None:
tasks: set[asyncio.Task[None]] = set()
speaker = Speaker()
transcript = Transcript()
async with (
AsyncOpenAI(
api_key=os.environ["KUGELAUDIO_API_KEY"],
base_url="https://api.kugelaudio.com/v1",
) as client,
client.live.connect() as connection,
):
await connection.session.start(
session={
"model": "fluid-1",
"instructions": "You answer the phone for a parcel service. Keep replies short.",
"audio": {"format": {"type": "audio/pcm", "rate": RATE}},
"delegation": {
"type": "responses",
"responses": {
"instructions": "You are the back office of a parcel service.",
"tools": [
{
"type": "function",
"name": "find_order",
"description": "Find an order by its number.",
"parameters": {
"type": "object",
"properties": {"number": {"type": "string"}},
"required": ["number"],
},
}
],
},
},
}
)
with speaker.stream:
async for event in connection:
if event.type == "session.started":
tasks.add(asyncio.create_task(stream_microphone(connection)))
print("Talk now. Press Ctrl+C to hang up.", end="")
elif event.type == "session.output_audio.delta":
speaker.play(event.delta)
elif event.type == "session.input_transcript.delta":
transcript.add("You", event.delta)
elif event.type == "session.output_transcript.delta":
transcript.add("Agent", event.delta)
elif event.type == "response.event" and event.event["type"] == "response.output_item.done":
item = event.event["item"]
await connection.send(
{
"type": "response.item.create",
"item": {
"type": "function_call_output",
"call_id": item["call_id"],
"output": run_tool(item["name"], item["arguments"]),
},
}
)
await connection.send({"type": "response.create"})
elif event.type == "error":
print("Error:", event.error.code, event.error.message)
elif event.type == "session.closed":
break
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nHung up.")Run it with python back_office.py and ask about order 4711. The agent's words print as it
speaks, and the tool's result is in what it says.
Return tool results
Return every tool call of a response, then send one response.create. KugelAudio waits 60
seconds for it. After that the response fails with tool_output_timeout, the caller hears that
it failed, and the call goes on. One delegation runs at most 16 model calls. The conversation
carries across the delegations of a call.
Choose the model
Set responses.model to gpt-6.1-sol, gpt-6-astra or gpt-6.1-terra, or leave it out and
KugelAudio chooses. These three run on OpenAI's API, so naming one sends the conversation and
your tool results to OpenAI. Any other model fails session.start with unsupported_model.
Supported settings
responses takes model, instructions, tools, tool_choice (auto, none or required),
parallel_tool_calls and reasoning.effort. Only function tools work: hosted tools and MCP from
OpenAI's delegation guide are
refused with invalid_event.