Free lesson · GenAI Agent Engineering
Build HTTP-based agent tools
You can wrap an external API as an agent tool: implement Bearer token authentication, manage credentials for multiple APIs cleanly, and structure the request/response surface to be invoked safely by an LLM.
Course: GenAI Agent Engineering · Chapter 9 · The HTTP Client
Free to read — no subscription required.
Introduction
When you wire an agent to a third-party API by sprinkling raw httpx calls through your codebase, every retry, auth header, timeout, and connection pool becomes a copy-paste hazard — and the first transient 503 in production silently corrupts a multi-step agent run. Teams that skip the tool-shaped abstraction discover the cost when one careless requests.get exhausts the event loop or leaks credentials into a traceback. By the end of this lesson, you'll be able to design an HTTP-based agent tool that owns its httpx.AsyncClient, enforces concurrency limits, validates responses with Pydantic, and cleans up cleanly through the async context manager protocol.
Key Terminology
- HTTP-based agent tool — a class that wraps a single upstream HTTP API behind a typed,
asyncinterface the agent can invoke; owning the client lets the tool control auth, retries, and connection pooling instead of leaking those concerns to the agent. - Lazy client initialization — deferring
httpx.AsyncClientcreation until the first request so the tool can be constructed in any thread/event-loop context without binding to one prematurely. - Semaphore-based rate limiting — an
asyncio.Semaphorethat caps in-flight requests from one tool so a burst of agent calls cannot overwhelm the upstream API or starve the local event loop. - Pydantic response validation — parsing every API response into a typed model at the boundary so downstream agent logic operates on validated fields, not raw dicts that may shift shape on the next API release.
- Async context manager protocol —
__aenter__/__aexit__methods that guarantee the underlying client's connection pool is closed even when the agent raises mid-task.
Concepts
HTTP-based agent tools share four load-bearing concerns: client ownership, concurrency control, response validation, and lifecycle cleanup. Get all four right and the tool composes safely into any agent runtime; miss one and the failure mode is silent (a leaked pool, a hung event loop, a malformed result that pollutes downstream reasoning).
Tool Ownership of the HTTP Client
The tool — not the agent, not a module-level singleton — owns its httpx.AsyncClient. That ownership is what makes per-tool timeouts, base URLs, and auth headers possible, and what lets the context manager protocol guarantee cleanup. Lazy initialization in _ensure_client avoids binding the client to whatever event loop happened to import the module first (see Code Walkthrough).
Bounded Concurrency via Semaphore
An agent in a loop can fire dozens of tool calls per turn; without a per-tool semaphore, a single misbehaving prompt can saturate the upstream API and trip its rate limiter for every other tenant. asyncio.Semaphore(max_concurrent) acquired inside search() bounds in-flight requests at the tool boundary — outside that boundary, the agent doesn't have to know or care.
Typed Validation at the API Boundary
Validating responses into Pydantic models at the moment they cross the network boundary turns an unbounded class of "the API returned something weird" bugs into a single, loud ValidationError you can log and recover from. The agent then operates on SearchResult instances with typed fields, not dict[str, Any] that may rename snippet to summary on the next API release.
Abstracting Shared Plumbing in a Base Client
When a codebase grows to half a dozen HTTP tools, retry logic, exponential backoff, auth headers, and the close/aenter/aexit triad are the same in each one. Lifting them into an abstract AsyncAPIClient base class — with _get_auth_headers left abstract per subclass — keeps the per-tool subclass focused on endpoint methods (see Code Walkthrough).
Code Walkthrough
The two snippets below demonstrate the four concepts from the previous section: tool-owned lazy clients, semaphore-bounded concurrency, Pydantic validation at the boundary, and an abstract base class that hoists shared retry/auth plumbing out of every concrete tool.
Concrete Tool: SearchTool
The implementation below wraps a search API end-to-end: a Pydantic SearchResult model, a SearchTool class with lazy client init and an asyncio.Semaphore, a search() method that validates rows into the model, and __aenter__/__aexit__ so the pool always shuts down.
Code snippet python
1import httpx 2import asyncio 3from pydantic import BaseModel, Field 4from typing import List, Optional 5from dataclasses import dataclass 6import logging 7 8logger = logging.getLogger(__name__) 9 10class SearchResult(BaseModel): 11 title: str 12 url: str 13 snippet: str 14 relevance_score: float = Field(ge=0.0, le=1.0) 15 16class SearchTool: 17 def __init__( 18 self, 19 api_key: str, 20 base_url: str = "https://api.search.example.com", 21 timeout: float = 30.0, 22 max_concurrent: int = 10 23 ): 24 self.api_key = api_key 25 self.base_url = base_url 26 self.timeout = timeout 27 self.semaphore = asyncio.Semaphore(max_concurrent) 28 self.client: Optional[httpx.AsyncClient] = None 29 30 async def _ensure_client(self) -> httpx.AsyncClient: 31 if self.client is None: 32 self.client = httpx.AsyncClient( 33 base_url=self.base_url, 34 headers={ 35 "Authorization": f"Bearer {self.api_key}", 36 "User-Agent": "AgentSearchTool/1.0" 37 }, 38 timeout=self.timeout 39 ) 40 return self.client 41 42 async def search( 43 self, 44 query: str, 45 max_results: int = 10 46 ) -> List[SearchResult]: 47 async with self.semaphore: 48 client = await self._ensure_client() 49 50 try: 51 response = await client.get( 52 "/search", 53 params={"q": query, "limit": max_results} 54 ) 55 response.raise_for_status() 56 57 data = response.json() 58 return [ 59 SearchResult.model_validate(r) 60 for r in data.get("results", []) 61 ] 62 63 except httpx.HTTPStatusError as e: 64 logger.error(f"Search API error: {e.response.status_code}") 65 raise 66 except Exception as e: 67 logger.error(f"Search failed: {e}") 68 raise 69 70 async def close(self): 71 if self.client: 72 await self.client.aclose() 73 self.client = None 74 75 async def __aenter__(self): 76 return self 77 78 async def __aexit__(self, *args): 79 await self.close() 80 81# Usage 82async def main(): 83 async with SearchTool(api_key="secret-key") as search: 84 results = await search.search("python async programming") 85 for result in results: 86 print(f"- {result.title}: {result.url}")
- Lines 10-14:
SearchResultmodel — Pydantic validates fields at the boundary;relevance_scoreis bounded to[0.0, 1.0]. - Lines 16-28:
__init__stores config and primes the semaphore;self.clientstaysNoneuntil first use. - Lines 30-40:
_ensure_clientlazy-builds theAsyncClientso it binds to the running event loop, not theimport-time loop. - Lines 42-67:
search()acquires the semaphore, calls the endpoint, validates rows intoSearchResult, and re-raises after logging. - Lines 69-78:
closeplus__aenter__/__aexit__make the tool anasynccontext manager so the pool always shuts down.
Base Client Class Pattern
When several HTTP tools share retry, auth, and lifecycle logic, lift those concerns into an abstract AsyncAPIClient and let each tool subclass implement only _get_auth_headers and its endpoint methods. The code below shows the base class (retry + exponential backoff baked into _request) and a concrete WeatherAPIClient subclass that supplies the auth header and two endpoint methods.
Code snippet python
1import httpx 2import asyncio 3from abc import ABC, abstractmethod 4from typing import Optional, Any, Dict 5from pydantic import BaseModel 6import logging 7 8logger = logging.getLogger(__name__) 9 10class AsyncAPIClient(ABC): 11 def __init__( 12 self, 13 base_url: str, 14 api_key: str, 15 timeout: float = 30.0, 16 max_retries: int = 3 17 ): 18 self.base_url = base_url 19 self.api_key = api_key 20 self.timeout = timeout 21 self.max_retries = max_retries 22 self.client: Optional[httpx.AsyncClient] = None 23 24 @abstractmethod 25 def _get_auth_headers(self) -> Dict[str, str]: 26 pass 27 28 async def _ensure_client(self) -> httpx.AsyncClient: 29 if self.client is None: 30 self.client = httpx.AsyncClient( 31 base_url=self.base_url, 32 headers=self._get_auth_headers(), 33 timeout=self.timeout 34 ) 35 return self.client 36 37 async def _request( 38 self, 39 method: str, 40 endpoint: str, 41 **kwargs 42 ) -> Any: 43 client = await self._ensure_client() 44 45 for attempt in range(self.max_retries): 46 try: 47 response = await client.request(method, endpoint, **kwargs) 48 response.raise_for_status() 49 return response.json() 50 51 except httpx.HTTPStatusError as e: 52 if e.response.status_code >= 500 and attempt < self.max_retries - 1: 53 await asyncio.sleep(2 ** attempt) 54 continue 55 raise 56 except httpx.TimeoutException: 57 if attempt < self.max_retries - 1: 58 await asyncio.sleep(2 ** attempt) 59 continue 60 raise 61 62 raise RuntimeError("Max retries exceeded") 63 64 async def get(self, endpoint: str, **kwargs) -> Any: 65 return await self._request("GET", endpoint, **kwargs) 66 67 async def post(self, endpoint: str, **kwargs) -> Any: 68 return await self._request("POST", endpoint, **kwargs) 69 70 async def close(self): 71 if self.client: 72 await self.client.aclose() 73 self.client = None 74 75 async def __aenter__(self): 76 return self 77 78 async def __aexit__(self, *args): 79 await self.close() 80 81# Concrete implementation for a specific API 82class WeatherAPIClient(AsyncAPIClient): 83 def _get_auth_headers(self) -> Dict[str, str]: 84 return {"X-API-Key": self.api_key} 85 86 async def get_current_weather(self, location: str) -> dict: 87 return await self.get( 88 "/weather/current", 89 params={"location": location} 90 ) 91 92 async def get_forecast(self, location: str, days: int = 7) -> dict: 93 return await self.get( 94 "/weather/forecast", 95 params={"location": location, "days": days} 96 )
- Lines 1-8: Import required modules including ABC for abstract base class definition.
- Lines 10-22: Define the abstract base class with common configuration parameters.
- Lines 24-26: Declare the abstract method that subclasses must implement for authentication.
- Lines 28-35: Implement lazy client initialization using the authentication headers from the subclass.
- Lines 37-61: Implement the core request method with built-in retry logic for transient failures.
- Lines 44-60: Retry loop that handles server errors and timeouts with exponential backoff.
- Lines 63-68: Provide convenience methods for common HTTP verbs.
- Lines 70-78: Implement resource management with close method and context manager protocol.
- Lines 80-94: Demonstrate a concrete implementation that provides the authentication headers and domain-specific methods.
You'll know it works when an async with WeatherAPIClient(...) as wx: block returns a typed dict on the first call, transparently retries with exponential backoff on a simulated 503, and wx.client is None after the block exits — confirming the lazy client was created, used, retried, and closed without leaking the underlying connection pool.
Do's and Don'ts
Having just seen how the SearchTool and AsyncAPIClient patterns hold together end-to-end, the items below distil that walkthrough into the habits that keep HTTP-based agent tools reliable in production.
Do's
- ✓Do let the tool own its
httpx.AsyncClient— lazy-initialize it in_ensure_clientso the client binds to the agent's running event loop, not theimport-time loop. - ✓Do gate every outbound call with an
asyncio.Semaphore— per-tool concurrency caps stop one agent turn from saturating the upstream API and tripping its rate limiter for the whole tenant. - ✓Do validate every response with Pydantic at the boundary — turn unbounded "API returned something weird" failures into one loud
ValidationErroryou can log and recover from.
Don'ts
- ✗Don't call
httpx.getdirectly from agent code — bypassing the tool wrapper loses retries, auth headers, connection pooling, and the close guarantee, and leaks transport concerns into the agent. - ✗Don't share one module-level
AsyncClientacross tools — they'd contend for the same pool, share timeouts that fit none of them, and tangle lifecycle ownership when one tool needs to shut down. - ✗Don't swallow
HTTPStatusErrorsilently — log the status and re-raise (or retry on 5xx only); a returnedNonelets the agent reason on missing data without knowing the call failed.
This lesson is free to read. Its hands-on lab — real code, in a cloud IDE — is part of the GenAI Agent Engineering subscription.
From · cancel anytime · Already a subscriber? Sign in →