AI Agent API Integration Tutorial: Connect Agents to Any External Service

An AI agent that can only reason is just a chatbot. The moment you connect it to external APIs -- Slack, databases, CRM systems, email -- it becomes a worker that can take real action.

This tutorial shows you how to build API integrations for AI agents: from simple REST calls to production-grade tools with authentication, retries, and error handling.

In this tutorial:

• How agents interact with APIs
• Building your first API tool
• Authentication patterns
• Error handling and retries
• Building a multi-service agent
• Rate limiting and cost control
• Testing API integrations

Related tutorials: AI Agent Tools Tutorial · AI Agent Python Tutorial · AI Agent Security Best Practices

How Agents Interact with APIs

AI agents interact with external services through tool functions -- also called function calling. The flow works like this:

User: "Send a Slack message to #engineering about the deploy"
                │
        Agent decides to use tool
                │
        Agent calls: send_slack_message(channel="#engineering", text="Deploy complete")
                │
        Your code executes the Slack API call
                │
        Result returns to agent: "Message sent successfully"
                │
        Agent responds to user: "Done! I've sent the message to #engineering."

The agent doesn't call APIs directly. It tells your code what to do, and your code handles the actual API interaction. This means you control exactly what agents can and can't do.

Building Your First API Tool

The Tool Definition Pattern

Every API tool has three parts: a description, parameters, and execution logic.

from openai import OpenAI
import json

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city. Returns temperature, conditions, and humidity.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name, e.g. 'San Francisco'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

The Execution Function

import requests

def get_weather(city: str, unit: str = "fahrenheit") -> str:
    api_key = os.environ.get("WEATHER_API_KEY")
    response = requests.get(
        "https://api.weatherapi.com/v1/current.json",
        params={"key": api_key, "q": city}
    )
    
    if response.status_code != 200:
        return f"Error: Could not fetch weather for {city}"
    
    data = response.json()
    temp = data["current"]["temp_f"] if unit == "fahrenheit" else data["current"]["temp_c"]
    condition = data["current"]["condition"]["text"]
    humidity = data["current"]["humidity"]
    
    return f"Weather in {city}: {temp}°{unit[0].upper()}, {condition}, {humidity}% humidity"

def execute_tool(name: str, arguments: dict) -> str:
    functions = {
        "get_weather": get_weather,
    }
    
    func = functions.get(name)
    if not func:
        return f"Unknown tool: {name}"
    
    try:
        return func(**arguments)
    except Exception as e:
        return f"Error executing {name}: {e}"

Running the Agent with API Access

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )
    
    message = response.choices[0].message
    
    if message.tool_calls:
        for tool_call in message.tool_calls:
            args = json.loads(tool_call.function.arguments)
            result = execute_tool(tool_call.function.name, args)
            
            messages.append(message)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
        
        final = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools
        )
        return final.choices[0].message.content
    
    return message.content

print(run_agent("What's the weather in Tokyo and London?"))

Authentication Patterns

API Keys

The most common pattern. Store keys in environment variables, never in code:

import os
from functools import wraps

def require_api_key(env_var: str):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            api_key = os.environ.get(env_var)
            if not api_key:
                return f"Error: {env_var} not configured. Set this environment variable."
            return func(*args, api_key=api_key, **kwargs)
        return wrapper
    return decorator

@require_api_key("SLACK_BOT_TOKEN")
def send_slack_message(channel: str, text: str, api_key: str = None) -> str:
    response = requests.post(
        "https://slack.com/api/chat.postMessage",
        headers={"Authorization": f"Bearer {api_key}"},
        json={"channel": channel, "text": text}
    )
    data = response.json()
    if data.get("ok"):
        return f"Message sent to {channel}"
    return f"Slack error: {data.get('error', 'unknown')}"

OAuth 2.0

For services that require OAuth (Google, GitHub, etc.):

import time

class OAuthManager:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.tokens = {}

    def set_tokens(self, service: str, access_token: str, refresh_token: str, expires_in: int):
        self.tokens[service] = {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "expires_at": time.time() + expires_in - 60
        }

    def get_token(self, service: str) -> str:
        token_data = self.tokens.get(service)
        if not token_data:
            raise ValueError(f"No tokens for {service}")
        
        if time.time() > token_data["expires_at"]:
            self._refresh(service)
        
        return token_data["access_token"]

    def _refresh(self, service: str):
        token_data = self.tokens[service]
        response = requests.post(self.token_url, data={
            "grant_type": "refresh_token",
            "refresh_token": token_data["refresh_token"],
            "client_id": self.client_id,
            "client_secret": self.client_secret,
        })
        new_tokens = response.json()
        self.set_tokens(
            service,
            new_tokens["access_token"],
            new_tokens.get("refresh_token", token_data["refresh_token"]),
            new_tokens["expires_in"]
        )

Error Handling and Retries

API calls fail. Networks drop. Rate limits hit. Your agent tools need to handle all of it.

import time
from typing import Callable

def with_retry(
    func: Callable,
    max_retries: int = 3,
    backoff_base: float = 1.0,
    retry_on_status: list[int] = [429, 500, 502, 503, 504]
) -> Callable:
    def wrapper(*args, **kwargs):
        last_error = None
        for attempt in range(max_retries):
            try:
                response = func(*args, **kwargs)
                
                if hasattr(response, 'status_code'):
                    if response.status_code in retry_on_status:
                        wait = backoff_base * (2 ** attempt)
                        time.sleep(wait)
                        continue
                    if response.status_code >= 400:
                        return f"API error {response.status_code}: {response.text[:200]}"
                
                return response
            except requests.exceptions.Timeout:
                last_error = "Request timed out"
                time.sleep(backoff_base * (2 ** attempt))
            except requests.exceptions.ConnectionError:
                last_error = "Connection failed"
                time.sleep(backoff_base * (2 ** attempt))
            except Exception as e:
                return f"Unexpected error: {e}"
        
        return f"Failed after {max_retries} retries: {last_error}"
    return wrapper

Building a Multi-Service Agent

Let's connect an agent to multiple APIs: Slack, a database, and email.

tools = [
    {
        "type": "function",
        "function": {
            "name": "send_slack_message",
            "description": "Send a message to a Slack channel",
            "parameters": {
                "type": "object",
                "properties": {
                    "channel": {"type": "string", "description": "Channel name like #general"},
                    "message": {"type": "string", "description": "Message text"}
                },
                "required": ["channel", "message"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_database",
            "description": "Run a SQL query against the analytics database. SELECT queries only.",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL query to execute"}
                },
                "required": ["sql"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "Send an email to one or more recipients",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {"type": "string", "description": "Email address"},
                    "subject": {"type": "string", "description": "Email subject"},
                    "body": {"type": "string", "description": "Email body text"}
                },
                "required": ["to", "subject", "body"]
            }
        }
    }
]

def execute_tool(name: str, args: dict) -> str:
    handlers = {
        "send_slack_message": send_slack_message,
        "query_database": query_database,
        "send_email": send_email,
    }
    handler = handlers.get(name)
    if not handler:
        return f"Unknown tool: {name}"
    try:
        return handler(**args)
    except Exception as e:
        return f"Error: {e}"

Now the agent can handle complex multi-service workflows:

response = run_agent(
    "Query the database for this week's top 10 selling products, "
    "send a summary to #sales on Slack, and email the detailed report to sales@company.com"
)

The agent will autonomously:

1. Run the SQL query
2. Format results for Slack
3. Send the Slack message
4. Create a detailed email
5. Send the email

Rate Limiting and Cost Control

Per-Tool Rate Limiting

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.calls = defaultdict(list)

    def check(self, tool_name: str) -> bool:
        now = time.time()
        self.calls[tool_name] = [
            t for t in self.calls[tool_name] if now - t < self.period
        ]
        
        if len(self.calls[tool_name]) >= self.max_calls:
            return False
        
        self.calls[tool_name].append(now)
        return True

limiter = RateLimiter(max_calls=10, period=60)

def execute_with_limits(name: str, args: dict) -> str:
    if not limiter.check(name):
        return f"Rate limit reached for {name}. Try again in a moment."
    return execute_tool(name, args)

Token Budget

class TokenBudget:
    def __init__(self, daily_limit: float):
        self.daily_limit = daily_limit
        self.spent = 0.0

    def track(self, prompt_tokens: int, completion_tokens: int):
        cost = (prompt_tokens * 0.0000025) + (completion_tokens * 0.00001)
        self.spent += cost
        if self.spent > self.daily_limit:
            raise RuntimeError(f"Daily budget exceeded: ${self.spent:.2f} / ${self.daily_limit:.2f}")

Use our AI Cost Calculator to estimate costs for different API workloads.

Testing API Integrations

Mock API Responses

import unittest
from unittest.mock import patch, Mock

class TestAgentTools(unittest.TestCase):
    @patch('requests.get')
    def test_get_weather(self, mock_get):
        mock_get.return_value = Mock(
            status_code=200,
            json=lambda: {
                "current": {
                    "temp_f": 72,
                    "condition": {"text": "Sunny"},
                    "humidity": 45
                }
            }
        )
        result = get_weather("San Francisco")
        self.assertIn("72", result)
        self.assertIn("Sunny", result)

    @patch('requests.post')
    def test_send_slack_failure(self, mock_post):
        mock_post.return_value = Mock(
            status_code=200,
            json=lambda: {"ok": False, "error": "channel_not_found"}
        )
        result = send_slack_message("#nonexistent", "test")
        self.assertIn("channel_not_found", result)

Integration Test with Real APIs

import pytest

@pytest.mark.skipif(not os.environ.get("RUN_INTEGRATION_TESTS"), reason="Needs real API keys")
def test_real_slack_integration():
    result = send_slack_message("#test-channel", "Integration test message")
    assert "sent" in result.lower() or "error" in result.lower()

Skip the Integration Work: Use Ivern

Ivern AI agents come with pre-built integrations:

• No API wrapper code -- tools are configured, not coded
• Built-in authentication -- add your keys once, agents use them everywhere
• Error handling included -- retries, rate limiting, and fallbacks are automatic
• Monitoring dashboard -- see every API call your agents make

Start building with connected agents: ivern.ai/signup

Key Takeaways

1. Agents don't call APIs directly -- they invoke tool functions that you control
2. Always authenticate via environment variables -- never hardcode API keys
3. Build retry logic into every tool -- networks are unreliable
4. Rate limit aggressively -- agents can make API calls faster than you expect
5. Test with mocks first, real APIs second -- save your rate limits for production

Next tutorials: AI Agent Tools Tutorial · AI Agent Security · AI Agent Monitoring