This guide walks you through creating a custom agent from scratch, step by step.
pip install -r requirements.txtTools are regular Python functions decorated with @tool. The decorator captures metadata that the LLM uses to understand what the tool does and how to call it.
from src.tool_registry import ToolRegistry, tool
registry = ToolRegistry()
@registry.register
@tool(
description="Fetch the current price of a stock by ticker symbol",
param_descriptions={
"ticker": "Stock ticker symbol (e.g., 'AAPL', 'GOOGL')",
},
tags=["finance"],
)
def get_stock_price(ticker: str) -> str:
# In production, call a real API here
prices = {"AAPL": 178.50, "GOOGL": 141.20, "MSFT": 378.90}
price = prices.get(ticker.upper())
if price is None:
return f"Unknown ticker: {ticker}"
return f"{ticker.upper()}: ${price:.2f}"Key points:
from src.agent import Agent, AgentConfig
from src.llm_client import OpenAIClient # or MockLLMClient for testing
config = AgentConfig(
name="stock-analyst",
system_prompt="""You are a financial analyst assistant.
When asked about stocks, use the get_stock_price tool to look up current prices.
Always provide context with your answers (market trends, comparisons).
If you don't have data for a stock, say so clearly.""",
max_steps=15,
temperature=0.1,
model="gpt-4o",
)System prompt tips:
from src.guardrails import GuardrailChain, ContentFilter, RateLimiter, LengthGuardrail
guardrails = GuardrailChain()
# Block attempts to extract API keys or credentials
guardrails.add(ContentFilter(
blocked_patterns=[r"api.?key", r"secret", r"password"],
), input_only=True)
# Prevent runaway loops
guardrails.add(RateLimiter(max_calls=20, window_seconds=60))
# Block excessively long inputs (prompt injection defense)
guardrails.add(LengthGuardrail(max_chars=10_000), input_only=True)from src.memory.buffer import ConversationBuffer
from src.memory.long_term import LongTermMemory
from src.memory.summarizer import ConversationSummarizer
# Short-term: keeps recent conversation in context
summarizer = ConversationSummarizer()
buffer = ConversationBuffer(
max_turns=30,
max_tokens=6000,
summarize_fn=summarizer,
)
# Long-term: remembers facts across conversations
long_term = LongTermMemory(persist_path="./data/analyst_memory.json")
long_term.store("User prefers concise bullet-point responses", category="preference", importance=0.9)agent = Agent(
config=config,
llm=OpenAIClient(api_key="YOUR_API_KEY_HERE"),
tools=registry,
memory=buffer,
guardrails=guardrails,
event_handler=lambda e: print(f" [{e.event_type}] {e.content}"),
)
# Single-turn interaction
result = agent.run("What's the current price of AAPL?")
print(result)
# Multi-turn (memory preserves context)
result2 = agent.run("How does that compare to MSFT?")
print(result2)For simple Q&A, the default agent loop is fine. For complex multi-step tasks:
from src.planning.react import ReActPlanner
planner = ReActPlanner(
agent_config=config,
llm=OpenAIClient(api_key="YOUR_API_KEY_HERE"),
tools=registry,
max_iterations=8,
)
result = planner.solve("Compare AAPL, GOOGL, and MSFT prices and recommend the best value")
print(planner.format_trace()) # See the reasoning stepsfrom src.planning.plan_execute import PlanAndExecutePlanner
planner = PlanAndExecutePlanner(
llm=OpenAIClient(api_key="YOUR_API_KEY_HERE"),
tools=registry,
max_replans=2,
)
result = planner.solve("Research the top 5 tech stocks, analyze their prices, and write a summary report")| Pattern | Best For | Overhead |
|---|---|---|
| Basic Agent | Single-tool calls, Q&A | Lowest |
| ReAct | Multi-step reasoning, information gathering | Low |
| Plan-and-Execute | Complex projects, tasks with dependencies | Medium |
| Supervisor/Worker | Tasks requiring multiple specializations | Highest |
1. Too many tools: Agents get confused with 15+ tools. Group related functionality into fewer, more capable tools.
2. Vague system prompts: "Be helpful" is not enough. Specify the domain, available tools, and expected behavior.
3. No max_steps limit: Always set one. LLMs can get stuck in loops.
4. Ignoring tool errors: If a tool returns an error, the agent might try again with the same bad inputs. Consider adding retry logic in the tool itself.
5. Not testing with MockLLMClient: Real API calls are slow and expensive for iterating on agent design. Script the expected behavior with the mock first.
Tools are the interface between the LLM's reasoning and the real world. A well-designed tool makes the agent more capable and reliable. A poorly designed tool causes confusion, errors, and wasted API calls.
Bad: process_data(action: str, data: str) where action is "read", "write", "delete", "transform"
Good: Separate read_file(), write_file(), delete_file(), transform_data() tools.
Why? LLMs are better at choosing between distinct tools than passing the right action parameter to a Swiss-army-knife tool.
The tool description is part of the prompt the LLM sees. Write it like you're explaining the tool to a junior developer:
# Bad
@tool(description="Database query")
def query_db(sql: str) -> str: ...
# Good
@tool(description="Execute a read-only SQL query against the analytics database. Returns results as a formatted table. Only SELECT statements are allowed. Tables available: users, orders, products, events.")
def query_db(sql: str) -> str: ...The tool result goes back into the conversation as text. Return human-readable strings:
# Bad - LLM gets "{'price': 178.5, 'change': -2.3}"
def get_price(ticker: str) -> dict:
return {"price": 178.5, "change": -2.3}
# Good - LLM gets clear, parseable text
def get_price(ticker: str) -> str:
return "AAPL: $178.50 (down $2.30, -1.27% today)"Never let exceptions escape. Return error messages the LLM can understand and act on:
@tool(description="Look up a user by email")
def find_user(email: str) -> str:
if "@" not in email:
return "Error: Invalid email format. Please provide a valid email address."
# ... lookup logic
if user is None:
return f"No user found with email '{email}'. Check the spelling or try a different email."
return f"Found: {user.name} (ID: {user.id}, role: {user.role})"LLMs have finite context windows. Truncate large results:
@tool(description="Search logs for a pattern")
def search_logs(pattern: str, max_results: int = 10) -> str:
matches = find_all_matches(pattern)
if len(matches) > max_results:
shown = matches[:max_results]
return format_results(shown) + f"\n... and {len(matches) - max_results} more results"
return format_results(matches)# Bad - the LLM has to guess the format
def schedule_meeting(time: str) -> str: ...
# Good - constrained and clear
@tool(param_descriptions={"hour": "Hour in 24h format (0-23)", "minute": "Minute (0-59)"})
def schedule_meeting(hour: int, minute: int, duration_minutes: int = 30) -> str: ...@tool(description="Search documents")
def search(query: str, limit: int = 10, category: str = "all") -> str: ...@tool(
description="Set the priority of a task",
param_descriptions={
"priority": "One of: 'low', 'medium', 'high', 'critical'",
},
)
def set_priority(task_id: str, priority: str) -> str:
valid = {"low", "medium", "high", "critical"}
if priority not in valid:
return f"Invalid priority '{priority}'. Must be one of: {', '.join(sorted(valid))}"
# ...ALLOWED_DIR = Path("/data/workspace")
def read_file(path: str) -> str:
resolved = (ALLOWED_DIR / path).resolve()
if not str(resolved).startswith(str(ALLOWED_DIR.resolve())):
return "Error: Access denied. Path is outside the allowed directory."
return resolved.read_text()@tool(
description="Send an email",
requires_approval=True, # Human must approve before sending
)
def send_email(to: str, subject: str, body: str) -> str: ...# NEVER do this
@tool(description="Run any SQL")
def raw_sql(query: str) -> str:
return db.execute(query) # SQL injection risk
# Instead, provide specific query tools
@tool(description="Get order details by order ID")
def get_order(order_id: str) -> str:
return db.execute("SELECT * FROM orders WHERE id = %s", [order_id])Always test tools independently before giving them to an agent:
# test_tools.py
def test_stock_price_valid_ticker():
result = get_stock_price("AAPL")
assert "AAPL" in result
assert "$" in result
def test_stock_price_invalid_ticker():
result = get_stock_price("INVALID")
assert "Unknown" in result # Graceful error, not an exception
def test_stock_price_case_insensitive():
result = get_stock_price("aapl")
assert "AAPL" in result # Should normalize inputWrap external APIs to present a simpler interface:
@tool(description="Get the weather forecast for a city")
def get_weather(city: str) -> str:
# Internally calls a complex API, but returns simple text
raw = weather_api.get(city=city, units="metric", days=1)
return f"{city}: {raw['temp']}°C, {raw['condition']}, wind {raw['wind_speed']} km/h"Create tools that combine multiple operations:
@tool(description="Analyze a CSV file: read it, compute statistics, and return a summary")
def analyze_csv(path: str) -> str:
content = read_file(path)
stats = compute_stats(content)
return format_stats_report(stats)This reduces the number of agent steps (and API calls) for common workflows.