Gestión de llamadas a herramientas en respuestas en streaming
Analice respuestas en streaming que contengan argumentos de llamadas a funciones recibidos token a token, almacene en búfer los fragmentos JSON y active la ejecución de la herramienta solo cuando la llamada esté completa.
Gestión de llamadas a herramientas en respuestas en streaming es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.
Partes de esta lección aún no han sido traducidas y se muestran en inglés.
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.
Preguntas frecuentes
¿La lección «Gestión de llamadas a herramientas en respuestas en streaming» es gratis?
Sí — el texto completo de «Gestión de llamadas a herramientas en respuestas en streaming» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.
¿Qué aprenderé en «Gestión de llamadas a herramientas en respuestas en streaming»?
Analice respuestas en streaming que contengan argumentos de llamadas a funciones recibidos token a token, almacene en búfer los fragmentos JSON y active la ejecución de la herramienta solo cuando la… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Engineering Academy?
No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.
¿Cuánto tiempo toma la lección «Gestión de llamadas a herramientas en respuestas en streaming»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?
Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Comprensión de la transmisión de tokens
- Consumo de streams con el SDK de Python
- Streaming en FastAPI con eventos enviados por el servidor
- Gestión de llamadas a herramientas en respuestas en streaming