Introduction: LLM agents extend language models beyond text generation into autonomous action. By connecting LLMs to tools—web search, code execution, APIs, databases—agents can gather information, perform calculations, and interact with external systems. This guide covers building tool-using agents from scratch: defining tools with schemas, implementing the reasoning loop, handling tool execution, managing conversation state, and building production-ready agent systems. These patterns form the foundation for building AI assistants that can actually do things, not just talk about them.
Defining Tools
from dataclasses import dataclass
from typing import Callable, Any
import json
import inspect
@dataclass
class Tool:
"""A tool that an agent can use."""
name: str
description: str
parameters: dict # JSON Schema
function: Callable[..., Any]
def to_openai_schema(self) -> dict:
"""Convert to OpenAI function calling format."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
def execute(self, **kwargs) -> Any:
"""Execute the tool with given arguments."""
return self.function(**kwargs)
def tool(name: str = None, description: str = None):
"""Decorator to create a tool from a function."""
def decorator(func: Callable) -> Tool:
tool_name = name or func.__name__
tool_desc = description or func.__doc__ or ""
# Extract parameters from type hints
sig = inspect.signature(func)
hints = func.__annotations__
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name == 'return':
continue
param_type = hints.get(param_name, str)
# Map Python types to JSON Schema
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
properties[param_name] = {
"type": type_map.get(param_type, "string"),
"description": f"Parameter: {param_name}"
}
if param.default == inspect.Parameter.empty:
required.append(param_name)
parameters = {
"type": "object",
"properties": properties,
"required": required
}
return Tool(
name=tool_name,
description=tool_desc,
parameters=parameters,
function=func
)
return decorator
# Define tools using decorator
@tool(name="calculator", description="Perform mathematical calculations")
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
# Safe evaluation of math expressions
allowed = set('0123456789+-*/.() ')
if not all(c in allowed for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression)
return str(result)
except Exception as e:
return f"Error: {e}"
@tool(name="web_search", description="Search the web for information")
def search_web(query: str, num_results: int = 5) -> str:
"""Search the web and return results."""
# In production, use actual search API
return f"Search results for '{query}': [simulated results]"
@tool(name="get_weather", description="Get current weather for a location")
def get_weather(location: str) -> str:
"""Get weather information."""
# In production, use weather API
return f"Weather in {location}: 72°F, Sunny"Basic Agent Loop
from openai import OpenAI
from typing import Optional
client = OpenAI()
class SimpleAgent:
"""Basic agent with tool-calling capability."""
def __init__(self, tools: list[Tool], model: str = "gpt-4o"):
self.tools = {t.name: t for t in tools}
self.model = model
self.messages = []
def _get_tool_schemas(self) -> list[dict]:
"""Get OpenAI-compatible tool schemas."""
return [t.to_openai_schema() for t in self.tools.values()]
def _execute_tool(self, name: str, arguments: dict) -> str:
"""Execute a tool and return result."""
if name not in self.tools:
return f"Error: Unknown tool '{name}'"
try:
result = self.tools[name].execute(**arguments)
return str(result)
except Exception as e:
return f"Error executing {name}: {e}"
def run(self, user_message: str, max_iterations: int = 10) -> str:
"""Run the agent loop."""
self.messages.append({"role": "user", "content": user_message})
for _ in range(max_iterations):
response = client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=self._get_tool_schemas(),
tool_choice="auto"
)
message = response.choices[0].message
self.messages.append(message)
# Check if we're done (no tool calls)
if not message.tool_calls:
return message.content
# Execute each tool call
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
result = self._execute_tool(name, args)
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
return "Max iterations reached"
# Usage
agent = SimpleAgent([calculate, search_web, get_weather])
response = agent.run("What's 15% of 847, and what's the weather in Seattle?")
print(response)ReAct Agent Pattern
from dataclasses import dataclass
from enum import Enum
class ActionType(str, Enum):
THINK = "think"
ACT = "act"
OBSERVE = "observe"
FINISH = "finish"
@dataclass
class AgentStep:
action_type: ActionType
content: str
tool_name: Optional[str] = None
tool_args: Optional[dict] = None
class ReActAgent:
"""Agent using Reasoning + Acting pattern."""
SYSTEM_PROMPT = """You are a helpful assistant that can use tools to answer questions.
For each step, you must:
1. THINK: Reason about what to do next
2. ACT: Choose a tool to use (or FINISH if done)
3. OBSERVE: See the result
Available tools:
{tools}
Format your response as:
THINK: [your reasoning]
ACT: [tool_name](arg1="value1", arg2="value2")
or
THINK: [your reasoning]
FINISH: [your final answer]"""
def __init__(self, tools: list[Tool], model: str = "gpt-4o"):
self.tools = {t.name: t for t in tools}
self.model = model
self.trace: list[AgentStep] = []
def _format_tools(self) -> str:
"""Format tool descriptions for prompt."""
lines = []
for tool in self.tools.values():
params = ", ".join(
f"{k}: {v['type']}"
for k, v in tool.parameters.get('properties', {}).items()
)
lines.append(f"- {tool.name}({params}): {tool.description}")
return "\n".join(lines)
def _parse_response(self, text: str) -> AgentStep:
"""Parse agent response into structured step."""
lines = text.strip().split('\n')
think_content = ""
for line in lines:
if line.startswith("THINK:"):
think_content = line[6:].strip()
elif line.startswith("ACT:"):
action = line[4:].strip()
# Parse tool call: tool_name(arg1="val1", ...)
if '(' in action:
name = action[:action.index('(')]
args_str = action[action.index('(')+1:action.rindex(')')]
# Simple arg parsing
args = {}
for part in args_str.split(','):
if '=' in part:
k, v = part.split('=', 1)
args[k.strip()] = v.strip().strip('"\'')
return AgentStep(
action_type=ActionType.ACT,
content=think_content,
tool_name=name,
tool_args=args
)
elif line.startswith("FINISH:"):
return AgentStep(
action_type=ActionType.FINISH,
content=line[7:].strip()
)
return AgentStep(action_type=ActionType.THINK, content=think_content)
def run(self, query: str, max_steps: int = 10) -> str:
"""Run the ReAct loop."""
system = self.SYSTEM_PROMPT.format(tools=self._format_tools())
messages = [
{"role": "system", "content": system},
{"role": "user", "content": query}
]
for step_num in range(max_steps):
response = client.chat.completions.create(
model=self.model,
messages=messages
)
text = response.choices[0].message.content
step = self._parse_response(text)
self.trace.append(step)
if step.action_type == ActionType.FINISH:
return step.content
if step.action_type == ActionType.ACT and step.tool_name:
# Execute tool
if step.tool_name in self.tools:
result = self.tools[step.tool_name].execute(**step.tool_args)
else:
result = f"Unknown tool: {step.tool_name}"
# Add observation
observation = f"OBSERVE: {result}"
self.trace.append(AgentStep(
action_type=ActionType.OBSERVE,
content=result
))
messages.append({"role": "assistant", "content": text})
messages.append({"role": "user", "content": observation})
return "Max steps reached without conclusion"
# Usage
react_agent = ReActAgent([calculate, search_web, get_weather])
result = react_agent.run("Calculate 20% tip on a $85 bill")
# View reasoning trace
for step in react_agent.trace:
print(f"{step.action_type.value}: {step.content}")Tool Registry
from typing import Dict, List, Optional
import importlib
import yaml
class ToolRegistry:
"""Central registry for agent tools."""
def __init__(self):
self._tools: Dict[str, Tool] = {}
self._categories: Dict[str, List[str]] = {}
def register(self, tool: Tool, category: str = "general"):
"""Register a tool."""
self._tools[tool.name] = tool
if category not in self._categories:
self._categories[category] = []
self._categories[category].append(tool.name)
def get(self, name: str) -> Optional[Tool]:
"""Get a tool by name."""
return self._tools.get(name)
def get_by_category(self, category: str) -> List[Tool]:
"""Get all tools in a category."""
names = self._categories.get(category, [])
return [self._tools[n] for n in names if n in self._tools]
def list_all(self) -> List[Tool]:
"""List all registered tools."""
return list(self._tools.values())
def load_from_config(self, config_path: str):
"""Load tools from YAML config."""
with open(config_path) as f:
config = yaml.safe_load(f)
for tool_config in config.get('tools', []):
# Dynamic import of tool function
module_path = tool_config['module']
func_name = tool_config['function']
module = importlib.import_module(module_path)
func = getattr(module, func_name)
tool = Tool(
name=tool_config['name'],
description=tool_config['description'],
parameters=tool_config['parameters'],
function=func
)
self.register(tool, tool_config.get('category', 'general'))
# Global registry
registry = ToolRegistry()
# Register tools
registry.register(calculate, "math")
registry.register(search_web, "search")
registry.register(get_weather, "utilities")
# Use in agent
math_tools = registry.get_by_category("math")
agent = SimpleAgent(math_tools)Async Tool Execution
import asyncio
from typing import Coroutine
class AsyncTool(Tool):
"""Tool with async execution support."""
async def execute_async(self, **kwargs) -> Any:
"""Execute tool asynchronously."""
if asyncio.iscoroutinefunction(self.function):
return await self.function(**kwargs)
else:
# Run sync function in thread pool
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, lambda: self.function(**kwargs))
class AsyncAgent:
"""Agent with parallel tool execution."""
def __init__(self, tools: list[AsyncTool], model: str = "gpt-4o"):
self.tools = {t.name: t for t in tools}
self.model = model
async def _execute_tools_parallel(
self,
tool_calls: list
) -> list[dict]:
"""Execute multiple tool calls in parallel."""
async def execute_one(tool_call):
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if name in self.tools:
result = await self.tools[name].execute_async(**args)
else:
result = f"Unknown tool: {name}"
return {
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
}
results = await asyncio.gather(*[
execute_one(tc) for tc in tool_calls
])
return list(results)
async def run(self, user_message: str, max_iterations: int = 10) -> str:
"""Run agent with async tool execution."""
messages = [{"role": "user", "content": user_message}]
for _ in range(max_iterations):
response = client.chat.completions.create(
model=self.model,
messages=messages,
tools=[t.to_openai_schema() for t in self.tools.values()],
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
if not message.tool_calls:
return message.content
# Execute all tool calls in parallel
tool_results = await self._execute_tools_parallel(message.tool_calls)
messages.extend(tool_results)
return "Max iterations reached"
# Usage
async def main():
agent = AsyncAgent([
AsyncTool("search1", "Search source 1", {}, search_web),
AsyncTool("search2", "Search source 2", {}, search_web),
])
# Both searches run in parallel
result = await agent.run("Compare info from source 1 and source 2 about Python")
print(result)
asyncio.run(main())Production Agent Service
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid
app = FastAPI()
class AgentRequest(BaseModel):
message: str
session_id: Optional[str] = None
tools: list[str] = [] # Tool names to enable
class AgentResponse(BaseModel):
response: str
session_id: str
tool_calls: list[dict]
tokens_used: int
# Session storage (use Redis in production)
sessions: dict[str, list] = {}
@app.post("/agent/chat", response_model=AgentResponse)
async def chat(request: AgentRequest):
"""Chat with the agent."""
# Get or create session
session_id = request.session_id or str(uuid.uuid4())
messages = sessions.get(session_id, [])
# Get enabled tools
enabled_tools = [
registry.get(name)
for name in request.tools
if registry.get(name)
]
# Add user message
messages.append({"role": "user", "content": request.message})
# Run agent
tool_calls_made = []
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=[t.to_openai_schema() for t in enabled_tools] if enabled_tools else None
)
message = response.choices[0].message
# Handle tool calls
if message.tool_calls:
messages.append(message)
for tc in message.tool_calls:
name = tc.function.name
args = json.loads(tc.function.arguments)
tool = registry.get(name)
if tool:
result = tool.execute(**args)
tool_calls_made.append({
"tool": name,
"args": args,
"result": str(result)[:500]
})
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": str(result)
})
# Get final response
response = client.chat.completions.create(
model="gpt-4o",
messages=messages
)
message = response.choices[0].message
messages.append({"role": "assistant", "content": message.content})
sessions[session_id] = messages
return AgentResponse(
response=message.content,
session_id=session_id,
tool_calls=tool_calls_made,
tokens_used=response.usage.total_tokens
)
@app.get("/tools")
async def list_tools():
"""List available tools."""
return {
"tools": [
{"name": t.name, "description": t.description}
for t in registry.list_all()
]
}References
- OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
- LangChain Agents: https://python.langchain.com/docs/modules/agents/
- ReAct Paper: https://arxiv.org/abs/2210.03629
- Anthropic Tool Use: https://docs.anthropic.com/claude/docs/tool-use
Conclusion
Tool-using agents transform LLMs from text generators into capable assistants that can take action. Start with well-defined tools—clear names, descriptions, and parameter schemas help the model choose correctly. Implement the basic agent loop first, then add sophistication with ReAct-style reasoning traces for transparency. Use a tool registry for clean organization and dynamic tool loading. For production, add async execution for parallel tool calls, session management for multi-turn conversations, and proper error handling. The combination of LLM reasoning and tool execution creates agents that can research, calculate, query APIs, and accomplish real tasks autonomously.
Discover more from Code, Cloud & Context
Subscribe to get the latest posts sent to your email.