Building a multi-step research agent with tool use

Building a multi-step research agent tool use pipeline requires more than a single LLM call. You need a loop that emits tool calls, executes them, and feeds results back until the model can answer. This tutorial walks through a concrete implementation using the OpenAI tool-calling protocol, real Wikipedia search, and an OpenAI-compatible gateway.

Prerequisites

Python 3.11 or newer
openai and requests libraries (pip install openai requests)
An API key for an OpenAI-compatible endpoint. We’ll point the client at n4n.ai’s single endpoint that fronts 240+ models, so we get automatic fallback if a provider is rate-limited.
Basic comfort with JSON Schema and Python functions

No frontend, no vector DB, no agent framework. Just the wire protocol and a loop.

Tool definitions

The model needs a contract. We define two tools: a search that returns candidate article titles, and a fetch that pulls the intro of a specific article. Keep descriptions terse but precise; the model relies on them to pick the right tool.

[
  {
    "type": "function",
    "function": {
      "name": "web_search",
      "description": "Search Wikipedia for a query and return up to N matches.",
      "parameters": {
        "type": "object",
        "properties": {
          "query": {"type": "string"},
          "limit": {"type": "integer", "default": 3}
        },
        "required": ["query"]
      }
    }
  },
  {
    "type": "function",
    "function": {
      "name": "fetch_article",
      "description": "Fetch the introductory extract of a Wikipedia article by title.",
      "parameters": {
        "type": "object",
        "properties": {
          "title": {"type": "string"}
        },
        "required": ["title"]
      }
    }
  }
]

Client and tool implementations

Set up the client. The base URL is the only gateway-specific line.

from openai import OpenAI
import json, requests

client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")
MODEL = "gpt-4o-mini"  # any model the gateway supports

def web_search(query: str, limit: int = 3):
    r = requests.get("https://en.wikipedia.org/w/api.php", params={
        "action": "query", "list": "search", "srsearch": query,
        "format": "json", "srlimit": limit
    }, timeout=10)
    data = r.json()["query"]["search"]
    return [{"title": i["title"], "snippet": i["snippet"]} for i in data]

def fetch_article(title: str):
    r = requests.get("https://en.wikipedia.org/w/api.php", params={
        "action": "query", "prop": "extracts", "exintro": 1,
        "explaintext": 1, "titles": title, "format": "json"
    }, timeout=10)
    pages = r.json()["query"]["pages"]
    return next(iter(pages.values()))["extract"]

These functions are plain Python. The agent loop will call them by name and pass parsed arguments.

The agent loop

A multi-step research agent tool use design centers on a while loop: call the model, check for tool_calls, execute, append results, repeat. Stop when the model returns text without tool calls.

tools = [...]  # the JSON above

def run_agent(user_query: str, max_steps: int = 8):
    messages = [{"role": "user", "content": user_query}]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model=MODEL, messages=messages, tools=tools
        )
        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)  # assistant message with tool_calls
        for tc in msg.tool_calls:
            fn = tc.function
            args = json.loads(fn.arguments)
            if fn.name == "web_search":
                result = web_search(**args)
            elif fn.name == "fetch_article":
                result = fetch_article(**args)
            else:
                result = {"error": "unknown tool"}
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result)
            })
    return "Agent exceeded step budget"

The assistant message with tool_calls must be appended verbatim to the conversation before the tool results. Omitting it breaks the protocol and the model loses context of what it asked.

Checkpoint: first pass

Run a simple query:

print(run_agent("What is the capital of France and who is its president?"))

Expected first model message (abbreviated):

{
  "role": "assistant",
  "tool_calls": [
    {"id": "call_1", "function": {"name": "web_search", "arguments": "{\"query\":\"capital of France\"}"}},
    {"id": "call_2", "function": {"name": "web_search", "arguments": "{\"query\":\"president of France\"}"}}
  ]
}

The loop executes both, appends two tool messages, and calls the model again. The second response typically contains the final answer. If you print messages after the first iteration, you’ll see the user message, the assistant tool_calls message, and two tool messages with tool_call_id fields matching the calls.

Parallel tool execution

The previous loop runs tool calls sequentially. For independent searches, that wastes latency. The OpenAI protocol allows multiple tool_calls in one assistant message; you should execute them concurrently.

from concurrent.futures import ThreadPoolExecutor

def execute(tc):
    fn = tc.function
    args = json.loads(fn.arguments)
    if fn.name == "web_search":
        return web_search(**args)
    if fn.name == "fetch_article":
        return fetch_article(**args)
    return {"error": "unknown tool"}

# inside the loop, replace the for-loop with:
if msg.tool_calls:
    messages.append(msg)
    with ThreadPoolExecutor() as ex:
        results = list(ex.map(execute, msg.tool_calls))
    for tc, res in zip(msg.tool_calls, results):
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": json.dumps(res)
        })
    continue

This cuts a two-search step from ~600 ms to ~300 ms if the HTTP calls are independent. The multi-step research agent tool use pattern gains the most from parallelism when the model fans out before synthesizing. You can extend this with asyncio if your tools are async-native; the threading version is enough for I/O-bound HTTP.

Full run with research query

Query:

answer = run_agent(
    "Research the 2024 Nobel Prize in Physics and summarize the winners' key contributions."
)
print(answer)

Typical flow:

Model calls web_search with "2024 Nobel Prize in Physics".
Tool returns titles like “2024 Nobel Prize in Physics”, “John Hopfield”, “Geoffrey Hinton”.
Model issues parallel fetch_article calls for the two laureates.
Model returns a synthesized paragraph.

Expected final output excerpt:

The 2024 Nobel Prize in Physics was awarded to John Hopfield and Geoffrey Hinton for foundational discoveries in machine learning. Hopfield developed a network for associative memory; Hinton invented Boltzmann machines enabling stochastic training…

(Exact text varies by model and temperature.)

Debugging the loop

When the agent returns garbage, dump messages as JSON before the final return. Common failure modes:

The model emits limit as a string "3". web_search(**args) will raise a TypeError. Coerce with int(args.get("limit", 3)).
A tool returns a massive extract and blows the context window. Truncate fetch_article to first 2000 chars.
The model calls a tool twice with the same args. That’s fine; caching the HTTP response locally saves tokens and latency.

Add a print(f"step {i}: {msg.tool_calls}") inside the loop during development. You’ll quickly see whether the model is planning or flailing.

Handling degradation

Because we route through n4n.ai, a provider outage on one backend doesn’t abort the loop—the gateway fails over to another model that supports the same call shape. Your code stays identical. Per-token usage metering is reported in the response, so you can log resp.usage per step without extra instrumentation.

When building multi-step research agent tool use in production, wrap client.chat.completions.create with a retry on APIConnectionError and cap max_steps to avoid runaway cost. A step budget of 8 is reasonable for research tasks; raise it only after measuring real traces.

Production notes

Schema drift: Validate tool arguments with pydantic before execution. The model occasionally hallucinates optional parameters not in the schema.
Cache hints: If your gateway honors provider cache-control, prefix stable system instructions with cache_control: ephemeral via extra headers to cut repeat token cost on long tool schemas.
Idempotency: Tool calls that hit external APIs should be safe to retry. Wikipedia GETs are; a “send email” tool is not. Mark side-effecting tools explicitly and consider human confirmation.
Streaming: For UX, stream the final assistant text. Tool steps stay non-streamed; you only stream when tool_calls is empty. The stream=True flag works identically on the chat endpoint.
Observability: Emit one span per tool execution with the tool_call_id as trace state. Correlating tool latency with model steps is the fastest way to find bottlenecks.

The loop above is ~70 lines. That’s the entire core of a research agent. Everything else is guardrails, caching, and telemetry. The protocol is stable; the hard part is deciding when to stop and how to present partial results.