Chaining tool calls agent loop turns demands disciplined message handling; treat the conversation as a state machine, not a growing text buffer. Most agent bugs trace back to dropped tool_call_ids, unhandled parallel invocations, or missing termination conditions. This guide lays out an ordered path to build a robust multi-turn loop.
1. Define the agent loop contract
The loop is simple to describe: send messages and tool schemas to the model, inspect the assistant message for tool_calls, execute each call, append the results as tool messages, and repeat. The simplicity hides the invariants.
from openai import OpenAI
import json
client = OpenAI(base_url="https://api.openai.com/v1", api_key="sk-...")
def run_agent(messages, tools, max_turns=10):
for turn in range(max_turns):
resp = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
msg = resp.choices[0].message
messages.append(msg) # assistant message, may contain tool_calls
if not msg.tool_calls:
return msg.content # final answer
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,
"content": json.dumps(result),
})
raise TimeoutError("Agent exceeded max turns")execute_tool is your local dispatcher. The contract: every assistant message with tool_calls must be followed by exactly one tool message per call, referencing the same id.
2. Preserve tool call and result pairs exactly
A common mistake is storing only the text of the assistant reply or stripping tool_calls to save tokens. The OpenAI-compatible API requires the full assistant message, including the tool_calls array, to remain in the conversation history. The subsequent tool messages must cite the matching tool_call_id.
{
"role": "assistant",
"content": null,
"tool_calls": [
{
"id": "call_abc",
"type": "function",
"function": { "name": "get_weather", "arguments": "{\"city\":\"SF\"}" }
}
]
}{
"role": "tool",
"tool_call_id": "call_abc",
"content": "{\"temp\": 15}"
}Drop or mutate the id and the next model call will error or silently misalign results. Keep the raw response objects in your message list; do not “clean” them.
3. Manage state and context growth
Each turn appends assistant text, tool calls, and tool outputs. After a dozen turns with verbose JSON, you will hit the context limit. Three options: trim old messages, summarize, or use provider prompt caching for static prefixes.
When you use an inference gateway such as n4n.ai, it forwards provider cache-control hints, so you can mark your system prompt and tool schemas as cached across turns to avoid re-paying for static prefixes.
Tradeoff: trimming tool results can discard information the model needs later. A practical pattern is to keep a separate structured log of actions outside the message array and inject a compressed summary instead of raw outputs.
# naive trim: keep system + last N messages
if len(messages) > 24:
messages = [messages[0]] + messages[-20:]Do this only after you have confirmed the dropped content is not referenced by pending calls.
4. Execute parallel tool calls within a turn
Modern models emit multiple tool_calls in a single assistant message. Execute them concurrently, but do not call the model again until all results are appended. Interleaving a model call mid-turn violates the pairing contract.
import asyncio, json
async def run_tools(tool_calls):
async def call_one(tc):
res = await execute_tool_async(tc.function.name, json.loads(tc.function.arguments))
return {"role": "tool", "tool_call_id": tc.id, "content": json.dumps(res)}
return await asyncio.gather(*(call_one(tc) for tc in tool_calls))
# inside async loop
if msg.tool_calls:
results = await run_tools(msg.tool_calls)
messages.extend(results)Pitfall: logging concurrent executions in a way that makes debugging impossible. Tag each call with its tool_call_id and use structured logs.
5. Termination and max iterations
Always enforce max_turns. Termination is when the assistant message has no tool_calls. But models sometimes return empty content and no calls—treat that as final. Chaining tool calls agent loop turns becomes stable only when you also detect stagnation.
seen = {}
for tc in msg.tool_calls:
key = (tc.function.name, tc.function.arguments)
seen[key] = seen.get(key, 0) + 1
if seen[key] > 3:
messages.append({"role": "system", "content": "Stop repeating the same call; provide answer or try different approach."})Injecting a system note mid-loop is effective but can shift tone; use it as a circuit breaker, not a routine nudge.
6. Error isolation and retries
Tools fail: timeouts, schema drift, upstream 500s. Catch the exception and return it as a tool message. The model can often recover by retrying with different arguments or choosing another path.
try:
result = execute_tool(name, args)
except Exception as e:
result = {"error": str(e)}
messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})If the model call itself fails due to provider rate limits, wrap the create call with backoff. If you route through n4n.ai, automatic fallback when a provider is rate-limited or degraded keeps the loop running without custom retry code. Otherwise, implement your own provider rotation.
Never let a tool or network exception escape the loop—that terminates the agent prematurely.
7. Streaming and incremental UX
Streaming is tempting for chat UIs, but streaming tool_calls requires accumulating deltas across chunks to reconstruct the full call before execution. For backend agents, non-streaming is simpler and usually fast enough.
# conceptual streaming accumulation
assistant_msg = {"role": "assistant", "content": "", "tool_calls": []}
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
assistant_msg["content"] += delta.content
if delta.tool_calls:
# merge by index, append arguments fragments
...Tradeoff: streaming adds client complexity and makes atomic turn execution harder. Skip it unless a human is waiting on tokens.
8. Test the loop like a state machine
Mock the model to return scripted tool_calls and assert your executor appends correct tool messages. Verify that a missing tool_call_id raises. Verify max_turns fires.
def test_loop_handles_parallel():
fake_msgs = []
def fake_create(**kwargs):
return make_msg_with_calls([("a", "{}"), ("b", "{}")])
# swap client, run two turns, assert four tool messages appendedWhen chaining tool calls agent loop turns, test the failure paths too: inject a tool exception and confirm the loop continues, inject a rate-limit error and confirm backoff.
Tradeoffs summary
- Strict message preservation vs token cost: caching and trimming help, but never break the id pairing.
- Parallel execution vs debuggability: concurrent calls save latency but require disciplined logging.
- Gateway fallback vs direct control: a gateway reduces retry code, but you cede some routing visibility.
Build the loop as a state machine with explicit transitions, and the model will handle the reasoning. The engineering work is in the message bookkeeping.