Lesson 5: Building Custom Tools — Web Search, Code Execution, API Integration

Build AI Agents: From Zero to Production

Part 2: Function Calling & Tool Use

xdev.asia

Introduction

In the previous lesson, you learned how function calling works. This article focuses on the actual build tools that agents need in production: web search, code execution sandbox, database queries, and REST API integration.

---

1. Web Search Tool

1.1 Using SerpAPI / Tavily

import httpx

def web_search(query: str, num_results: int = 5) -> str:
    """Tìm kiếm web và trả về kết quả top."""
    response = httpx.get("https://api.tavily.com/search", params={
        "api_key": TAVILY_API_KEY,
        "query": query,
        "max_results": num_results,
    })
    results = response.json()["results"]
    return "\n".join([
        f"- [{r['title']}]({r['url']}): {r['content'][:200]}"
        for r in results
    ])

---

2. Code Execution Sandbox

import subprocess
import tempfile

def execute_python(code: str, timeout: int = 10) -> str:
    """Chạy Python code trong sandbox."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        f.flush()
        try:
            result = subprocess.run(
                ['python', f.name],
                capture_output=True, text=True, timeout=timeout,
                env={"PATH": "/usr/bin"}  # Restricted env
            )
            return result.stdout or result.stderr
        except subprocess.TimeoutExpired:
            return "Error: Code execution timed out"

---

3. Database Query Tool

import sqlite3

def query_database(sql: str) -> str:
    """Chạy SQL query (read-only) trên database."""
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries allowed"
    
    conn = sqlite3.connect("app.db")
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        return json.dumps({"columns": columns, "rows": rows[:50]})
    except Exception as e:
        return f"SQL Error: {e}"
    finally:
        conn.close()

---

4. Error Handling & Retry Logic

def safe_tool_execute(tool_func, args, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            result = tool_func(**args)
            return {"status": "success", "data": result}
        except Exception as e:
            if attempt == max_retries:
                return {"status": "error", "error": str(e)}
            time.sleep(1)

---

Summary

• Web search, code execution, database query — the 3 most important tools
• Security first: sandbox code execution, read-only DB, input validation
• Error handling + retry logic is required for production agents
• Tool observability: log all tool calls for debugging

Exercises

1. Implement the actual web_search tool with the Tavily API
2. Build safer sandbox execution code (using Docker)
3. Create a REST API tool that can call any endpoint
4. Implement rate limiting for tools (avoid spam API calls)