Most LLM APIs dump tool calls as a single JSON blob at the end of a response. sse streaming partial tool calls changes that: the model emits argument fragments incrementally over Server-Sent Events, and your client must reassemble them before execution. This guide walks through a complete implementation against an OpenAI-compatible chat completions endpoint.
Step 1: Configure the chat completion request for streaming tool calls
Set stream: true and supply a tools array. The server returns delta.tool_calls items inside each chunk instead of a finalized message.tool_calls at the end.
{
"model": "gpt-4o-mini",
"stream": true,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"parameters": {
"type": "object",
"properties": { "city": { "type": "string" } },
"required": ["city"]
}
}
}
],
"messages": [{ "role": "user", "content": "Weather in Paris?" }]
}The first chunk containing a tool call typically carries id, type, and function.name. Subsequent chunks carry only function.arguments fragments. If you front your requests with an OpenAI-compatible gateway such as n4n.ai, automatic fallback on provider rate limits keeps the sse streaming partial tool calls connection alive without client changes.
Step 2: Open the SSE stream and read raw lines
Use a client that does not buffer the response body. In Python, requests with stream=True works; in Node, fetch with response.body.getReader() is correct.
import os, json, requests
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
"Content-Type": "application/json",
}
payload = { ... } # from Step 1
with requests.post(url, json=payload, headers=headers, stream=True) as r:
for raw in r.iter_lines():
if not raw:
continue
if raw.startswith(b"data: "):
event = raw[6:].decode("utf-8").strip()
if event == "[DONE]":
break
chunk = json.loads(event)
# handle chunk (next steps)The iter_lines helper splits on newlines, which matches the SSE framing of data: {json}\n\n. Do not assume each network packet is one event.
TypeScript reader
const resp = await fetch(url, { method: "POST", headers, body: JSON.stringify(payload) });
const reader = resp.body!.getReader();
const decoder = new TextDecoder();
let buf = "";
const acc: Record<number, any> = {};
while (true) {
const { value, done } = await reader.read();
if (done) break;
buf += decoder.decode(value, { stream: true });
let idx;
while ((idx = buf.indexOf("\n\n")) !== -1) {
const event = buf.slice(0, idx); buf = buf.slice(idx + 2);
if (event.startsWith("data: ")) {
const data = event.slice(6);
if (data === "[DONE]") break;
handleChunk(JSON.parse(data), acc);
}
}
}This manual buffer split is required because fetch exposes a byte stream, not line semantics.
Step 3: Extract tool call deltas from each chunk
Every chunk has choices[0].delta. When the model invokes a tool, delta.tool_calls is a list of objects keyed by index. The index identifies which parallel tool call the fragment belongs to.
def handle_chunk(chunk, acc):
delta = chunk["choices"][0]["delta"]
for tc in delta.get("tool_calls", []) or []:
idx = tc["index"]
if idx not in acc:
acc[idx] = {
"id": tc.get("id"),
"type": tc.get("type", "function"),
"function": {"name": "", "arguments": ""},
}
if tc.get("id"):
acc[idx]["id"] = tc["id"]
if tc.get("function", {}).get("name"):
acc[idx]["function"]["name"] = tc["function"]["name"]
if tc.get("function", {}).get("arguments"):
acc[idx]["function"]["arguments"] += tc["function"]["arguments"]This accumulation pattern is the core of sse streaming partial tool calls: you stitch fragments per index, not per request.
Step 4: Reconstruct and parse partial arguments safely
The arguments field is a JSON string built incrementally. It may be {, then "city", then :"Par, then is"}. Do not call json.loads on each fragment. Wait until the stream closes, then parse the full string.
If you need to act on partial arguments (e.g., to show a live UI), use an incremental JSON parser such as jsonyx or a stack-based scanner. For execution, defer until complete.
import json
def finalize(acc):
calls = []
for idx in sorted(acc):
item = acc[idx]
args_str = item["function"]["arguments"]
try:
args = json.loads(args_str)
except json.JSONDecodeError:
args = {} # or raise, depending on strictness
calls.append({
"id": item["id"],
"type": item["type"],
"function": {"name": item["function"]["name"], "arguments": args},
})
return callsA robust client treats malformed final arguments as a stream error and retries the request.
Step 5: Execute the tool and continue the conversation
After finalizing, call your local function. Then return the result as a role: "tool" message with the matching tool_call_id.
def get_weather(city):
return f"Sunny in {city}"
messages = [{"role": "user", "content": "Weather in Paris?"}]
# assume calls[0] is the weather call
tool_result = get_weather(calls[0]["function"]["arguments"]["city"])
messages.append({
"role": "tool",
"tool_call_id": calls[0]["id"],
"content": tool_result,
})Send messages back with stream: true again if you want the model’s final natural language answer streamed. The same sse streaming partial tool calls logic applies if the model chains another tool.
Step 6: Handle multiple parallel tool calls
Models can emit several tool_calls with different indices in the same or interleaved chunks. Your accumulator must keep them independent. When finalizing, preserve index order; most runtimes execute them concurrently.
import concurrent.futures
def run_calls(calls):
with concurrent.futures.ThreadPoolExecutor() as ex:
futures = {ex.submit(execute, c): c for c in calls}
for f in concurrent.futures.as_completed(futures):
c = futures[f]
result = f.result()
messages.append({"role":"tool","tool_call_id":c["id"],"content":result})Do not assume the first index is the only one. SSE ordering is per-index, not global.
Step 7: Verify success
Verification is concrete: assert that the streamed arguments parse to valid JSON, that the tool executed, and that the follow-up completion finishes. A minimal test:
def test_stream():
acc = {}
for chunk in captured_chunks:
handle_chunk(chunk, acc)
calls = finalize(acc)
assert calls[0]["function"]["name"] == "get_weather"
assert calls[0]["function"]["arguments"]["city"] == "Paris"
assert get_weather("Paris")
print("OK: sse streaming partial tool calls reconstructed correctly")Run this against a recorded stream or a live call with curl -N and pipe through jq to eyeball fragments:
curl -N https://api.example.com/v1/chat/completions \
-H "Authorization: Bearer $KEY" -H "Content-Type: application/json" \
-d @payload.json | grep '^data:' | tail -5You should see the arguments string grow across lines and a final [DONE]. If the connection drops mid-argument, the accumulator will have a truncated JSON; treat that as a retry signal.
Edge cases that will bite you
- Blank lines: SSE requires an empty line between events. Ignore empty
rawbytes. - Role fragments: Some providers send
delta.roleon the first chunk; ignore it for tool logic. - Id latency:
idmay arrive a chunk afterindex. Always key onindex, backfillid. - Arguments empty string: A chunk may carry
arguments: ""legitimately; concatenating is safe. - Provider cache hints: Gateways that forward
cache-controlheaders can reuse prompt prefixes; your streaming parser stays identical.
Getting sse streaming partial tool calls right means treating the stream as a state machine, not a queue of messages. Build the accumulator once, feed it every chunk, and execute only on a clean finish.