Building reliable multi-step tool use autonomous agents requires more than chaining a few function calls. You need a deliberate loop that plans, executes, observes, and recovers. This guide lays out an ordered path to ship agents that handle parallel and sequential tool calls without collapsing under partial failure.
1. Define a strict tool contract
The first failure point is a loose tool schema. If your parameters object allows optional fields the model fills with garbage, every downstream step inherits that garbage. Write explicit Pydantic models and derive the JSON schema from them.
from pydantic import BaseModel, Field
class SearchArgs(BaseModel):
query: str = Field(..., description="Exact SQL-like filter string")
top_k: int = Field(3, ge=1, le=10)
class FetchPriceArgs(BaseModel):
sku: str = Field(..., pattern=r"^[A-Z0-9]{6,12}$")Expose only the fields the model must provide. Hide internal routing, retries, and auth behind the executor. Multi-step tool use autonomous agents amplify small schema ambiguities into cascading nulls, so treat the schema as a wire protocol, not a suggestion.
Validate the model’s arguments with the model before calling the backend. A 20-token validation failure is cheaper than a 2,000-token side-effect call.
2. Separate planning from execution
Do not ask the model to both decide and act in the same turn. Use a planner prompt that returns a static step graph. This makes the run inspectable and lets you reject unsafe plans before spending tokens on execution.
{
"steps": [
{"id": "s1", "tool": "search", "args": {"query": "gpu>=80gb"}, "depends_on": []},
{"id": "s2", "tool": "fetch_price", "args": {"sku": "{{s1.hits[0].sku}}"}, "depends_on": ["s1"]}
]
}The planner should emit dependency edges. If it can’t, your executor must infer them, which is slower and error-prone. Keep the planner stateless; pass it the current world state as a compact summary. Reject plans that touch payment or delete tools unless a human flag is set in the client request.
A good planner prompt includes: available tool names, their schemas, the current state summary, and a hard limit on step count. It does not include the full conversation history—that belongs to the execution observer.
3. Execute with explicit state
Maintain a single state dict keyed by step id. Never let the model mutate it directly. The executor writes outputs, and a template renderer resolves {{step_id.path}} references before each call.
state = {}
for step in ordered_steps:
if any(state[d] is None for d in step["depends_on"]):
state[step["id"]] = None # mark skipped
continue
args = render(step["args"], state)
state[step["id"]] = run_tool(step["tool"], args)Skipped steps propagate as None. That forces the planner (or a repair step) to acknowledge the gap instead of hallucinating a value. The render function must fail closed: if a referenced path is missing, return None rather than an empty string. Empty strings sneak through validators and cause silent bad calls.
Persist state to disk or a key-value store after each layer. If the process restarts, you resume from the last layer instead of replaying side effects.
4. Decide parallel vs sequential at runtime
Parallelism cuts latency but multiplies failure surface. Compute a topological layer: steps with no unresolved dependencies run together. If a layer has side effects (POST, payment), force sequential regardless.
def layers(steps):
ready, done = [], set()
remaining = steps.copy()
while remaining:
layer = [s for s in remaining if set(s["depends_on"]) <= done]
if not layer:
raise CycleError("Step graph has a cycle")
ready.append(layer)
done |= {s["id"] for s in layer}
remaining = [s for s in remaining if s not in layer]
return readyFor read-only tools, fire the layer with asyncio.gather. For writes, await one by one and checkpoint. Tradeoff: parallel reads can trip rate limits on shared credentials. Cap concurrent requests at the lowest documented limit for the underlying API.
If a parallel layer partially fails, do not auto-retry the whole layer. Retry only the failed steps; their siblings already succeeded and replaying them may double-charge.
5. Handle partial failure with compensation
Most agents ignore idempotency. If charge_card succeeds but send_email fails, you have a silent partial order. Tag each tool with idempotency_key = f"{run_id}:{step_id}" and implement a compensating action where feasible.
async def run_tool(name, args, key):
try:
return await registry[name](**args, idempotency_key=key)
except RateLimitError:
await asyncio.sleep(2)
return await registry[name](**args, idempotency_key=key) # safe replay
except ValidationError as e:
return {"error": str(e)}Common pitfall: treating all errors as retryable. A 400 from a search API means your query is bad; retrying wastes tokens. A 429 means back off. Classify before looping.
Compensation example: if reserve_inventory succeeded but charge_card failed, call release_inventory with the same reservation id. Without that, multi-step tool use autonomous agents leave dangling holds that block real customers.
6. Stream intermediate results back to the model
After each layer, send a compact observation to the model: succeeded step ids, truncated outputs, and explicit nulls. This lets multi-step tool use autonomous agents re-plan instead of plowing ahead.
obs = {"s1": state["s1"][:200], "s2": None, "note": "s2 skipped: s1 empty"}
messages.append({"role": "user", "content": json.dumps(obs)})Keep observations under a token budget. Truncate lists, drop verbose stack traces. The model needs shape, not dumps. If the observation shows a dependency returned None, the model should emit a repair plan (e.g., broader search) rather than inventing a SKU.
7. Enforce budgets and fallback
Autonomous loops burn tokens fast. Set a hard cap on steps (e.g., 12) and a token ceiling per run. If you route through a gateway like n4n.ai, automatic fallback when a provider is rate-limited keeps a long agent loop from dying mid-plan, and per-token metering makes the ceiling enforceable. Gateways that honor client routing directives and forward provider cache-control hints—such as n4n.ai—let you pin the planner to a cached completion, cutting cost on repeated state summaries.
Do not let the agent call the planner recursively without a decrementing counter. A planner that plans to plan is how you get $40 runs. Log every planner invocation with its token cost before executing.
8. Minimal reference implementation
Below is a tight loop combining the above. It uses the OpenAI-compatible chat endpoint, which any compliant gateway serves.
from openai import OpenAI
import json, asyncio
client = OpenAI(base_url="https://api.example-gateway.com/v1", api_key="sk-...")
def plan(state_summary):
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role":"system","content":"Emit step graph JSON."},
{"role":"user","content":state_summary}],
response_format={"type":"json_object"})
return json.loads(resp.choices[0].message.content)["steps"]
async def execute(steps, state):
for layer in layers(steps):
tasks = [run_tool(s["tool"], render(s["args"], state), f"run:{s['id']}") for s in layer]
results = await asyncio.gather(*tasks, return_exceptions=True)
for s, r in zip(layer, results):
state[s["id"]] = None if isinstance(r, Exception) else r
# main guard omitted for brevitySwap example-gateway for your endpoint. The contract stays identical.
Common pitfalls to avoid
- Hidden state in prompts: If the model must remember a value from step 3 to use in step 7, put it in
state, not in conversation history. History gets truncated. - Over-parallelism: Firing 20 scrapes at once triggers IP bans. Batch with respect to rate limits.
- No dead-letter queue: Failed steps should land in a structured log, not just a
None. You will debug from that log. - Assuming tool schemas are stable: Providers change field names. Pin versions or validate on startup.
- Skipping compensation tests: Write a unit test that forces
chargeto fail afterreserveand assertsreleaseis called.
Multi-step tool use autonomous agents are only as good as the edges between steps. Build the graph, execute it visibly, and fail loudly.