Building a langchain multi-tool agent example that actually works in production requires more than chaining a few tools together. You need reliable tool definitions, proper error handling, and a clear execution loop that doesn’t spiral into infinite retries. This guide walks through assembling an agent with three complementary tools — web search for current information, a calculator for precise arithmetic, and a Python REPL for data manipulation — using LangChain’s modern tool-calling abstractions.
The agent we’ll build follows the ReAct pattern: reason, act, observe. Each step the model decides which tool to invoke (or none), executes it, incorporates the result, and repeats until it can answer. We’ll use OpenAI’s function-calling models since they handle structured tool calls natively, but the same structure works with any provider that supports tool calling.
Step 1: Set up the environment and dependencies
Create a fresh virtual environment and install the minimal set of packages. We need LangChain core, the OpenAI integration, a search provider, and a safe code execution sandbox.
python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "langchain>=0.2.0" "langchain-openai>=0.1.0" "langchain-community>=0.2.0" \
"duckduckgo-search>=6.0" "python-dotenv>=1.0"
Create a .env file with your API key. If you route through a gateway like n4n.ai, set the base URL there instead of pointing directly at OpenAI.
# .env
OPENAI_API_KEY=sk-...
# OPENAI_API_BASE=https://api.n4n.ai/v1 # optional gateway
Verify the install works:
# test_imports.py
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
print("Imports OK")
Run it: python test_imports.py. You should see “Imports OK” with no errors.
Step 2: Define the calculator tool
LangChain ships with a LLMMathChain but it adds an extra LLM call for simple arithmetic. A direct Python eval with a restricted namespace is faster, deterministic, and easier to audit. Wrap it as a Tool so the agent sees a clean schema.
# tools/calculator.py
import math
import re
from langchain.tools import Tool
_SAFE_NAMES = {
k: v for k, v in math.__dict__.items() if not k.startswith("_")
}
_SAFE_NAMES.update({"abs": abs, "round": round, "min": min, "max": max, "sum": sum})
def _eval_expression(expr: str) -> str:
# Allow only digits, operators, parentheses, dots, and safe function names
if not re.fullmatch(r"[\d\s\+\-\*\/\.\,\%\(\)a-zA-Z_]+", expr):
return "Error: invalid characters in expression"
try:
result = eval(expr, {"__builtins__": {}}, _SAFE_NAMES)
return str(result)
except Exception as e:
return f"Error: {e}"
calculator_tool = Tool(
name="calculator",
description=(
"Evaluate a mathematical expression. Input must be a single valid Python "
"expression using basic operators (+, -, *, /, %, **) and functions from "
"the math module (sin, cos, sqrt, log, pi, e, etc.). No assignments, "
"imports, or statements allowed."
),
func=_eval_expression,
)
Test it in isolation:
# test_calculator.py
from tools.calculator import calculator_tool
print(calculator_tool.invoke({"expr": "sqrt(2) * 10"}))
print(calculator_tool.invoke({"expr": "log(1000, 10)"}))
print(calculator_tool.invoke({"expr": "__import__('os').system('ls')"}) # blocked
Expected output: 14.142135623730951, 3.0, Error: invalid characters in expression.
Step 3: Configure the search tool
DuckDuckGo’s HTML scrape works without an API key and returns concise snippets. For production workloads you’d swap this for a proper search API (SerpAPI, Tavily, Bing) with rate limits and structured results. The community wrapper handles the parsing.
# tools/search.py
from langchain_community.tools import DuckDuckGoSearchRun
search_tool = DuckDuckGoSearchRun(
name="web_search",
description=(
"Search the web for current information. Use this when you need facts, "
"prices, news, or data not in your training knowledge. Input a concise "
"query string. Returns the top result snippets."
),
)
Quick verification:
# test_search.py
from tools.search import search_tool
print(search_tool.invoke({"query": "current price of gold per ounce USD"}))
You should see a few sentences with a recent price and source attribution.
Step 4: Build a sandboxed Python REPL tool
The agent needs to manipulate data — filter JSON, compute statistics, transform structures. A full Jupyter kernel is overkill and unsafe. Use subprocess with a tight timeout, no network, and a restricted filesystem view. This runs in the same process but isolates each execution.
# tools/python_repl.py
import subprocess
import sys
import tempfile
import os
from langchain.tools import Tool
_TIMEOUT_SECONDS = 10
_MAX_OUTPUT_CHARS = 4000
def _run_python(code: str) -> str:
# Write to a temp file to avoid shell injection and allow multi-line code
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(code)
tmp_path = f.name
try:
proc = subprocess.run(
[sys.executable, tmp_path],
capture_output=True,
text=True,
timeout=_TIMEOUT_SECONDS,
env={**os.environ, "PYTHONPATH": ""}, # no access to project modules
)
out = proc.stdout
err = proc.stderr
if proc.returncode != 0:
return f"Execution failed (exit {proc.returncode}):\n{err}"
if not out and not err:
return "(no output)"
return out[:_MAX_OUTPUT_CHARS]
except subprocess.TimeoutExpired:
return f"Error: execution timed out after {_TIMEOUT_SECONDS}s"
finally:
try:
os.unlink(tmp_path)
except OSError:
pass
python_repl_tool = Tool(
name="python_repl",
description=(
"Execute a Python script in a sandboxed subprocess. Use for data "
"processing, calculations too complex for the calculator, or transforming "
"search results. The environment has stdlib only — no third-party packages, "
"no network, no filesystem access beyond the script itself. Print results "
"to stdout. Timeout is 10 seconds."
),
func=_run_python,
)
Test:
# test_repl.py
from tools.python_repl import python_repl_tool
print(python_repl_tool.invoke({
"code": "import json; data=[1,2,3,4,5]; print(sum(x**2 for x in data))"
}))
print(python_repl_tool.invoke({"code": "import requests; print('oops')"}) # fails
First call prints 55. Second call fails with ModuleNotFoundError: No module named 'requests'.
Step 5: Assemble the prompt and agent
LangChain’s create_tool_calling_agent expects a prompt with an agent_scratchpad placeholder. The system message should explain the toolset and encourage concise tool calls. Keep the prompt tight — every token costs latency and money.
# agent/prompt.py
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
SYSTEM_PROMPT = """You are a helpful assistant with access to three tools:
1. calculator — evaluate math expressions (basic ops, math module functions).
2. web_search — query the web for current information.
3. python_repl — run Python code for data manipulation, stats, or complex logic.
Guidelines:
- Prefer calculator for simple arithmetic; use python_repl for multi-step computation.
- Use web_search when you lack current facts. Cite sources from results.
- Think step by step. Call one tool at a time. Wait for the result before deciding next.
- If a tool errors, adjust and retry once. Do not loop indefinitely.
- When you have enough information, answer directly without invoking tools.
"""
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
Step 6: Wire the agent executor
Instantiate the model, bind tools, create the agent, and wrap in an AgentExecutor with sane defaults. Set max_iterations to prevent runaway loops and handle_parsing_errors to recover from malformed tool calls.
# agent/executor.py
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from agent.prompt import prompt
from tools.calculator import calculator_tool
from tools.search import search_tool
from tools.python_repl import python_repl_tool
def build_agent_executor(model_name: str = "gpt-4o-mini") -> AgentExecutor:
llm = ChatOpenAI(
model=model_name,
temperature=0,
# If using a gateway, the base_url is picked up from OPENAI_API_BASE env var
)
tools = [calculator_tool, search_tool, python_repl_tool]
agent = create_tool_calling_agent(llm, tools, prompt)
return AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=8,
handle_parsing_errors=True,
return_intermediate_steps=True,
)
Step 7: Run end-to-end queries
Create a small CLI to test the full loop. The executor returns a dict with output and intermediate_steps — useful for debugging and logging.
# main.py
import sys
from agent.executor import build_agent_executor
def main():
if len(sys.argv) < 2:
print("Usage: python main.py \"your question here\"")
sys.exit(1)
query = " ".join(sys.argv[1:])
executor = build_agent_executor()
result = executor.invoke({"input": query})
print("\n=== FINAL ANSWER ===")
print(result["output"])
print("\n=== STEPS ===")
for step in result["intermediate_steps"]:
action, observation = step
print(f"Tool: {action.tool}")
print(f"Input: {action.tool_input}")
print(f"Output: {observation[:200]}...")
print("---")
if __name__ == "__main__":
main()
Run a few verification queries:
# 1. Pure calculation
python main.py "What is the monthly payment on a $350,000 mortgage at 6.75% for 30 years?"
# 2. Search + calculation
python main.py "Find the current US federal funds rate and compute the effective annual rate if compounded monthly."
# 3. Search + Python data processing
python main.py "Search for the 2023 GDP of the top 5 economies, then compute their combined share of world GDP (assume world GDP $105T)."
# 4. Multi-step reasoning
python main.py "A rectangle's area is 150 sq ft. Its length is 5 ft more than its width. Find the dimensions."
Expected behavior
Query 1 — The agent calls calculator with the mortgage formula: P * r * (1+r)^n / ((1+r)^n - 1) where r = 0.0675/12, n = 360. Returns ~$2,270.
Query 2 — Agent calls web_search for “federal funds rate 2024”, extracts ~5.33%, then calls calculator with (1 + 0.0533/12)^12 - 1 → ~5.47%.
Query 3 — Agent searches for each GDP, then uses python_repl to parse numbers, sum top-5, divide by 105T, format as percentage.
Query 4 — Agent sets up quadratic: w * (w + 5) = 150, solves via calculator or python_repl → width 10 ft, length 15 ft.
If any query stalls or loops, check verbose=True output. The most common failure modes:
- Search returns noisy snippets → refine the query in the system prompt.
- Python REPL times out → the code has an infinite loop; add a hard iteration limit inside the sandbox.
- Model picks wrong tool → sharpen tool descriptions with negative examples (“do not use calculator for multi-step logic”).
Step 8: Add structured logging and observability
Production agents need traces. LangChain’s callbacks hook into every LLM call, tool invocation, and agent step. Here’s a minimal JSONL logger you can ship to a log aggregator.
# observability/logger.py
import json
import time
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, Optional
from uuid import uuid4
class JsonlCallbackHandler(BaseCallbackHandler):
def __init__(self, filepath: str = "agent_traces.jsonl"):
self.filepath = filepath
self.run_id = str(uuid4())
self.step = 0
def _write(self, event: Dict[str, Any]):
event["run_id"] = self.run_id
event["timestamp"] = time.time()
with open(self.filepath, "a") as f:
f.write(json.dumps(event) + "\n")
def on_llm_start(self, serialized: Dict, prompts: list, **kwargs):
self._write({"event": "llm_start", "prompts": prompts})
def on_llm_end(self, response, **kwargs):
self._write({"event": "llm_end", "usage": getattr(response, "usage", None)})
def on_tool_start(self, serialized: Dict, input_str: str, **kwargs):
self.step += 1
self._write({"event": "tool_start", "step": self.step, "tool": serialized.get("name"), "input": input_str})
def on_tool_end(self, output: str, **kwargs):
self._write({"event": "tool_end", "step": self.step, "output": output[:500]})
def on_tool_error(self, error: Exception, **kwargs):
self._write({"event": "tool_error", "step": self.step, "error": str(error)})
Attach it when invoking:
# main.py (updated)
from observability.logger import JsonlCallbackHandler
callbacks = [JsonlCallbackHandler()]
result = executor.invoke({"input": query}, config={"callbacks": callbacks})
Each run produces agent_traces.jsonl with ordered events. You can replay, compute latency percentiles, or feed into tool calls fail, and detect prompt regressions.
Step 9: Harden for deployment
Before exposing this as an API, address three gaps:
1. Input validation — Wrap the entry point with Pydantic to reject oversized or malicious payloads.
# api/schemas.py
from pydantic import BaseModel, Field
class AgentRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=2000)
session_id: str | None = None
2. Concurrency control — The Python REPL subprocess is CPU-bound. Run the executor in a thread pool with a semaphore, or offload to a worker queue (Celery, RQ) with a dedicated sandbox container.
3. Provider fallback — If your primary model provider hits rate limits or degrades, swap the ChatOpenAI instance for a fallback model. A gateway that handles this transparently (honoring x-routing-directive headers and forwarding provider cache-control hints) avoids building custom retry logic.
# agent/executor.py (fallback pattern)
from langchain_openai import ChatOpenAI
def get_llm(primary: str = "gpt-4o-mini", fallback: str = "gpt-3.5-turbo") -> ChatOpenAI:
# In practice, wrap with a retry/fallback policy or use a gateway that does this
return ChatOpenAI(model=primary, temperature=0)
Verification checklist
Run through this list before considering the agent done:
- All three tools respond correctly in isolation (Steps 2–4 tests pass).
- Agent answers Query 1–4 with correct results and reasonable latency (<15s total).
-
verbose=Trueshows clean ReAct loops: thought → tool call → observation → repeat. - No infinite loops on ambiguous queries (max_iterations caps at 8).
- JSONL traces capture every LLM call, tool input, and tool output.
- Error injection (bad search query, syntax error in Python) returns a graceful message, not a stack trace.
- Memory footprint stays flat across 50+ consecutive invocations (no conversation history accumulation unless you add it).
Extending the toolkit
The three-tool foundation covers a wide range of analyst-style tasks. Common additions:
| Tool | Use case | Implementation hint |
|---|---|---|
| SQL executor | Query internal databases | Use read-only connection, parameterized queries, row limit |
| File reader | Ingest PDFs, CSVs from a bucket | Presigned URLs, streaming parse, size caps |
| HTTP client | Call internal APIs | Allowlist hosts, timeout, retry policy |
| Chart generator | Plotly/Altair to base64 PNG | Run in python_repl, return data URI |
Each new tool follows the same pattern: pure function → Tool wrapper → description that tells the model when and when not to use it. Keep the tool count under 10; beyond that, tool selection accuracy drops and you need a router or hierarchical agent.
You now have a working langchain multi-tool agent example that searches, calculates, and runs code — observable, sandboxed, and ready to wrap in an API. The pattern scales: add tools, tighten prompts, swap models, and route through a gateway for resilience. The core loop stays the same.