openai SDK pointing at a local Ollama server. View Claude version → · OS Track Index →
M06: Multi-Tool Orchestration
In M05 you gave the model one tool at a time. Now you'll orchestrate multiple tools together — running them in parallel, chaining their outputs, and handling failures gracefully. This is where agents become truly powerful.
Learning Objectives
- Explain the difference between parallel and sequential tool calls, and when to use each
- Implement the agentic loop that processes multiple tool calls and chains results
- Engineer effective tool descriptions that help the model select the right tool
- Build a ToolRegistry that dynamically adds and removes tools based on context
- Handle partial failures, retries, and circuit breakers in multi-tool workflows
Parallel Tool Calls — When and Why
BEFORE: Imagine a kitchen where only one sous chef handles all prep work — chopping onions, then dicing tomatoes, then mincing garlic, one task after another while the head chef waits.
PAIN: Dinner service grinds to a halt because three 10-minute tasks take 30 minutes total, and every dish that depends on those ingredients sits idle the entire time.
MAPPING: Parallel tool callsWhen the model requests multiple tool calls in a single response because the calls are independent. The client executes all tools concurrently and returns each result as its own role:"tool" message, reducing round trips and wall-clock time. solve this exactly the way a smart head chef would — assign each task to a different sous chef so all three run simultaneously. The total time drops to the slowest single task (10 minutes), not the sum of all three (30 minutes). In agent terms, the model emits multiple tool_calls in one response, your code runs them concurrently, and the wall-clock time equals the slowest tool, not the total.
tool_callsEntries in response.choices[0].message.tool_calls. Each has a unique id, a function.name, and a function.arguments string. Your code reads them, executes the functions, and returns the results. When multiple tool_calls appear in one response, the calls are independent and can run in parallel. in a single assistant message when the calls are independent. Your client code then executes all of those tools concurrently. Remember from M05: your code runs the tools, not the model. Once all tools finish, you append all the tool resultsEach tool result is a message with role:"tool" carrying the matching tool_call_id and the output string. For parallel calls you append one such message per tool before the next API call. — one role:"tool" message per tool — and send them back in the next turn. The payoff? One API round trip handles N tools, instead of N separate round trips.
So what does parallel execution actually look like in an API response? When the model decides it needs multiple independent tools, it returns multiple tool_calls in a single response. Here's an example:
Three tool_calls, each with its own id. Your code sees all three, fires them off concurrently (using ThreadPoolExecutor in Python or Promise.all in Node.js), and appends all three results — one role:"tool" message each — before the next call. One round trip, three tools executed.
Sequential Tool Chains — Output Feeds Input
BEFORE: Imagine trying to build a car by dumping all the raw materials — steel, rubber, glass — into a room and hoping a finished vehicle appears. Without a defined sequence, nothing fits together.
PAIN: You can't install a windshield before the frame is welded, and you can't paint the body before it's assembled. Doing steps out of order wastes materials and produces a broken result.
MAPPING: A sequential tool chain works like an assembly line — Station A (search) produces URLs, Station B (fetch_page) takes those URLs and produces page text, Station C (summarize) takes that text and produces bullet points. Each stage transforms the data and passes it forward. Skipping a station or running them out of order means the next stage gets the wrong input and the whole pipeline breaks.
finish_reasonA field in the API response indicating why generation stopped. "stop" means the model finished normally. "tool_calls" means the model wants to call a tool and is waiting for your code to execute it and return the result. becomes "stop" — meaning the model has finished its work and is ready to present a final answer.
Here's what a sequential chain looks like in practice. Notice how each step's output becomes the next step's input — data transforms at each stage:
Each round trip is a full API call. The model sees the previous result, decides what to do next, and issues another tool call. This continues until the chain is complete and the model returns its final text response.
Tool Selection — How the Model Picks the Right Tool
BEFORE: Imagine walking into a hardware store where every tool is in an unmarked cardboard box — no labels, no descriptions, just numbered bins. You need to drive a small screw into a circuit board, but you have no idea which bin has the right screwdriver.
PAIN: You end up grabbing random boxes, trying tools that don't fit, stripping the screw, and wasting an hour on a two-minute job. Worse, you might use a power drill and destroy the delicate board entirely.
MAPPING: This is exactly what happens when the model gets tool definitions with vague descriptions like "searches stuff". The model reads the name and description of each tool to decide which one fits the user's request. A label like "Phillips-head screwdriver, size #2, for small electronics" maps directly to a good tool description: "Search the web using a query string. Returns top 5 results. Use for current events or factual questions." The clearer the label, the more accurate the selection.
fetch_page next than send_email.
Tool Description Engineering
"name": "search", "description": "searches stuff" — too vague. The model doesn't know what it searches, when to use it, or what it returns.
"name": "web_search", "description": "Search the web using a query string. Returns top 5 results with title, URL, and snippet. Use for current events, factual questions, or when the user asks to look something up online."
Keep 4–5 tools per agent maximum. Tool selection accuracy degrades rapidly above 5. Anti-pattern: one agent with 18+ tools. Instead, distribute tools across specialized subagents.
"More tools = more capable agent" — This is the most counterintuitive misconception in multi-tool orchestration. In practice, tool selection accuracy degrades noticeably once you pass 5–6 tools. Each additional tool adds more descriptions for the model to evaluate, more chances for ambiguity between similar tools, and more input tokens per request. An agent with 4 focused tools will outperform an agent with 18 scattered ones almost every time.
"The model automatically parallelizes independent tools" — The model CAN return multiple tool call blocks in one response, and it often does when the calls are clearly independent. But you still need to write the parallel execution logic in YOUR code (ThreadPoolExecutor, Promise.all). If your code processes tool calls sequentially even when the model sends multiple, you lose the speedup. Parallelism requires effort on both sides.
"Sequential is always worse than parallel" — Not at all. When Tool B needs the output of Tool A (e.g., fetch a page from a URL that search returned), they MUST run sequentially. Forcing parallel execution on dependent tools produces incorrect results — Tool B would run with no input. The right approach is to parallelize independent tools and chain dependent ones.
"Dynamic registration is premature optimization" — For a 3-tool agent, yes. For a production agent with 15–20 tools, it's essential. Each tool definition consumes 200–500 input tokens. Sending 20 tools with every request means 4,000–10,000 extra tokens per call — that's real cost at scale. And the accuracy benefit of fewer tools is arguably more important than the token savings.
"The model always picks the right tool" — The model is remarkably good at tool selection, but it's not infallible. Ambiguous descriptions, overlapping tool capabilities, and misleading parameter names all cause misselection. This is why tool description engineering matters — it's the single highest-leverage thing you can do to improve agent reliability.
Dynamic Tool Registration
BEFORE: Imagine a surgeon walking into the operating room and finding every instrument the hospital owns laid out on the tray — orthopedic saws, dental drills, eye surgery lasers, and the cardiac tools they actually need. Hundreds of instruments, all within reach.
PAIN: The surgeon wastes time scanning past irrelevant tools, risks grabbing the wrong instrument under pressure, and the tray is so cluttered that the correct scalpel is buried under equipment meant for a completely different specialty.
MAPPING: Dynamic tool registration is like a surgical nurse who curates the tray — only cardiac instruments are laid out for a heart surgery. In agent terms, instead of sending all 20 tools with every API call, you filter the tools array based on the current task context. Fewer tools means the model scans less, picks more accurately, and you burn fewer input tokens on irrelevant definitions.
tools array you pass gets serialized into input tokens — the model reads every tool's name, description, and parameter schema before deciding which to use. This is called token overheadTool definitions consume input tokens. Each tool's name, description, and parameter schema are serialized and sent with every API call. Sending 20 tools when only 3 are relevant wastes tokens on every request.. Dynamic tool registration means you build a different tools array for each request based on what the user actually needs. Why bother? Three reasons: (1) Cost — fewer tools means fewer input tokens burned on every call. (2) Accuracy — the model picks better when it has 4 focused options instead of 20 scattered ones. (3) Security via least privilegeA security principle: give each request access only to the tools it actually needs. An admin-only tool like delete_user should not be available during a regular user's research query. — a regular user's research query should never see the delete_user tool, even if it exists in your system.
Handling Errors in Multi-Tool Workflows
BEFORE: Imagine a relay race where the team has no backup plan — four runners, one baton, and if anyone trips, the entire team is disqualified. No substitutes, no recovery protocol.
PAIN: In the real race, the second runner twists an ankle at the handoff. Without a plan, the baton hits the ground, the team freezes, and they forfeit a race they were winning. One failure cascades into total failure.
MAPPING: Multi-tool workflows face the same risk — if fetch_page returns a 404, does your entire agent crash? Error handling gives you the backup plan: return a structured error so the model can reason about alternatives (like switching to web_search), implement retries with exponential backoff for transient failures, and add circuit breakers that disable a tool after repeated failures so the agent doesn't waste time on a broken endpoint.
That relay race analogy maps directly to real tool result messages. When a tool fails, you don't just drop the baton — you hand the model a structured report explaining what went wrong, so it can pick a new runner. Here's exactly what that looks like in practice:
The model makes very different decisions based on which one you return. The first says "the endpoint is broken, try something else." The second says "nothing matches your query." Confusing the two — returning an empty result when the tool actually crashed — leads the model to conclude there's genuinely no data, which is a silent, hard-to-debug failure.
Error Handling Strategies
- Per-tool try/catch: Wrap each tool in error handling; return descriptive messages as tool result
- Let the model adapt: Return structured error JSON — the model can often find alternatives on its own
- Tool-level retries: Exponential backoff for transient failures (timeouts, rate limits)
- Circuit breakers: After N consecutive failures, disable the tool and notify the user
- Graceful degradation: Return partial results — some data is better than no data
"If a tool fails, just return an empty result" — This is one of the most dangerous mistakes in agent development. An empty result ({"results": []}) tells the model "I checked and found nothing." An error message tells the model "I couldn't even check." The model makes completely different decisions based on which one you return. The first leads to "there's no data on this topic." The second leads to "let me try a different approach." Confusing them creates silent, hard-to-debug failures.
"Retrying a failed tool 100 times will eventually work" — Brute-force retries burn tokens and time. If an endpoint is down, it's down. Use exponential backoff (wait 1s, then 2s, then 4s) with a maximum of 3 retries for transient failures like timeouts. For persistent failures, use a circuit breaker instead.
"The model can recover from any error automatically" — The model is good at adapting when you give it structured error information. But it can only work with the tools you've provided. If the only way to get data is through a broken tool and no alternative exists, the model will inform the user rather than magically producing correct information. Always design your tool set with fallback options where possible.
Circuit Breakers in Depth
A circuit breaker is borrowed from electrical engineering — when too much current flows through a wire, the breaker trips and cuts the circuit to prevent a fire. In agent code, it works the same way: you track consecutive failures for each tool, and after a threshold (typically 3–5 failures in a row), you temporarily disable that tool. This prevents the agent from wasting tokens and time retrying a broken endpoint over and over.
Here's how it works internally. You maintain a counter per tool (e.g., {"fetch_page": 0, "web_search": 0}). Every time a tool succeeds, its counter resets to zero. Every time it fails, the counter increments. When the counter hits your threshold, the circuit "opens" — meaning that tool is removed from the tools array sent to the model on the next API call. The model never even sees it as an option, so it naturally picks alternatives. After a cooldown period (say 60 seconds), you can "half-open" the circuit by adding the tool back for one test call to see if the endpoint has recovered.
How does this differ from simple retries? Retries happen within a single tool call — you might try the same HTTP request 3 times with exponential backoff before giving up. Circuit breakers operate across tool calls — they track a pattern of repeated failures over time and make a system-level decision to stop using that tool entirely. You'd typically use both together: retry transient failures within a call, and circuit-break persistent failures across calls.
Code Walkthrough: Research Assistant Agent
This agent demonstrates all orchestration patterns: parallel search, sequential fetch-and-summarize, dynamic tool registration, and error recovery.
Step 1: Define the Tools
{"type": "function", "function": {...}} with a parameters key (not input_schema). The model reads these schemas to decide which tools to call and what arguments to pass. It never executes them directly. The interesting part is the description field — notice how each one explains not just what the tool does, but when to use it. Vague descriptions like "does stuff" leave the model guessing, and a guessing agent is an unreliable agent.
# pip install openai
from openai import OpenAI
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
tools = [
{
"type": "function",
"function": {
"name": "web_search",
"description": (
"Search the web for current information. Returns top 3 "
"results with title, URL, and snippet. Use for recent "
"events, factual questions, or general research."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "fetch_page",
"description": (
"Fetch the full text content of a web page by URL. "
"Returns page text (max 5000 chars). Use after "
"web_search to get full content from a result URL."
),
"parameters": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "Full URL to fetch"}
},
"required": ["url"]
}
}
},
{
"type": "function",
"function": {
"name": "summarize_text",
"description": (
"Summarize long text into key points (3-5 bullets). "
"Use after fetch_page to condense page content."
),
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to summarize"},
"max_points": {
"type": "integer",
"description": "Max bullet points (default 5)"
}
},
"required": ["text"]
}
}
},
{
"type": "function",
"function": {
"name": "format_citation",
"description": (
"Format a source as an academic citation. Use after "
"summaries are ready to create proper references."
),
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Article title"},
"url": {"type": "string", "description": "Source URL"},
"accessed_date": {"type": "string", "description": "e.g. '2025-01-15'"}
},
"required": ["title", "url"]
}
}
},
{
"type": "function",
"function": {
"name": "save_to_file",
"description": "Save content to a local file. Returns file path.",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "Output filename"},
"content": {"type": "string", "description": "Content to save"}
},
"required": ["filename", "content"]
}
}
}
]// npm install openai
import OpenAI from 'openai';
const client = new OpenAI({ baseURL: 'http://localhost:11434/v1', apiKey: 'ollama' });
const tools = [
{
type: "function",
function: {
name: "web_search",
description:
"Search the web for current information. Returns top 3 " +
"results with title, URL, and snippet. Use for recent " +
"events, factual questions, or general research.",
parameters: {
type: "object",
properties: {
query: { type: "string", description: "Search query" }
},
required: ["query"]
}
}
},
{
type: "function",
function: {
name: "fetch_page",
description:
"Fetch the full text content of a web page by URL. " +
"Returns page text (max 5000 chars). Use after " +
"web_search to get full content from a result URL.",
parameters: {
type: "object",
properties: {
url: { type: "string", description: "Full URL to fetch" }
},
required: ["url"]
}
}
},
{
type: "function",
function: {
name: "summarize_text",
description:
"Summarize long text into key points (3-5 bullets). " +
"Use after fetch_page to condense page content.",
parameters: {
type: "object",
properties: {
text: { type: "string", description: "Text to summarize" },
max_points: { type: "integer", description: "Max bullet points (default 5)" }
},
required: ["text"]
}
}
},
{
type: "function",
function: {
name: "format_citation",
description:
"Format a source as an academic citation. Use after " +
"summaries are ready to create proper references.",
parameters: {
type: "object",
properties: {
title: { type: "string", description: "Article title" },
url: { type: "string", description: "Source URL" },
accessed_date: { type: "string", description: "e.g. '2025-01-15'" }
},
required: ["title", "url"]
}
}
},
{
type: "function",
function: {
name: "save_to_file",
description: "Save content to a local file. Returns file path.",
parameters: {
type: "object",
properties: {
filename: { type: "string", description: "Output filename" },
content: { type: "string", description: "Content to save" }
},
required: ["filename", "content"]
}
}
}
];Step 2: Implement Tools with Error Handling
execute_tool dispatcher function. Instead of having your agentic loop know the internals of every tool, it just calls execute_tool(name, inputs) and gets back a result. This keeps the loop clean: it doesn't care whether it's calling a web scraper or a database — it just passes a name and inputs and gets JSON back. Adding a new tool later means writing one function and adding one entry to the dictionary. No changes to the loop.
# Mock implementations (replace with real APIs in production)
def web_search(query: str) -> dict:
time.sleep(0.2) # Simulate latency
return {"results": [
{"title": f"Result 1: {query}", "url": "https://example.com/1",
"snippet": f"Overview of {query}..."},
{"title": f"Result 2: {query}", "url": "https://example.com/2",
"snippet": f"Developments in {query}..."},
{"title": f"Result 3: {query}", "url": "https://broken.example.com/404",
"snippet": f"Deep dive into {query}..."},
]}
def fetch_page(url: str) -> dict:
time.sleep(0.3)
if "broken" in url or "404" in url:
raise ConnectionError(f"404 Not Found: {url}")
return {"content": f"Full page content from {url}. " * 20}
def summarize_text(text: str, max_points: int = 5) -> dict:
return {"summary": [f"Key point {i+1}" for i in range(min(max_points, 5))]}
def format_citation(title: str, url: str, accessed_date: str = None) -> dict:
date = accessed_date or "2025-01-15"
return {"citation": f'"{title}." Available at: {url}. Accessed: {date}.'}
def save_to_file(filename: str, content: str) -> dict:
return {"status": "saved", "path": f"/output/{filename}", "bytes": len(content)}
# Dispatcher with per-tool error handling
tool_functions = {
"web_search": web_search, "fetch_page": fetch_page,
"summarize_text": summarize_text, "format_citation": format_citation,
"save_to_file": save_to_file,
}
def execute_tool(name: str, inputs: dict) -> tuple[str, bool]:
"""Execute a tool, returning (result_json, is_error)."""
func = tool_functions.get(name)
if not func:
return json.dumps({"error": f"Unknown tool: {name}"}), True
try:
result = func(**inputs)
return json.dumps(result), False
except Exception as e:
return json.dumps({"error": str(e)}), True// Mock implementations
async function webSearch(query) {
await new Promise(r => setTimeout(r, 200));
return { results: [
{ title: `Result 1: ${query}`, url: "https://example.com/1",
snippet: `Overview of ${query}...` },
{ title: `Result 2: ${query}`, url: "https://example.com/2",
snippet: `Developments in ${query}...` },
{ title: `Result 3: ${query}`, url: "https://broken.example.com/404",
snippet: `Deep dive into ${query}...` },
]};
}
async function fetchPage(url) {
await new Promise(r => setTimeout(r, 300));
if (url.includes("broken") || url.includes("404"))
throw new Error(`404 Not Found: ${url}`);
return { content: `Full page content from ${url}. `.repeat(20) };
}
function summarizeText(text, maxPoints = 5) {
return { summary: Array.from({ length: Math.min(maxPoints, 5) },
(_, i) => `Key point ${i + 1}`) };
}
function formatCitation(title, url, accessedDate) {
const date = accessedDate || "2025-01-15";
return { citation: `"${title}." Available at: ${url}. Accessed: ${date}.` };
}
function saveToFile(filename, content) {
return { status: "saved", path: `/output/${filename}`, bytes: content.length };
}
const toolFunctions = {
web_search: (i) => webSearch(i.query),
fetch_page: (i) => fetchPage(i.url),
summarize_text: (i) => summarizeText(i.text, i.max_points),
format_citation: (i) => formatCitation(i.title, i.url, i.accessed_date),
save_to_file: (i) => saveToFile(i.filename, i.content),
};
async function executeTool(name, inputs) {
const func = toolFunctions[name];
if (!func)
return { result: JSON.stringify({ error: `Unknown tool: ${name}` }), isError: true };
try {
const result = await func(inputs);
return { result: JSON.stringify(result), isError: false };
} catch (e) {
return { result: JSON.stringify({ error: e.message }), isError: true };
}
}Step 3: The Agentic Loop with Parallel Execution
finish_reason: "stop"). With the OpenAI-compatible API, tool calls come back in response.choices[0].message.tool_calls. Each call has .function.name and arguments parsed via json.loads(tc.function.arguments). Tool results go back as {"role": "tool", "tool_call_id": tc.id, "content": result_str} messages. The critical safety measure is the max_iterations limit — without it, a confused model could loop indefinitely, burning tokens.
def run_research_agent(question: str, available_tools=None) -> str:
"""Run the agentic loop with parallel tool execution."""
active_tools = available_tools or tools
messages = [
{"role": "system", "content": "You are a research assistant. Search multiple sources in parallel when possible."},
{"role": "user", "content": question}
]
max_iterations = 10 # Safety limit
for iteration in range(max_iterations):
try:
response = client.chat.completions.create(
model="mistral",
tools=active_tools,
messages=messages,
)
except Exception as e:
return f"API error: {e}"
choice = response.choices[0]
tool_calls = choice.message.tool_calls or []
if choice.finish_reason == "stop" or not tool_calls:
# Model is done — extract final text
return choice.message.content or ""
# Append assistant message with tool_calls
messages.append({
"role": "assistant",
"content": choice.message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in tool_calls
]
})
# Execute tools — parallel when multiple requested
if len(tool_calls) > 1:
# PARALLEL: use ThreadPoolExecutor
tool_results = []
with ThreadPoolExecutor(max_workers=len(tool_calls)) as pool:
futures = {
pool.submit(execute_tool, tc.function.name, json.loads(tc.function.arguments)): tc.id
for tc in tool_calls
}
for future in as_completed(futures):
tid = futures[future]
result_json, is_err = future.result()
tool_results.append({
"role": "tool",
"tool_call_id": tid,
"content": result_json,
})
else:
# SEQUENTIAL: single tool
tc = tool_calls[0]
result_json, is_err = execute_tool(tc.function.name, json.loads(tc.function.arguments))
tool_results = [{
"role": "tool",
"tool_call_id": tc.id,
"content": result_json,
}]
messages.extend(tool_results)
return "Max iterations reached."
# Run it
answer = run_research_agent(
"Research the latest developments in AI agents. "
"Search multiple sources and summarize findings."
)
print(answer)async function runResearchAgent(question, availableTools) {
const activeTools = availableTools || tools;
const messages = [
{ role: "system", content: "You are a research assistant. Search multiple sources in parallel when possible." },
{ role: "user", content: question }
];
const maxIterations = 10;
for (let i = 0; i < maxIterations; i++) {
let response;
try {
response = await client.chat.completions.create({
model: 'mistral',
tools: activeTools,
messages,
});
} catch (e) {
return `API error: ${e.message}`;
}
const choice = response.choices[0];
const toolCalls = choice.message.toolCalls || [];
if (choice.finish_reason === "stop" || toolCalls.length === 0) {
return choice.message.content || "";
}
// Append assistant message with tool_calls
messages.push({
role: "assistant",
content: choice.message.content,
tool_calls: toolCalls.map(tc => ({
id: tc.id,
type: "function",
function: { name: tc.function.name, arguments: tc.function.arguments }
}))
});
let toolResults;
if (toolCalls.length > 1) {
// PARALLEL: Promise.all
toolResults = await Promise.all(
toolCalls.map(async (tc) => {
const { result } = await executeTool(tc.function.name, JSON.parse(tc.function.arguments));
return {
role: "tool",
tool_call_id: tc.id,
content: result,
};
})
);
} else {
// SEQUENTIAL: single tool
const tc = toolCalls[0];
const { result } = await executeTool(tc.function.name, JSON.parse(tc.function.arguments));
toolResults = [{
role: "tool",
tool_call_id: tc.id,
content: result,
}];
}
messages.push(...toolResults);
}
return "Max iterations reached.";
}
const answer = await runResearchAgent(
"Research the latest developments in AI agents. " +
"Search multiple sources and summarize findings."
);
console.log(answer);choices[0].message.tool_calls, your code runs them concurrently using ThreadPoolExecutor (Python) or Promise.all (Node.js). When it returns a single tool call, it runs sequentially. Tool results go back as role: "tool" messages keyed by tool_call_id. The loop repeats until finish_reason is "stop".
Step 4: Dynamic Tool Registry
ToolRegistry class stores tools with category tags and filters them on demand. The payoff is simple: instead of sending all 20 tools with every API call, you call get_tools_for_context(tags=["research"]) and get back only the 3 tools relevant to the current phase. One small trap to watch for: tag names are case-sensitive. Pick a convention (lowercase recommended) and stick with it.
class ToolRegistry:
"""Manages tools and filters them by context."""
def __init__(self):
self._tools: dict[str, dict] = {}
self._tags: dict[str, set[str]] = {}
def register(self, tool: dict, tags: list[str] = None):
# tool is the full {"type":"function","function":{...}} object
name = tool["function"]["name"]
self._tools[name] = tool
self._tags[name] = set(tags or [])
def unregister(self, name: str):
self._tools.pop(name, None)
self._tags.pop(name, None)
def get_tools_for_context(
self, tags: list[str] = None, names: list[str] = None
) -> list[dict]:
if names:
return [self._tools[n] for n in names if n in self._tools]
if tags:
tag_set = set(tags)
return [
self._tools[n] for n, t in self._tags.items()
if t & tag_set
]
return list(self._tools.values())
# Usage
registry = ToolRegistry()
registry.register(tools[0], tags=["research", "search"])
registry.register(tools[1], tags=["research", "fetch"])
registry.register(tools[2], tags=["research", "analysis"])
registry.register(tools[3], tags=["citation"])
registry.register(tools[4], tags=["output"])
# Phase 1: only research tools
research_tools = registry.get_tools_for_context(tags=["research"])
# => [web_search, fetch_page, summarize_text]
# Phase 2: add citation tools after summaries are ready
cite_tools = registry.get_tools_for_context(
names=["format_citation", "save_to_file"]
)class ToolRegistry {
constructor() {
this._tools = new Map();
this._tags = new Map();
}
register(tool, tags = []) {
const name = tool.function.name;
this._tools.set(name, tool);
this._tags.set(name, new Set(tags));
}
unregister(name) {
this._tools.delete(name);
this._tags.delete(name);
}
getToolsForContext({ tags, names } = {}) {
if (names)
return names.filter(n => this._tools.has(n)).map(n => this._tools.get(n));
if (tags) {
const tagSet = new Set(tags);
const result = [];
for (const [name, toolTags] of this._tags) {
for (const t of tagSet) {
if (toolTags.has(t)) { result.push(this._tools.get(name)); break; }
}
}
return result;
}
return [...this._tools.values()];
}
}
// Usage
const registry = new ToolRegistry();
registry.register(tools[0], ["research", "search"]);
registry.register(tools[1], ["research", "fetch"]);
registry.register(tools[2], ["research", "analysis"]);
registry.register(tools[3], ["citation"]);
registry.register(tools[4], ["output"]);
const researchTools = registry.getToolsForContext({ tags: ["research"] });
const citeTools = registry.getToolsForContext({
names: ["format_citation", "save_to_file"]
});ToolRegistry that tags tools by category ("research", "citation", "output") and filters them on demand. Note that the registry now uses tool["function"]["name"] to extract the tool name, matching the OpenAI tool format. In a real workflow, you'd call get_tools_for_context(tags=["research"]) during the search phase to give the model only 3 tools, then switch to names=["format_citation", "save_to_file"] when summaries are ready.
Hands-On Exercise
What You'll Build
A multi-tool research agent that searches multiple sources in parallel, fetches pages sequentially, handles errors gracefully, and uses a ToolRegistry to filter tools by context. Runs against a local Ollama server with Mistral.
Time Estimate: 30–45 minutes
Prerequisites: Python 3.10+ (or Node.js 18+), Ollama installed and running (ollama run mistral), and completion of M05.
Files You'll Create: multi_tool_agent.py (or multi_tool_agent.mjs for Node.js)
Environment Setup
mkdir multi-tool-lab && cd multi-tool-lab
python -m venv venv && source venv/bin/activate # Windows: venv\Scripts\activate
pip install openai
# Make sure Ollama is running: ollama run mistralmkdir multi-tool-lab && cd multi-tool-lab
npm init -y && npm install openai
# Make sure Ollama is running: ollama run mistralStep 1: Define Tools, Mock Implementations & ToolRegistry
What: This step sets up everything the agent needs: 5 tool schemas in OpenAI format, mock functions behind each schema, an execute_tool dispatcher with error handling, and a ToolRegistry for filtering tools by context.
Why: Separating tool definitions (what the model sees) from tool implementations (what your code runs) is a fundamental pattern. The ToolRegistry adds the ability to dynamically filter which tools are sent to the model based on context.
Create a new file called multi_tool_agent.py and add the following:
from openai import OpenAI
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
# ── Tool Schemas (OpenAI format) ──────────────────────────────
tools = [
{
"type": "function",
"function": {
"name": "web_search",
"description": (
"Search the web for current information. Returns top 3 "
"results with title, URL, and snippet. Use for recent "
"events, factual questions, or general research."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "fetch_page",
"description": (
"Fetch the full text content of a web page by URL. "
"Returns page text (max 5000 chars). Use after "
"web_search to get full content from a result URL."
),
"parameters": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "Full URL to fetch"}
},
"required": ["url"]
}
}
},
{
"type": "function",
"function": {
"name": "summarize_text",
"description": (
"Summarize long text into key points (3-5 bullets). "
"Use after fetch_page to condense page content."
),
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to summarize"},
"max_points": {
"type": "integer",
"description": "Max bullet points (default 5)"
}
},
"required": ["text"]
}
}
},
{
"type": "function",
"function": {
"name": "format_citation",
"description": (
"Format a source as an academic citation. Use after "
"summaries are ready to create proper references."
),
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Article title"},
"url": {"type": "string", "description": "Source URL"},
"accessed_date": {"type": "string", "description": "e.g. '2025-01-15'"}
},
"required": ["title", "url"]
}
}
},
{
"type": "function",
"function": {
"name": "save_to_file",
"description": "Save content to a local file. Returns file path.",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "Output filename"},
"content": {"type": "string", "description": "Content to save"}
},
"required": ["filename", "content"]
}
}
}
]
# ── Mock Implementations ─────────────────────────────────────
def web_search(query: str) -> dict:
time.sleep(0.2)
return {"results": [
{"title": f"Result 1: {query}", "url": "https://example.com/1",
"snippet": f"Overview of {query}..."},
{"title": f"Result 2: {query}", "url": "https://example.com/2",
"snippet": f"Developments in {query}..."},
{"title": f"Result 3: {query}", "url": "https://broken.example.com/404",
"snippet": f"Deep dive into {query}..."},
]}
def fetch_page(url: str) -> dict:
time.sleep(0.3)
if "broken" in url or "404" in url:
raise ConnectionError(f"404 Not Found: {url}")
return {"content": f"Full page content from {url}. " * 20}
def summarize_text(text: str, max_points: int = 5) -> dict:
return {"summary": [f"Key point {i+1}" for i in range(min(max_points, 5))]}
def format_citation(title: str, url: str, accessed_date: str = None) -> dict:
date = accessed_date or "2025-01-15"
return {"citation": f'"{title}." Available at: {url}. Accessed: {date}.'}
def save_to_file(filename: str, content: str) -> dict:
return {"status": "saved", "path": f"/output/{filename}", "bytes": len(content)}
# ── Dispatcher with Error Handling ───────────────────────────
tool_functions = {
"web_search": web_search, "fetch_page": fetch_page,
"summarize_text": summarize_text, "format_citation": format_citation,
"save_to_file": save_to_file,
}
def execute_tool(name: str, inputs: dict) -> tuple[str, bool]:
"""Execute a tool, returning (result_json, is_error)."""
func = tool_functions.get(name)
if not func:
return json.dumps({"error": f"Unknown tool: {name}"}), True
try:
result = func(**inputs)
return json.dumps(result), False
except Exception as e:
return json.dumps({"error": str(e), "tool": name}), True
# ── ToolRegistry ─────────────────────────────────────────────
class ToolRegistry:
def __init__(self):
self._tools: dict[str, dict] = {}
self._tags: dict[str, set[str]] = {}
def register(self, tool: dict, tags: list[str] = None):
name = tool["function"]["name"]
self._tools[name] = tool
self._tags[name] = set(tags or [])
def unregister(self, name: str):
self._tools.pop(name, None)
self._tags.pop(name, None)
def get_tools_for_context(self, tags: list[str] = None, names: list[str] = None) -> list[dict]:
if names:
return [self._tools[n] for n in names if n in self._tools]
if tags:
tag_set = set(tags)
return [self._tools[n] for n, t in self._tags.items() if t & tag_set]
return list(self._tools.values())
# Register tools with tags
registry = ToolRegistry()
registry.register(tools[0], tags=["research", "search"])
registry.register(tools[1], tags=["research", "fetch"])
registry.register(tools[2], tags=["research", "analysis"])
registry.register(tools[3], tags=["citation"])
registry.register(tools[4], tags=["output"])
print("✓ Tools, dispatcher, and registry ready.")
print(f" All tools: {[t['function']['name'] for t in registry.get_tools_for_context()]}")
print(f" Research only: {[t['function']['name'] for t in registry.get_tools_for_context(tags=['research'])]}")Run it:
ModuleNotFoundError: No module named 'openai'→ Runpip install openai- Connection refused → Make sure Ollama is running:
ollama serveandollama run mistral - Tag filter returns nothing unexpectedly → Tag names are case-sensitive.
"Research"and"research"are different.
Step 2: Add the Agentic Loop with Parallel Execution
What: This step adds the orchestration engine — the loop that sends messages to the model, executes tool calls (in parallel when multiple are returned), feeds results back, and repeats until the model is done.
Add the following to the bottom of multi_tool_agent.py (after the registry setup):
# ── Agentic Loop with Parallel Execution ─────────────────────
def run_agent(question: str, tool_tags: list[str] = None, verbose: bool = True) -> str:
"""Run the multi-tool agent. Optionally filter tools by tag."""
if tool_tags:
active_tools = registry.get_tools_for_context(tags=tool_tags)
else:
active_tools = registry.get_tools_for_context()
if verbose:
print(f"\n{'='*60}")
print(f"Question: {question}")
print(f"Active tools: {[t['function']['name'] for t in active_tools]}")
print(f"{'='*60}")
messages = [
{"role": "system", "content": "You are a research assistant. When asked to compare or research multiple topics, search for each one in parallel. When asked to fetch and summarize a page, do it sequentially."},
{"role": "user", "content": question}
]
max_iterations = 10
for iteration in range(max_iterations):
response = client.chat.completions.create(
model="mistral",
tools=active_tools,
messages=messages,
)
choice = response.choices[0]
tool_calls = choice.message.tool_calls or []
if choice.finish_reason == "stop" or not tool_calls:
final_text = choice.message.content or ""
if verbose:
print(f"\n✓ Agent finished in {iteration + 1} iteration(s)")
return final_text
# Append assistant message
messages.append({
"role": "assistant",
"content": choice.message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in tool_calls
]
})
# Show what model requested
if verbose:
mode = "PARALLEL" if len(tool_calls) > 1 else "SEQUENTIAL"
print(f"\n Iteration {iteration + 1} [{mode}]:")
for tc in tool_calls:
print(f" → {tc.function.name}({tc.function.arguments[:80]}...)")
# Execute tools — parallel when multiple
if len(tool_calls) > 1:
tool_results = []
with ThreadPoolExecutor(max_workers=len(tool_calls)) as pool:
futures = {
pool.submit(execute_tool, tc.function.name, json.loads(tc.function.arguments)): tc.id
for tc in tool_calls
}
for future in as_completed(futures):
tid = futures[future]
result_json, is_err = future.result()
if verbose and is_err:
print(f" ✗ Error for {tid}: {result_json[:60]}")
tool_results.append({
"role": "tool",
"tool_call_id": tid,
"content": result_json,
})
else:
tc = tool_calls[0]
result_json, is_err = execute_tool(tc.function.name, json.loads(tc.function.arguments))
if verbose and is_err:
print(f" ✗ Error: {result_json[:60]}")
tool_results = [{
"role": "tool",
"tool_call_id": tc.id,
"content": result_json,
}]
messages.extend(tool_results)
return "Max iterations reached."
# ── Test Scenarios ───────────────────────────────────────────
if __name__ == "__main__":
# Test 1: Parallel search
print("\n" + "▶ TEST 1: PARALLEL SEARCH ".ljust(60, "─"))
result1 = run_agent(
"Search for information about these 3 topics: AI agents, "
"prompt engineering, and tool use patterns.",
tool_tags=["research"]
)
print(f"\nResult preview: {result1[:200]}...")
# Test 2: Sequential chain (search → fetch → summarize)
print("\n" + "▶ TEST 2: SEQUENTIAL CHAIN ".ljust(60, "─"))
result2 = run_agent(
"Search for 'Mistral AI tool use', then fetch the first "
"result page and summarize its content.",
tool_tags=["research"]
)
print(f"\nResult preview: {result2[:200]}...")
# Test 3: Error recovery (fetch_page will 404 on broken URL)
print("\n" + "▶ TEST 3: ERROR RECOVERY ".ljust(60, "─"))
result3 = run_agent(
"Fetch and summarize this page: https://broken.example.com/404",
tool_tags=["research"]
)
print(f"\nResult preview: {result3[:200]}...")
# Test 4: Dynamic tool filtering (citation tools only)
print("\n" + "▶ TEST 4: DYNAMIC TOOL FILTERING ".ljust(60, "─"))
result4 = run_agent(
"Format a citation for an article titled 'Multi-Tool AI Agents' "
"from https://example.com/agents, accessed today.",
tool_tags=["citation"]
)
print(f"\nResult preview: {result4[:200]}...")Run the full agent:
Look for these key behaviors in your output:
- Test 1: Should show
[PARALLEL]with 3web_searchcalls in one iteration - Test 2: Should show
[SEQUENTIAL]across 3–4 iterations - Test 3: Should show
✗ Errorfollowed by the model adapting - Test 4: Should show
Active tools: ['format_citation']— only 1 tool
- Model doesn't call tools at all → Mistral's tool calling support varies by version. Ensure you have
mistralormistral-nemopulled in Ollama. - Connection error → Make sure Ollama is running on port 11434:
curl http://localhost:11434/v1/models - Test 1 shows SEQUENTIAL instead of PARALLEL → Local models may not always parallelize. Try rephrasing: "Search for these 3 topics simultaneously: ..."
Verify Everything Works
Run the complete file end-to-end. All 4 tests should complete without crashing, demonstrating parallel execution, sequential chaining, error recovery, and dynamic tool filtering against your local Ollama instance.
You've built a production-pattern multi-tool agent running entirely locally with Ollama/Mistral! You can extend this by swapping mock implementations for real APIs, adding a circuit breaker counter, or implementing execution timing to compare parallel vs sequential wall-clock times.
- Add execution timing to each tool call and print a trace waterfall showing parallel vs sequential sections
- Implement a circuit breaker class that disables a tool after 3 consecutive failures
- Add a cost tracker that estimates token usage per iteration based on message length
Knowledge Check
Q1: Given three tool calls where B needs A's result, but C is independent of both, what's the optimal execution strategy?
Q2: Rank these tool descriptions from LEAST to MOST effective: (1) "queries data" | (2) "Run a SQL query against the users DB. Returns matching rows. Use when asking about user accounts." | (3) "database tool"
Q3: A tool fails with a network timeout. What's the BEST way to report this to the model?
Q4: You have 20 tools averaging 400 input tokens each. Filtering to 5 per request saves approximately how many tokens?
Q5: In a 3-tool sequential chain (search → fetch → summarize), how many API round trips before the final text response?
Q6: The model returns 3 tool calls in one response. How should you return the results?
Module Summary
Key Concepts Recap
- Parallel tool calls: Multiple independent tools in one response. Execute concurrently, return all results in one message.
- Sequential chains: Output of one feeds the next. Each step is a full API round trip via the agentic loop.
- Tool selection: Description quality directly determines accuracy. Include what, when, and returns.
- Dynamic registration: Filter tools by context to save tokens, improve accuracy, enforce least privilege.
- Error handling: Return structured error JSON. Let the model adapt. Use circuit breakers for persistent failures.
- OpenAI tool format: Tools are
{"type":"function","function":{...}}withparameters(notinput_schema). Tool results userole:"tool"withtool_call_id.
Next: M08 — Conversation Management
You've built agents that call multiple tools across multiple turns. As conversations grow, the entire message history is resent on every API call — which gets expensive fast. M08 teaches you how to manage context window budgets, implement sliding windows, and use summarization to keep costs under control for long-running agents.