Handling Tool Calls in Streamed Responses
Parse streaming responses that contain function call arguments arriving token by token, buffer the JSON fragments, and trigger tool execution only when the call is complete.
Handling Tool Calls in Streamed Responses is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
Tool Calls Arrive Differently in Streams
When an LLM decides to call a function, the response structure changes. Instead of a content string, the delta contains a tool_calls array. But in a streamed response, the function call arguments arrive token by token as a partial JSON string — you do not receive a complete JSON object in a single chunk. You must buffer these fragments and reassemble the complete JSON before you can parse and execute the tool call.
# In a non-streaming response, tool call is complete:
# choice.message.tool_calls[0].function.arguments = '{"city": "Paris"}'
# In a streaming response, arguments arrive in pieces:
# chunk 1: delta.tool_calls[0].function.arguments = '{'
# chunk 2: delta.tool_calls[0].function.arguments = '"city"'
# chunk 3: delta.tool_calls[0].function.arguments = ': "'
# chunk 4: delta.tool_calls[0].function.arguments = 'Paris'
# chunk 5: delta.tool_calls[0].function.arguments = '"}'
# You must concatenate these before JSON.parse can workDetecting a Tool Call in the Stream
Check each chunk's finish_reason to know when to expect tool calls. When finish_reason is 'tool_calls', the model has decided to call a function and the stream is ending. When finish_reason is 'stop', the model produced a normal text response. While streaming, check whether chunk.choices[0].delta.tool_calls is non-None to identify tool call argument fragments.
async def detect_stream_type(messages, tools):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=tools,
stream=True,
)
response_type = 'text'
async for chunk in stream:
choice = chunk.choices[0]
if choice.delta.tool_calls: # tool call fragment arriving
response_type = 'tool_call'
if choice.finish_reason == 'tool_calls':
print('Model wants to call a function')
elif choice.finish_reason == 'stop':
print('Normal text response')
return response_typeBuffering Tool Call Arguments
Use a dictionary keyed by tool call index to accumulate argument fragments from each chunk. The index identifies which tool call is being streamed — a model may call multiple functions in one response. For each chunk with a non-null tool_calls delta, append the argument fragment to the appropriate buffer entry keyed by the tool call index.
from collections import defaultdict
async def collect_streamed_tool_calls(messages, tools):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=tools,
stream=True,
)
tool_call_buffers = defaultdict(lambda: {'name': '', 'id': '', 'arguments': ''})
text_buffer = ''
async for chunk in stream:
delta = chunk.choices[0].delta
if delta.content: # text content
text_buffer += delta.content
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if tc.id:
tool_call_buffers[idx]['id'] = tc.id
if tc.function.name:
tool_call_buffers[idx]['name'] += tc.function.name
if tc.function.arguments:
tool_call_buffers[idx]['arguments'] += tc.function.arguments
return text_buffer, dict(tool_call_buffers)Parsing and Executing Tool Calls
After the stream ends and you have complete argument strings, parse each one with json.loads and dispatch to the appropriate Python function. Execute tool calls in the order they were requested (or in parallel if they are independent), then format the results as tool response messages to send back in the follow-up API call.
import json
# Example tool registry
tools_registry = {
'get_weather': lambda city, unit='celsius': {'temp': 22, 'desc': 'sunny', 'city': city},
'search_docs': lambda query, top_k=3: [{'title': 'Doc 1', 'snippet': 'Relevant info...'}],
}
def execute_tool_calls(tool_call_buffers: dict) -> list[dict]:
tool_messages = []
for idx in sorted(tool_call_buffers.keys()):
tc = tool_call_buffers[idx]
func_name = tc['name']
args = json.loads(tc['arguments'])
if func_name in tools_registry:
result = tools_registry[func_name](**args)
else:
result = {'error': f'Unknown function: {func_name}'}
tool_messages.append({
'role': 'tool',
'tool_call_id': tc['id'],
'content': json.dumps(result),
})
return tool_messagesThe Full Streaming Tool Call Loop
The complete streaming tool call flow requires a multi-turn conversation loop. The first request may return tool calls; you execute them and append the results to the message history; a second request returns the final text answer. This loop may iterate multiple times if the model chooses to make additional tool calls based on the results of earlier ones.
async def streaming_agent_loop(initial_messages, tools):
messages = list(initial_messages)
max_iterations = 5
for iteration in range(max_iterations):
text, tool_calls = await collect_streamed_tool_calls(messages, tools)
if tool_calls:
# Append assistant message with tool calls
assistant_msg = {
'role': 'assistant',
'content': text or None,
'tool_calls': [
{'id': tc['id'], 'type': 'function',
'function': {'name': tc['name'], 'arguments': tc['arguments']}}
for tc in tool_calls.values()
]
}
messages.append(assistant_msg)
# Execute tools and append results
tool_results = execute_tool_calls(tool_calls)
messages.extend(tool_results)
else:
# No more tool calls — final text response
print('Final answer:', text)
return text
return 'Max iterations reached'Streaming Text While Buffering Tool Calls
In practice, you want to stream text to the client immediately while simultaneously buffering any tool call arguments. This requires distinguishing between chunks that carry content (stream immediately) and chunks that carry tool_calls (buffer for later execution). Only after the stream ends and tool calls are complete can you execute tools and continue.
async def stream_with_tools(messages, tools):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini', messages=messages, tools=tools, stream=True
)
tool_buffers = defaultdict(lambda: {'name': '', 'id': '', 'arguments': ''})
text_parts = []
async for chunk in stream:
delta = chunk.choices[0].delta
finish = chunk.choices[0].finish_reason
if delta.content:
text_parts.append(delta.content)
yield ('text', delta.content) # stream to client immediately
if delta.tool_calls:
for tc in delta.tool_calls:
if tc.id: tool_buffers[tc.index]['id'] = tc.id
if tc.function.name: tool_buffers[tc.index]['name'] += tc.function.name
if tc.function.arguments: tool_buffers[tc.index]['arguments'] += tc.function.arguments
if finish == 'tool_calls':
yield ('tool_calls', dict(tool_buffers)) # signal tool execution neededParallel Tool Call Execution
When the model returns multiple tool calls simultaneously (a feature called parallel function calling), execute them in parallel with asyncio.gather rather than sequentially. Sequential execution adds unnecessary latency — if the model calls a weather API and a database lookup simultaneously, there is no reason to wait for one to finish before starting the other.
import asyncio
async def execute_tool_calls_parallel(tool_call_buffers: dict) -> list[dict]:
async def execute_one(idx, tc):
func_name = tc['name']
args = json.loads(tc['arguments'])
if func_name in async_tools_registry:
result = await async_tools_registry[func_name](**args)
else:
result = {'error': f'Unknown function: {func_name}'}
return {
'role': 'tool',
'tool_call_id': tc['id'],
'content': json.dumps(result),
}
tasks = [execute_one(idx, tc) for idx, tc in sorted(tool_call_buffers.items())]
return await asyncio.gather(*tasks)Streaming the Final Answer After Tool Use
After executing tool calls and appending results to the message history, make a second streaming request to get the model's final answer. Stream this response directly to the client. This two-request pattern (initial request with tool calls + follow-up request with tool results) is the standard agent turn cycle, and both requests can stream their text to the UI.
async def full_tool_calling_stream(question: str, tools: list):
messages = [{'role': 'user', 'content': question}]
# First request: may produce tool calls
tool_buffers = {}
text1 = ''
async for event_type, data in stream_with_tools(messages, tools):
if event_type == 'text':
text1 += data
yield data # stream partial text if any
elif event_type == 'tool_calls':
tool_buffers = data
if tool_buffers:
# Execute tools, then get final streaming answer
tool_results = await execute_tool_calls_parallel(tool_buffers)
messages += [{ # assistant tool call message
'role': 'assistant',
'tool_calls': [
{'id': tc['id'], 'type': 'function',
'function': {'name': tc['name'], 'arguments': tc['arguments']}}
for tc in tool_buffers.values()
]
}] + tool_results
# Second request: final answer streams directly
async for token in token_stream(messages): # from earlier lesson
yield tokenDisplaying Tool Call Progress in the UI
Users should see what the agent is doing while waiting for tool calls to complete. Before executing tools, stream a status event to the client indicating which function is being called and with what arguments. After execution, stream a completion status. This transparency greatly improves perceived responsiveness and helps users debug unexpected tool use.
import json
async def stream_with_progress(question, tools):
messages = [{'role': 'user', 'content': question}]
tool_buffers = {}
async for event_type, data in stream_with_tools(messages, tools):
if event_type == 'tool_calls':
tool_buffers = data
for tc in tool_buffers.values():
args = json.loads(tc['arguments'])
yield f'data: {json.dumps({"type": "tool_start", "function": tc["name"], "args": args})}\n\n'
result = tools_registry.get(tc['name'], lambda **kw: {})(** args)
yield f'data: {json.dumps({"type": "tool_done", "function": tc["name"]})}\n\n'
# Then stream final answer...Error Handling in Tool Call Streams
Tool execution can fail — APIs return errors, functions raise exceptions, JSON parsing fails. Always catch exceptions in tool execution and return a structured error response to the model. The model can then decide to retry with different arguments, call a fallback tool, or explain to the user that the requested action failed. Never let an uncaught tool exception crash the streaming loop.
def safe_execute_tool(func_name: str, args: dict) -> str:
try:
if func_name not in tools_registry:
return json.dumps({'error': f'Function {func_name!r} not found'})
result = tools_registry[func_name](**args)
return json.dumps(result)
except TypeError as e:
return json.dumps({'error': f'Invalid arguments: {str(e)}'})
except Exception as e:
return json.dumps({'error': f'Execution failed: {str(e)}'})
# Tool result message with error handled
tool_message = {
'role': 'tool',
'tool_call_id': tc['id'],
'content': safe_execute_tool(tc['name'], json.loads(tc['arguments'])),
}Comparing Streaming vs Non-Streaming for Agents
For agentic applications, streaming adds complexity but significant UX value. Without streaming, the user sees nothing during a multi-step tool-calling loop that might take 10-30 seconds. With streaming, they see intermediate text, tool call notifications, and the final answer appearing token by token. The added code complexity is usually worth it for interactive applications, but background agents that run autonomously can use non-streaming for simpler code.
Quick Check
Test your understanding of tool calls in streamed responses from this lesson.
Lesson Recap
In this lesson you learned: tool call arguments arrive as JSON fragments in streamed responses and must be buffered by tool call index before parsing, execute tools after stream completion then make a second streaming request for the final answer, and parallel execution with asyncio.gather minimizes latency when the model calls multiple functions simultaneously. Always catch tool execution errors to prevent crashing the agent loop. Next up we implement response caching to reduce API costs.
Frequently asked questions
Is the “Handling Tool Calls in Streamed Responses” lesson free?
Yes — the full text of “Handling Tool Calls in Streamed Responses” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.
What will I learn in “Handling Tool Calls in Streamed Responses”?
Parse streaming responses that contain function call arguments arriving token by token, buffer the JSON fragments, and trigger tool execution only when the call is complete. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Engineering Academy?
No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Handling Tool Calls in Streamed Responses” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Engineering Academy lesson?
Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Understanding Token Streaming
- Consuming Streams with the Python SDK
- Streaming in FastAPI with Server-Sent Events
- Handling Tool Calls in Streamed Responses