> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cfosilvia.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 04 streaming tools

# Streaming with Tool Calls

Combining streaming with tool calling requires special handling. Tool calls arrive as fragments that must be accumulated.

## How Tool Calls Stream

Tool calls arrive in delta fragments, not as complete objects:

```
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_abc","function":{"name":"search"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"q"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"uery\":"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":" \"weather"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"}"}}]}}]}
data: [DONE]
```

## Key Learnings ⚠️

### 1. `id` May Be Null

The `tool_call.id` might come as `null` in some responses. Always check:

```python theme={null}
# ❌ Wrong - may set id to None
if "id" in tc_delta:
    tool_calls[idx]["id"] = tc_delta["id"]

# ✅ Correct - only set if truthy
if tc_delta.get("id"):
    tool_calls[idx]["id"] = tc_delta["id"]
```

**Fallback ID:** If `id` is still empty after streaming completes, generate one:

```python theme={null}
import uuid
if not tool_calls[idx]["id"]:
    tool_calls[idx]["id"] = f"call_{uuid.uuid4().hex[:24]}"
```

### 2. Arguments Are Fragmented

Arguments stream character-by-character. Accumulate them:

```python theme={null}
if tc_delta.get("function", {}).get("arguments"):
    tool_calls[idx]["function"]["arguments"] += tc_delta["function"]["arguments"]
```

### 3. Index Identifies Tool Calls

For parallel tool calls, `index` identifies which tool call to update:

```python theme={null}
idx = tc_delta.get("index", 0)
if idx not in tool_calls:
    tool_calls[idx] = {"id": "", "function": {"name": "", "arguments": ""}}
```

## Complete Streaming Tool Handler

```python theme={null}
import json
import uuid
import httpx

def stream_with_tools(messages, tools):
    """Stream a response and accumulate any tool calls."""
    
    content_parts = []
    tool_calls = {}  # index -> {id, type, function: {name, arguments}}
    
    with httpx.Client(timeout=300.0) as client:
        with client.stream("POST", ENDPOINT, json={
            "messages": messages,
            "tools": tools,
            "stream": True
        }) as response:
            for chunk in response.iter_text():
                for line in chunk.split("\n"):
                    if not line.startswith("data: "):
                        continue
                    data = line[6:]
                    if data == "[DONE]":
                        continue
                    
                    parsed = json.loads(data)
                    delta = parsed["choices"][0].get("delta", {})
                    
                    # Accumulate content
                    if delta.get("content"):
                        content_parts.append(delta["content"])
                    
                    # Accumulate tool calls
                    if "tool_calls" in delta:
                        for tc in delta["tool_calls"]:
                            idx = tc.get("index", 0)
                            
                            if idx not in tool_calls:
                                tool_calls[idx] = {
                                    "id": "",
                                    "type": "function",
                                    "function": {"name": "", "arguments": ""}
                                }
                            
                            # Only set id if truthy
                            if tc.get("id"):
                                tool_calls[idx]["id"] = tc["id"]
                            
                            func = tc.get("function", {})
                            if func.get("name"):
                                tool_calls[idx]["function"]["name"] = func["name"]
                            if func.get("arguments"):
                                tool_calls[idx]["function"]["arguments"] += func["arguments"]
    
    # Generate fallback IDs if needed
    for idx in tool_calls:
        if not tool_calls[idx]["id"]:
            tool_calls[idx]["id"] = f"call_{uuid.uuid4().hex[:24]}"
    
    # Build result
    result = {
        "role": "assistant",
        "content": "".join(content_parts) or ""
    }
    if tool_calls:
        result["tool_calls"] = [tool_calls[i] for i in sorted(tool_calls.keys())]
    
    return result
```

## Multi-Turn Streaming Loop

```python theme={null}
while True:
    msg = stream_with_tools(messages, tools)
    
    if not msg.get("tool_calls"):
        print(msg["content"])
        break
    
    messages.append(msg)
    
    for tc in msg["tool_calls"]:
        result = execute_tool(tc["function"]["name"], 
                              json.loads(tc["function"]["arguments"]))
        messages.append({
            "role": "tool",
            "tool_call_id": tc["id"],
            "name": tc["function"]["name"],
            "content": json.dumps(result)
        })
```

## See Also

* [tests/test\_stream\_tools.py](../tests/test_stream_tools.py) - Full working example


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.