REST API Integration Agent
What is an API Integration Agent?
API integration agents automate interactions with REST APIs by discovering endpoints, handling authentication, constructing requests, processing responses, and recovering from errors. They enable LLMs to interact with external services programmatically.
The key capabilities are: OpenAPI/Swagger parsing, automatic endpoint discovery, OAuth2/API key management, request construction with validation, intelligent error handling with retries and circuit breakers, and response transformation.
Why This Matters
Every modern application exposes REST APIs, but integrating with dozens of APIs manually is time-consuming and error-prone. API integration agents eliminate this friction by automatically understanding API specifications and translating natural language requests into proper API calls.
Real-World Analogy
An API integration agent is like a universal remote control for your entertainment system. Instead of juggling multiple remotes (API documentation, authentication flows, error handling), you simply say "play jazz music" and the agent knows which device to control, which button to press, and how to handle if the device doesn't respond.
Project Overview
We will build an API integration agent that:
- Parses OpenAPI/Swagger specifications automatically
- Manages multiple authentication methods (API keys, OAuth2, Bearer tokens)
- Constructs requests from natural language instructions
- Handles rate limiting with exponential backoff and jitter
- Processes and transforms responses (JSON, XML, CSV)
- Logs all API interactions for audit and debugging
- Implements circuit breaker pattern for fault tolerance
Expected outcome: An agent that can interact with any REST API using natural language.
Difficulty: Advanced (requires understanding of REST APIs, OpenAPI, and authentication)
Tools & Setup
| Tool | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Core language |
| httpx | 0.27+ | Async HTTP client |
| openai | 1.0+ | LLM backbone |
| pydantic | 2.0+ | Data validation |
| pyyaml | 6.0+ | OpenAPI parsing |
| tenacity | 8.0+ | Retry logic |
Step 1: Environment Setup
python -m venv venv
source venv/bin/activate
pip install httpx openai pydantic pyyaml tenacity
export OPENAI_API_KEY="sk-your-key"
Step 2: OpenAPI Parser
# discovery/openapi_parser.py
import yaml
import json
from typing import Dict, List, Optional
import httpx
import logging
logger = logging.getLogger(__name__)
class OpenAPIParser:
"""Parse OpenAPI/Swagger specifications for endpoint discovery."""
def __init__(self):
self.spec: Optional[Dict] = None
self.endpoints: List[Dict] = []
self.base_url: str = ""
def parse_file(self, file_path: str) -> Dict:
with open(file_path, "r") as f:
if file_path.endswith((".yaml", ".yml")):
self.spec = yaml.safe_load(f)
else:
self.spec = json.load(f)
self._extract_endpoints()
logger.info("Parsed %d endpoints from %s", len(self.endpoints), file_path)
return self.spec
def parse_url(self, url: str) -> Dict:
response = httpx.get(url, timeout=10.0)
response.raise_for_status()
if url.endswith((".yaml", ".yml")):
self.spec = yaml.safe_load(response.text)
else:
self.spec = response.json()
self._extract_endpoints()
logger.info("Parsed %d endpoints from URL", len(self.endpoints))
return self.spec
def _extract_endpoints(self) -> None:
if not self.spec or "paths" not in self.spec:
return
servers = self.spec.get("servers", [])
if servers and isinstance(servers[0], dict):
self.base_url = servers[0].get("url", "")
for path, methods in self.spec["paths"].items():
for method, details in methods.items():
if method.lower() in ("get", "post", "put", "patch", "delete"):
parameters = details.get("parameters", [])
request_body = details.get("requestBody", {})
responses = details.get("responses", {})
self.endpoints.append({
"path": path,
"method": method.upper(),
"summary": details.get("summary", ""),
"description": details.get("description", ""),
"parameters": parameters,
"request_body": request_body,
"responses": responses,
"tags": details.get("tags", []),
})
def get_endpoints_summary(self) -> str:
summary_lines: List[str] = []
for ep in self.endpoints:
params = len(ep.get("parameters", []))
has_body = bool(ep.get("request_body"))
summary_lines.append(
f"{ep['method']} {ep['path']} - {ep['summary']} "
f"(params: {params}, body: {has_body})"
)
return "\n".join(summary_lines)
def find_endpoint(self, description: str) -> Optional[Dict]:
desc_lower = description.lower()
for ep in self.endpoints:
if desc_lower in ep.get("summary", "").lower():
return ep
if desc_lower in ep.get("description", "").lower():
return ep
return None
Step 3: Auth Manager and Request Builder
# auth/manager.py
from typing import Dict, Optional
import httpx
import logging
logger = logging.getLogger(__name__)
class AuthManager:
"""Manage multiple authentication methods for API calls."""
def __init__(self):
self.credentials: Dict[str, Dict] = {}
def add_api_key(self, name: str, key: str, header: str = "Authorization", prefix: str = "Bearer") -> None:
self.credentials[name] = {
"type": "api_key",
"key": key,
"header": header,
"prefix": prefix,
}
logger.info("Added API key credential: %s", name)
def add_oauth2(self, name: str, token: str, refresh_token: str = "") -> None:
self.credentials[name] = {
"type": "oauth2",
"token": token,
"refresh_token": refresh_token,
"header": "Authorization",
}
logger.info("Added OAuth2 credential: %s", name)
def add_bearer(self, name: str, token: str) -> None:
self.credentials[name] = {
"type": "bearer",
"token": token,
"header": "Authorization",
}
def get_auth_headers(self, name: str) -> Dict[str, str]:
cred = self.credentials.get(name)
if not cred:
logger.warning("Credential not found: %s", name)
return {}
cred_type = cred["type"]
header = cred["header"]
if cred_type in ("api_key", "oauth2", "bearer"):
return {header: f"{cred.get('prefix', 'Bearer')} {cred['token']}"}
return {}
def remove_credential(self, name: str) -> None:
self.credentials.pop(name, None)
# execution/request_builder.py
import httpx
from typing import Dict, Any, Optional
import logging
import time
logger = logging.getLogger(__name__)
class RequestBuilder:
"""Build and send HTTP requests with comprehensive error handling."""
def __init__(self, timeout: float = 30.0, max_response_size: int = 10_000_000):
self.client = httpx.AsyncClient(timeout=timeout, follow_redirects=True)
self.max_response_size = max_response_size
async def build_and_send(
self,
method: str,
url: str,
params: Optional[Dict] = None,
json_data: Optional[Dict] = None,
headers: Optional[Dict] = None,
) -> Dict[str, Any]:
start_time = time.time()
try:
response = await self.client.request(
method=method,
url=url,
params=params,
json=json_data,
headers=headers or {},
)
latency_ms = (time.time() - start_time) * 1000
content_type = response.headers.get("content-type", "")
data = self._parse_response(response, content_type)
return {
"success": response.status_code < 400,
"status_code": response.status_code,
"headers": dict(response.headers),
"data": data,
"text": response.text[:5000] if len(response.text) > 5000 else response.text,
"latency_ms": round(latency_ms, 2),
}
except httpx.TimeoutException:
return {"success": False, "error": "Request timed out", "status_code": 408}
except httpx.RequestError as e:
return {"success": False, "error": f"Request failed: {str(e)}", "status_code": 0}
except Exception as e:
return {"success": False, "error": f"Unexpected error: {str(e)}", "status_code": 0}
def _parse_response(self, response: httpx.Response, content_type: str) -> Any:
if "json" in content_type:
return response.json()
elif "xml" in content_type:
return response.text
elif "csv" in content_type:
return response.text
return response.text
async def close(self) -> None:
await self.client.aclose()
# execution/error_handler.py
import time
import random
from typing import Dict, Any, Callable
from functools import wraps
import logging
logger = logging.getLogger(__name__)
class CircuitBreaker:
"""Circuit breaker pattern for fault tolerance."""
def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.last_failure_time: float = 0
self.state = "closed"
def record_failure(self) -> None:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
logger.warning("Circuit breaker opened after %d failures", self.failure_count)
def record_success(self) -> None:
self.failure_count = 0
self.state = "closed"
def is_open(self) -> bool:
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half-open"
return False
return True
return False
class ErrorHandler:
"""Handle API errors with exponential backoff and circuit breaker."""
def __init__(self, max_retries: int = 3, backoff_factor: float = 1.5):
self.max_retries = max_retries
self.backoff_factor = backoff_factor
self.circuit_breaker = CircuitBreaker()
def with_retry(self, func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Dict[str, Any]:
if self.circuit_breaker.is_open():
return {"success": False, "error": "Circuit breaker is open", "status_code": 503}
last_error = None
for attempt in range(self.max_retries):
result = await func(*args, **kwargs)
if result.get("success"):
self.circuit_breaker.record_success()
return result
last_error = result.get("error", "Unknown error")
status = result.get("status_code", 500)
if status == 429 or status >= 500:
wait = self.backoff_factor ** attempt + random.uniform(0, 1)
logger.warning("Attempt %d failed, retrying in %.1fs", attempt + 1, wait)
time.sleep(wait)
else:
break
self.circuit_breaker.record_failure()
return {"success": False, "error": f"Failed after {self.max_retries} retries: {last_error}"}
return wrapper
Step 4: Complete Agent
# agent.py
from discovery.openapi_parser import OpenAPIParser
from auth.manager import AuthManager
from execution.request_builder import RequestBuilder
from execution.error_handler import ErrorHandler
from openai import OpenAI
from typing import Dict, Any, Optional
import json
import logging
import hashlib
logger = logging.getLogger(__name__)
class APIIntegrationAgent:
"""Complete API integration agent with LLM-powered endpoint selection."""
def __init__(self, model: str = "gpt-4o"):
self.parser = OpenAPIParser()
self.auth = AuthManager()
self.request_builder = RequestBuilder()
self.error_handler = ErrorHandler()
self.llm = OpenAI()
self.model = model
self._response_cache: Dict[str, Dict] = {}
def load_api(self, spec_path: Optional[str] = None, spec_url: Optional[str] = None) -> str:
if spec_path:
self.parser.parse_file(spec_path)
elif spec_url:
self.parser.parse_url(spec_url)
else:
raise ValueError("Either spec_path or spec_url must be provided")
return f"Loaded {len(self.parser.endpoints)} endpoints"
async def call_api(
self,
instruction: str,
auth_name: Optional[str] = None,
use_cache: bool = True,
) -> Dict[str, Any]:
cache_key = hashlib.md5(instruction.encode()).hexdigest()
if use_cache and cache_key in self._response_cache:
logger.info("Cache hit for instruction: %s", instruction[:50])
return self._response_cache[cache_key]
endpoint = self.parser.find_endpoint(instruction)
if not endpoint:
endpoint = await self._llm_find_endpoint(instruction)
if not endpoint:
return {"success": False, "error": "No matching endpoint found"}
headers = self.auth.get_auth_headers(auth_name) if auth_name else {}
request_info = self._build_request(endpoint, instruction)
base_url = self.parser.base_url or "https://api.example.com"
url = f"{base_url}{endpoint.get('path', '/')}"
wrapped = self.error_handler.with_retry(self.request_builder.build_and_send)
result = await wrapped(
method=endpoint.get("method", "GET"),
url=url,
params=request_info.get("params"),
json_data=request_info.get("body"),
headers=headers,
)
if use_cache and result.get("success"):
self._response_cache[cache_key] = result
return result
async def _llm_find_endpoint(self, instruction: str) -> Optional[Dict]:
endpoints_desc = self.parser.get_endpoints_summary()
try:
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": f"Given these API endpoints:\n{endpoints_desc}\n\n"
"Find the best endpoint for the user's request. "
"Return JSON with 'path' and 'method'.",
},
{"role": "user", "content": instruction},
],
temperature=0.0,
max_tokens=100,
)
return json.loads(response.choices[0].message.content)
except Exception as e:
logger.error("LLM endpoint selection failed: %s", e)
return None
def _build_request(self, endpoint: Dict, instruction: str) -> Dict[str, Any]:
params: Dict[str, str] = {}
body: Dict[str, Any] = {}
for param in endpoint.get("parameters", []):
name = param.get("name", "")
if param.get("required"):
params[name] = f"value_for_{name}"
request_body = endpoint.get("request_body", {})
if request_body:
content = request_body.get("content", {})
json_content = content.get("application/json", {})
schema = json_content.get("schema", {})
if schema.get("properties"):
for prop_name in schema["properties"]:
body[prop_name] = f"value_for_{prop_name}"
return {"params": params, "body": body}
Why This Matters
Modern software is built on APIs â payments, email, CRM, analytics, and more. Manually integrating with each API requires reading documentation, handling auth, managing errors, and maintaining code. API integration agents automate this entire process, reducing integration time from days to minutes.
Real-World Analogy
An API integration agent is like a skilled travel agent who knows every airline's booking system. You say "find me the cheapest flight to Paris next week," and they know exactly which systems to query, how to authenticate, which parameters to use, and how to handle if a system is down.
Mathematical Foundation
API Reliability Score:
Intuition: Percentage of API calls that return successful responses. Target >99% for production reliability.
Retry Success Probability:
Where:
- â per-attempt success probability
- â number of retry attempts
Intuition: With and retries, success probability is .
Performance Considerations
| Metric | Value | Notes |
|---|---|---|
| OpenAPI Parse Time | 100-500ms | Depends on spec size |
| Endpoint Discovery | <100ms | In-memory search |
| LLM Endpoint Selection | 2-4s | GPT-4 |
| Request Execution | 200ms-5s | Depends on target API |
| Retry Overhead | 1-5s | For failed requests |
| Circuit Breaker Recovery | 60s | Configurable timeout |
| Cache Hit Rate | 20-40% | Depends on instruction diversity |
| End-to-End Latency | 0.5-8s | With caching, much faster |
Security Considerations
- Credential Management: Never expose API keys in responses or logs; use environment variables
- OAuth2 Token Refresh: Automatically refresh expired tokens before they cause failures
- Request Validation: Validate all inputs against OpenAPI schemas before sending
- Rate Limiting: Respect API rate limits to avoid IP bans
- Audit Logging: Log all API calls for compliance and debugging
- Response Sanitization: Strip sensitive data from cached responses
- TLS Enforcement: Always use HTTPS for API communication
Testing & Evaluation
import pytest
from discovery.openapi_parser import OpenAPIParser
from auth.manager import AuthManager
from execution.error_handler import CircuitBreaker
def test_parse_openapi():
parser = OpenAPIParser()
parser.parse_file("petstore.yaml")
assert len(parser.endpoints) > 0
def test_auth_manager():
manager = AuthManager()
manager.add_api_key("test", "abc123")
headers = manager.get_auth_headers("test")
assert "Authorization" in headers
assert "Bearer abc123" in headers["Authorization"]
def test_circuit_breaker():
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=1.0)
assert not cb.is_open()
for _ in range(3):
cb.record_failure()
assert cb.is_open()
def test_endpoint_finding():
parser = OpenAPIParser()
parser.spec = {"paths": {"/users": {"get": {"summary": "List users"}}}}
parser._extract_endpoints()
ep = parser.find_endpoint("list users")
assert ep is not None
assert ep["method"] == "GET"
Interview Q&A
Q1: How does OpenAPI parsing enable automatic API integration? A: OpenAPI specs provide machine-readable endpoint definitions including paths, parameters, request/response schemas, and authentication. This enables automatic discovery, request construction, and validation without manual coding. The parser extracts all endpoints, their HTTP methods, required parameters, and response formats.
Q2: What is the difference between API key and OAuth2 authentication? A: API keys are static tokens passed in headers, suitable for server-to-server communication. OAuth2 uses dynamic tokens with refresh flows, supporting granular scopes and token expiration. OAuth2 is more secure for user-delegated access because tokens can be revoked and have limited lifetimes.
Q3: How would you handle API versioning in production? A: Pin to specific API versions in the base URL (e.g., /v2/), monitor deprecation headers, implement version negotiation logic, maintain compatibility adapters for version transitions, and cache version-specific endpoint schemas separately.
Q4: What is the token bucket algorithm for rate limiting?
A: Tokens are added at a fixed rate to a bucket with maximum capacity. Each request consumes one token. If the bucket is empty, requests are queued or rejected. This allows burst traffic while maintaining average rate limits. The bucket refills at a rate of tokens_per_second.
Q5: How do you handle non-JSON API responses? A: Detect content-type header and route to appropriate parser: JSON for application/json, XML for application/xml, CSV for text/csv, and binary for other types. Return raw text as fallback. Use content negotiation headers when available.
Q6: How would you implement circuit breaker pattern for APIs? A: Track failure count within a time window. When failures exceed threshold (e.g., 5 in 60s), open circuit and reject calls immediately. After timeout, allow one test call (half-open state). If successful, close circuit; otherwise, keep open. This prevents cascade failures.
Q7: What are the security considerations for API integration agents? A: Never expose API keys in responses, use OAuth2 for user-delegated access, validate all inputs against schemas, implement request signing for sensitive APIs, audit all API calls, enforce HTTPS, and implement proper credential rotation.
Q8: How would you handle APIs with inconsistent response formats? A: Build response adapters per API, use schema inference to detect structure, implement fallback parsing strategies, maintain a response format registry for known APIs, and use LLM-based extraction for unstructured responses.
Common Pitfalls & Solutions
| Pitfall | Impact | Solution |
|---|---|---|
| Rate limiting | IP bans, service disruption | Implement exponential backoff with jitter, respect Retry-After headers |
| Auth token expiry | Failed requests | Auto-refresh tokens before expiration, cache tokens with TTL |
| API versioning | Broken integrations | Pin to specific versions, monitor deprecation headers |
| Error handling gaps | Unhandled exceptions | Comprehensive retry logic with circuit breaker |
| Data validation | Invalid requests | Schema-based request/response validation |
| Memory leaks | Resource exhaustion | Use connection pooling, async clients, and timeouts |
| Timeout issues | Slow responses | Set appropriate timeouts per endpoint, implement streaming |
| Cache staleness | Incorrect data | Implement TTL and invalidation strategies |
Summary with Key Takeaways
- OpenAPI parsing enables automatic endpoint discovery without manual coding
- Auth management handles multiple authentication methods (API keys, OAuth2, Bearer)
- Error handling with retries and circuit breakers ensures reliability during failures
- LLM-powered endpoint selection enables natural language API interaction
- Always validate requests against API schemas before sending
- Circuit breaker patterns prevent cascade failures in microservice architectures
- Response caching reduces latency and API costs for repeated queries