Connecting Agents to Real APIs

Moving Beyond Mock Tools

Every agent tutorial shows mock functions that return fake data. “Here’s your stock price: $100.” The real challenge — and where most production systems spend their time — is wrapping actual APIs into reliable agent tools.

Real APIs have authentication, rate limits, pagination, inconsistent response formats, and transient failures. Your agent tools need to handle all of this gracefully, because the agent has no concept of HTTP headers or retry logic. It just expects to ask a question and get an answer.

This lesson builds a GitHub issue triage agent — a practical tool that searches open issues, labels them, assigns them to team members, and posts triage comments. Every step involves a real API call.

The GitHub Issue Triage Agent

Use case: A development team has 200 open GitHub issues. The agent will:

1. Fetch all open issues and identify unlabeled ones
2. Analyze each issue’s content and suggest appropriate labels
3. Apply labels based on the analysis
4. Assign issues to the right team member based on area of expertise
5. Post a standardized triage comment

This requires both read and write operations — a realistic production scenario.

Wrapping a REST API as an Agent Tool

The pattern for any REST API tool is the same:

import requests
import os
import time
from typing import Optional
from functools import wraps

# GitHub API configuration
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
GITHUB_API_BASE = "https://api.github.com"

HEADERS = {
    "Authorization": f"Bearer {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json",
    "X-GitHub-Api-Version": "2022-11-28"
}


def github_request(
    method: str,
    endpoint: str,
    params: dict = None,
    json_body: dict = None,
    max_retries: int = 3
) -> dict:
    """
    Make a GitHub API request with retry logic and rate limit handling.
    This is the foundation all GitHub tools are built on.
    """
    url = f"{GITHUB_API_BASE}{endpoint}"
    
    for attempt in range(max_retries):
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=HEADERS,
                params=params,
                json=json_body,
                timeout=15
            )
            
            # Handle rate limiting
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 60))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                time.sleep(retry_after)
                continue
            
            # Handle 403 with rate limit headers (GitHub-specific)
            if response.status_code == 403:
                remaining = int(response.headers.get("X-RateLimit-Remaining", 1))
                if remaining == 0:
                    reset_time = int(response.headers.get("X-RateLimit-Reset", time.time() + 60))
                    wait_time = max(0, reset_time - time.time()) + 1
                    print(f"Rate limit hit. Waiting {wait_time:.0f} seconds...")
                    time.sleep(wait_time)
                    continue
            
            # Raise for other 4xx/5xx errors
            response.raise_for_status()
            
            # Handle empty responses (e.g., 204 No Content)
            if response.status_code == 204:
                return {"success": True}
            
            return response.json()
        
        except requests.exceptions.Timeout:
            if attempt == max_retries - 1:
                return {"error": f"Request timed out after {max_retries} attempts"}
            time.sleep(2 ** attempt)  # Exponential backoff
        
        except requests.exceptions.HTTPError as e:
            return {"error": f"HTTP {response.status_code}: {str(e)}", 
                    "body": response.text[:500]}
        
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                return {"error": f"Request failed: {str(e)}"}
            time.sleep(2 ** attempt)
    
    return {"error": "All retry attempts failed"}

This github_request function handles three common real-world problems:

1. Auth: the token is added to every request via headers
2. Rate limiting: detects 429 responses and waits before retrying
3. Transient failures: retries with exponential backoff on timeouts/errors

Every tool you build on top of this inherits these properties automatically.

Defining the GitHub Tools

import json

def list_open_issues(repo: str, labels: str = "", per_page: int = 30) -> str:
    """
    Fetch open issues from a repository.
    
    Args:
        repo: Repository in format "owner/repo" (e.g. "pytorch/pytorch")
        labels: Comma-separated labels to filter by (optional)
        per_page: Number of issues to fetch (max 100)
    """
    params = {
        "state": "open",
        "per_page": min(per_page, 100),
        "sort": "created",
        "direction": "desc"
    }
    if labels:
        params["labels"] = labels
    
    result = github_request("GET", f"/repos/{repo}/issues", params=params)
    
    if "error" in result:
        return f"Error fetching issues: {result['error']}"
    
    # Format for the agent — return clean, relevant fields
    issues = []
    for issue in result:
        if "pull_request" in issue:  # Skip PRs, only real issues
            continue
        issues.append({
            "number": issue["number"],
            "title": issue["title"],
            "body": issue.get("body", "")[:500],  # Truncate long bodies
            "labels": [l["name"] for l in issue.get("labels", [])],
            "assignees": [a["login"] for a in issue.get("assignees", [])],
            "created_at": issue["created_at"][:10]
        })
    
    return json.dumps(issues, indent=2)


def apply_label_to_issue(repo: str, issue_number: int, labels: list[str]) -> str:
    """
    Apply labels to a GitHub issue.
    
    Args:
        repo: Repository in format "owner/repo"
        issue_number: The issue number (integer)
        labels: List of label names to apply (must already exist in the repo)
    """
    result = github_request(
        "POST",
        f"/repos/{repo}/issues/{issue_number}/labels",
        json_body={"labels": labels}
    )
    
    if "error" in result:
        return f"Error applying labels: {result['error']}"
    
    applied = [l["name"] for l in result]
    return f"Applied labels {applied} to issue #{issue_number}"


def assign_issue(repo: str, issue_number: int, assignees: list[str]) -> str:
    """
    Assign a GitHub issue to one or more team members.
    
    Args:
        repo: Repository in format "owner/repo"
        issue_number: The issue number (integer)
        assignees: List of GitHub usernames to assign
    """
    result = github_request(
        "POST",
        f"/repos/{repo}/issues/{issue_number}/assignees",
        json_body={"assignees": assignees}
    )
    
    if "error" in result:
        return f"Error assigning issue: {result['error']}"
    
    return f"Assigned issue #{issue_number} to {assignees}"


def post_comment(repo: str, issue_number: int, body: str) -> str:
    """
    Post a comment on a GitHub issue.
    
    Args:
        repo: Repository in format "owner/repo"
        issue_number: The issue number (integer)
        body: The comment text (markdown supported)
    """
    result = github_request(
        "POST",
        f"/repos/{repo}/issues/{issue_number}/comments",
        json_body={"body": body}
    )
    
    if "error" in result:
        return f"Error posting comment: {result['error']}"
    
    return f"Comment posted on issue #{issue_number}: {result.get('html_url', '')}"

Building the Triage Agent

import anthropic

client = anthropic.Anthropic()

TRIAGE_TOOLS = [
    {
        "name": "list_open_issues",
        "description": """Fetch open GitHub issues from a repository.
        Use this first to see what issues need triage.
        Returns: list of issues with number, title, body preview, current labels, and assignees.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "repo": {"type": "string", "description": "Repository as 'owner/repo'"},
                "labels": {"type": "string", "description": "Filter by label (optional)"},
                "per_page": {"type": "integer", "description": "How many issues to fetch (default 30)"}
            },
            "required": ["repo"]
        }
    },
    {
        "name": "apply_label_to_issue",
        "description": """Apply labels to a GitHub issue.
        Use after analyzing an issue's content to categorize it.
        Common labels: 'bug', 'enhancement', 'documentation', 'question', 'good first issue'.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "repo": {"type": "string"},
                "issue_number": {"type": "integer"},
                "labels": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["repo", "issue_number", "labels"]
        }
    },
    {
        "name": "assign_issue",
        "description": """Assign a GitHub issue to team members.
        Use based on the issue's technical area and team expertise.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "repo": {"type": "string"},
                "issue_number": {"type": "integer"},
                "assignees": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["repo", "issue_number", "assignees"]
        }
    },
    {
        "name": "post_comment",
        "description": """Post a triage comment on a GitHub issue.
        Use to acknowledge the issue and provide initial guidance.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "repo": {"type": "string"},
                "issue_number": {"type": "integer"},
                "body": {"type": "string", "description": "Markdown comment body"}
            },
            "required": ["repo", "issue_number", "body"]
        }
    }
]

TOOL_FUNCTIONS = {
    "list_open_issues": list_open_issues,
    "apply_label_to_issue": apply_label_to_issue,
    "assign_issue": assign_issue,
    "post_comment": post_comment
}

TRIAGE_SYSTEM = """You are a GitHub issue triage agent. Your job is to help development teams 
manage their issue backlog.

Team expertise:
- @alice-dev: backend/API, Python, databases
- @bob-frontend: frontend, JavaScript/React, CSS
- @carol-infra: DevOps, Docker, Kubernetes, CI/CD
- @dave-ml: machine learning, model training, data pipelines

When triaging issues:
1. Fetch unlabeled issues first
2. Analyze title and body to determine: type (bug/enhancement/question/docs), area (backend/frontend/infra/ml)
3. Apply appropriate labels
4. Assign to the relevant team member based on area
5. Post a brief, friendly triage comment acknowledging the issue

Be efficient — triage multiple issues in one pass before posting comments."""


def run_triage_agent(repo: str, max_issues: int = 10) -> str:
    messages = [{
        "role": "user",
        "content": f"Triage unlabeled open issues in {repo}. Process up to {max_issues} issues."
    }]
    
    for _ in range(20):  # More iterations for batch operations
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            system=TRIAGE_SYSTEM,
            tools=TRIAGE_TOOLS,
            messages=messages
        )
        
        if response.stop_reason == "end_turn":
            return response.content[0].text
        
        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    fn = TOOL_FUNCTIONS[block.name]
                    result = fn(**block.input)
                    print(f"[{block.name}] {str(block.input)[:80]}...")
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            
            messages.append({"role": "user", "content": tool_results})
    
    return "Triage complete (max iterations reached)."


# Run
summary = run_triage_agent("your-org/your-repo", max_issues=5)
print(summary)

REST vs Database vs File System Tools

Different types of external systems have different patterns:

REST API tools (like GitHub above):

• Auth via headers (API keys, OAuth tokens)
• Responses are JSON — always validate the structure
• Rate limits are common — build in retry logic
• Pagination means you might need multiple calls for large datasets

Database tools:

• Use parameterized queries to prevent SQL injection
• Limit result sizes — a SELECT * with no limit can return millions of rows
• Wrap writes in transactions where possible
• Return counts, not raw data, for validation

import sqlite3

def query_database(sql: str, params: tuple = ()) -> str:
    """Execute a read-only SQL query. Max 100 rows returned."""
    # Safety: only allow SELECT statements
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: only SELECT queries are allowed"
    
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    cursor.execute(sql + " LIMIT 100", params)  # Always add LIMIT
    
    columns = [d[0] for d in cursor.description]
    rows = cursor.fetchall()
    conn.close()
    
    return json.dumps({"columns": columns, "rows": rows, "count": len(rows)})

File system tools:

• Always validate paths to prevent directory traversal attacks
• Set a working directory and reject paths outside it
• Return file contents with size limits
• Log all write operations

import os

SAFE_DIR = "/app/workspace"  # Only allow access within this directory

def read_file_safe(filepath: str) -> str:
    """Read a file, restricted to the workspace directory."""
    # Resolve the full path and check it's within SAFE_DIR
    full_path = os.path.realpath(os.path.join(SAFE_DIR, filepath))
    
    if not full_path.startswith(SAFE_DIR):
        return "Error: Access denied. Path is outside the workspace."
    
    if not os.path.exists(full_path):
        return f"Error: File not found: {filepath}"
    
    file_size = os.path.getsize(full_path)
    if file_size > 100_000:  # 100KB limit
        return f"Error: File too large ({file_size} bytes). Max 100KB."
    
    with open(full_path) as f:
        return f.read()

Summary

• Wrap all REST API calls in a shared request function that handles auth, rate limiting, and retries
• Format tool outputs to be clean and agent-friendly — not raw API responses
• REST tools need auth headers and retry logic; database tools need parameterized queries and LIMIT clauses; file system tools need path validation
• The GitHub triage agent shows a complete write-capable agent: reads issues, applies labels, assigns, and comments
• Return structured error messages from tools — the agent can reason about errors and try alternatives
• Always test tools against the real API in isolation before wiring them into the agent loop

Next: the Capstone — building a complete research-and-report agent from scratch, combining everything in this course.