Assistants API streaming with tool calls - implementation guide
I spent a week getting streaming to work properly with the Assistants API, especially with tool calls. Here's a complete implementation:
from openai import OpenAIclient = OpenAI()
with client.beta.threads.runs.stream(
thread_id=thread.id,
assistant_id=assistant.id,
) as stream:
for event in stream:
if event.event == "thread.message.delta":
for block in event.data.delta.content:
if block.type == "text":
print(block.text.value, end="", flush=True)
elif event.event == "thread.run.requires_action":
tool_outputs = []
for call in event.data.required_action.submit_tool_outputs.tool_calls:
result = handle_tool_call(call)
tool_outputs.append({
"tool_call_id": call.id,
"output": json.dumps(result)
})
# Submit and continue streaming
stream.submit_tool_outputs(tool_outputs)
The key gotcha: you MUST submit tool outputs through the stream object, not through a separate API call.
Log in to reply to this topic.