Skip to content
Home
Data Engineering
Generative AI

Agent-Native Orchestration: Engineering Robust Tool Wrappers for CLI and API Interactions

Mausam

When I tried to let a GPT‑4‑based agent fetch book metadata from Open Library, the agent kept inventing fields, ignored rate limits, and crashed on network hiccups. If you’ve ever embedded a raw httpx.get call inside a prompt, you know the pain: the LLM has no guarantee the request succeeded, and you lose the ability to describe the tool’s contract. In this tutorial I’ll show you how to wrap that endpoint (or any legacy CLI) in a Pydantic‑driven, async‑aware tool that validates inputs, retries intelligently, and surfaces structured errors—so the agent can focus on reasoning, not on brittle plumbing.

Key Takeaways

  • Explicit Pydantic schemas turn arbitrary endpoints into self‑documenting, type‑checked tools that prevent malformed agent calls.
  • Asynchronous httpx.AsyncClient with exponential backoff and rate‑limit handling keeps the agent responsive under load.
  • Structured error messages let the agent recover gracefully instead of aborting the whole chain of thought.
  • Embedding logging and optional Prometheus metrics gives you observability for production‑grade tool usage.
  • The same “CLI‑Anything” pattern applies to shell commands, making the approach universally reusable.

The Problem

In my disaster‑analytics pipeline the agent needed to pull the latest books on “data science” from Open Library to enrich a knowledge base. I initially wrote the request directly in the prompt:

agent: "Fetch https://openlibrary.org/search.json?q=data+science&limit=5"

That approach suffered from three fatal flaws:

  1. No input validation – the LLM could send an empty query or a malicious string.
  2. No guaranteed output shape – the agent would hallucinate fields that don’t exist in the JSON.
  3. No resilience – 429 or 5xx responses caused the whole reasoning loop to stall.

We need a formal tool interface that isolates those concerns from the agent’s core logic.

Data and Sources

Open Library Search API: https://openlibrary.org/developers/api
httpx async client docs: https://www.python-httpx.org/async/
Pydantic v2 reference: https://docs.pydantic.dev/latest/
LangChain tools docs: https://python.langchain.com/docs/modules/agents/tools/
Data accessed on 2026-10-24.

Loading the Data

Our tool will fetch live data from Open Library, so the first step is a thin async loader that returns raw JSON. We keep it separate to illustrate the “raw call” versus “wrapped tool” contrast.

import httpx
import asyncio

async def _raw_fetch(query: str, limit: int = 5) -> dict:
    url = "https://openlibrary.org/search.json"
    params = {"q": query, "limit": limit}
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(url, params=params)
        resp.raise_for_status()
        return resp.json()

The Core Logic

Step 1 – Define a strict schema for inputs and outputs. Pydantic guarantees that the agent can only pass the fields we expect.

from pydantic import BaseModel, Field
from typing import List, Optional

class OpenLibrarySearchInput(BaseModel):
    query: str = Field(..., description="Search term for books.")
    limit: int = Field(5, ge=1, le=20, description="Maximum number of results to return.")

class BookInfo(BaseModel):
    title: str
    author: Optional[str] = None
    first_publish_year: Optional[int] = None

class OpenLibrarySearchOutput(BaseModel):
    results: List[BookInfo]
    total: int
    error: Optional[str] = None

Step 2 – Build the robust async function. It validates inputs, retries on transient failures, respects Retry-After, and normalizes the response into BookInfo objects.

import logging
import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logger = logging.getLogger("openlibrary_tool")
logger.setLevel(logging.INFO)

# Retry on network errors or 5xx/429 status codes
@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=8),
    retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
)
async def _fetch_with_retry(input: OpenLibrarySearchInput) -> dict:
    try:
        raw = await _raw_fetch(input.query, input.limit)
        return raw
    except httpx.HTTPStatusError as exc:
        # Handle 429 specially to respect Retry-After
        if exc.response.status_code == 429:
            retry_after = int(exc.response.headers.get("Retry-After", "5"))
            logger.warning("Rate limited, sleeping %s seconds", retry_after)
            await asyncio.sleep(retry_after)
        raise

async def search_open_library(input: OpenLibrarySearchInput) -> OpenLibrarySearchOutput:
    try:
        raw = await _fetch_with_retry(input)
        docs = raw.get("docs", [])
        books = []
        for doc in docs:
            try:
                books.append(
                    BookInfo(
                        title=doc["title"],
                        author=doc["author_name"][0] if doc.get("author_name") else None,
                        first_publish_year=doc.get("first_publish_year"),
                    )
                )
            except (KeyError, TypeError) as parse_err:
                logger.error("Parsing error: %s", parse_err)
                continue
        return OpenLibrarySearchOutput(results=books, total=raw.get("numFound", 0))
    except httpx.RequestError as net_err:
        logger.error("Network error: %s", net_err)
        return OpenLibrarySearchOutput(results=[], total=0, error="Network issue, please retry.")
    except httpx.HTTPStatusError as http_err:
        logger.error("HTTP error %s: %s", http_err.response.status_code, http_err)
        return OpenLibrarySearchOutput(
            results=[],
            total=0,
            error=f"Open Library returned {http_err.response.status_code}.",
        )

Step 3 – Expose the function as a LangChain tool. The decorator pulls the Pydantic schema into the tool description automatically, giving the LLM a clear contract.

from langchain.tools import tool

@tool(
    name="OpenLibrarySearch",
    description="Searches Open Library for books matching a query and returns a concise list.",
    args_schema=OpenLibrarySearchInput,
)
async def open_library_tool(query: str, limit: int = 5) -> str:
    # Convert raw args into the Pydantic model
    input_model = OpenLibrarySearchInput(query=query, limit=limit)
    result = await search_open_library(input_model)
    if result.error:
        return f"error: {result.error}"
    # Serialize to a LLM‑friendly JSON string
    return result.json()

Putting It Together

Now we wire the tool into an async‑compatible LangChain agent. The agent will automatically discover the tool’s schema and decide when to call it.

from langchain.chat_models import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent

async def run_agent():
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    tools = [open_library_tool]
    agent = create_openai_functions_agent(llm, tools, verbose=True)
    executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)

    query = "Find 3 books on data science by Joel Grus and give their titles and first publish years."
    result = await executor.ainvoke({"input": query})
    print("Agent final answer:")
    print(result["output"])

if __name__ == "__main__":
    asyncio.run(run_agent())

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
Agent‑Native Orchestration demo:
Wrap Open Library Search as a reliable LangChain tool with Pydantic validation,
async retries, and structured error handling.
"""

import asyncio
import logging
from typing import List, Optional

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from pydantic import BaseModel, Field
from langchain.chat_models import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import tool

# ----------------------------------------------------------------------
# Logging configuration
# ----------------------------------------------------------------------
logger = logging.getLogger("openlibrary_tool")
handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s - %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# ----------------------------------------------------------------------
# Pydantic schemas
# ----------------------------------------------------------------------
class OpenLibrarySearchInput(BaseModel):
    """Validated input for the Open Library search tool."""
    query: str = Field(..., description="Search term for books.")
    limit: int = Field(
        5, ge=1, le=20, description="Maximum number of results to return."
    )

class BookInfo(BaseModel):
    title: str
    author: Optional[str] = None
    first_publish_year: Optional[int] = None

class OpenLibrarySearchOutput(BaseModel):
    results: List[BookInfo]
    total: int
    error: Optional[str] = None

# ----------------------------------------------------------------------
# Raw async fetch (demonstrates the naive approach)
# ----------------------------------------------------------------------
async def _raw_fetch(query: str, limit: int = 5) -> dict:
    url = "https://openlibrary.org/search.json"
    params = {"q": query, "limit": limit}
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(url, params=params)
        resp.raise_for_status()
        return resp.json()

# ----------------------------------------------------------------------
# Robust fetch with retry & rate‑limit handling
# ----------------------------------------------------------------------
@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=8),
    retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
)
async def _fetch_with_retry(input: OpenLibrarySearchInput) -> dict:
    try:
        return await _raw_fetch(input.query, input.limit)
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 429:
            retry_after = int(exc.response.headers.get("Retry-After", "5"))
            logger.warning("Rate limited – sleeping %s seconds", retry_after)
            await asyncio.sleep(retry_after)
        raise

# ----------------------------------------------------------------------
# Core tool logic
# ----------------------------------------------------------------------
async def search_open_library(input: OpenLibrarySearchInput) -> OpenLibrarySearchOutput:
    try:
        raw = await _fetch_with_retry(input)
        docs = raw.get("docs", [])
        books: List[BookInfo] = []
        for doc in docs:
            try:
                books.append(
                    BookInfo(
                        title=doc["title"],
                        author=doc["author_name"][0] if doc.get("author_name") else None,
                        first_publish_year=doc.get("first_publish_year"),
                    )
                )
            except (KeyError, TypeError) as parse_err:
                logger.error("Parsing error for doc %s: %s", doc.get("key"), parse_err)
                continue
        return OpenLibrarySearchOutput(results=books, total=raw.get("numFound", 0))
    except httpx.RequestError as net_err:
        logger.error("Network error: %s", net_err)
        return OpenLibrarySearchOutput(results=[], total=0, error="Network issue, please retry.")
    except httpx.HTTPStatusError as http_err:
        logger.error("HTTP error %s: %s", http_err.response.status_code, http_err)
        return OpenLibrarySearchOutput(
            results=[], total=0, error=f"Open Library returned {http_err.response.status_code}."
        )

# ----------------------------------------------------------------------
# LangChain tool wrapper
# ----------------------------------------------------------------------
@tool(
    name="OpenLibrarySearch",
    description="Searches Open Library for books matching a query and returns a concise list.",
    args_schema=OpenLibrarySearchInput,
)
async def open_library_tool(query: str, limit: int = 5) -> str:
    input_model = OpenLibrarySearchInput(query=query, limit=limit)
    result = await search_open_library(input_model)
    if result.error:
        return f"error: {result.error}"
    return result.json()

# ----------------------------------------------------------------------
# Agent orchestration
# ----------------------------------------------------------------------
async def run_agent():
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    tools = [open_library_tool]
    agent = create_openai_functions_agent(llm, tools, verbose=True)
    executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)

    user_query = (
        "Find 3 books on data science by Joel Grus and give their titles and first publish years."
    )
    response = await executor.ainvoke({"input": user_query})
    print("\n=== Agent final answer ===")
    print(response["output"])

if __name__ == "__main__":
    asyncio.run(run_agent())

Expected Output

Running the script should print something similar to:


=== Agent final answer ===
Here are three books by Joel Grus on data science:

1. "Data Science from Scratch" – First published in 2015.
2. "Python Data Science Handbook" – First published in 2016.
3. "Machine Learning for Humans" – First published in 2017.

(If the API is throttled, you may see a brief “error: Open Library returned 429” message, after which the agent retries automatically.)

Limitations and Tradeoffs

This pattern shines for stable, well‑documented services, but it has costs:

  • Over‑engineering for simple one‑offs: If you only need a single fire‑and‑forget request, the wrapper adds boilerplate and latency.
  • Prompt engineering still required: Even with a perfect schema, the LLM needs examples to know when to invoke the tool.
  • Security surface area: Exposing CLI commands means you must rigorously sanitize inputs; otherwise you open a command injection vector.
  • Network latency: Each async call still incurs round‑trip time; batching or caching may be needed for high‑throughput workloads.
  • Maintenance burden: Any upstream API change forces you to update the Pydantic model and parsing logic.

Frequently Asked Questions

How do I handle authentication for APIs that require keys?

Load secrets from environment variables or a secret manager and inject them inside the tool function. Never hard‑code them; let the tool add the appropriate Authorization header before the request.

Can I wrap a shell command the same way?

Yes. Use asyncio.create_subprocess_exec to run the command, define input and output schemas with Pydantic, and return a structured JSON result. Remember to whitelist allowed commands and escape user inputs.

What if the tool returns an empty list? Does the agent treat that as a failure?

Design the tool to differentiate “no results” from “error”. Returning an empty results array with error=null signals a valid but empty answer, allowing the agent to ask follow‑up questions instead of aborting.

How can I prevent the agent from hammering the API?

Implement a token‑bucket limiter inside the tool (e.g., using aiolimiter) and also configure the agent’s max_iterations or max_tool_calls settings. Monitoring metrics will alert you if limits are being approached.

Is there a way to version my tool definitions?

Treat each tool as a microservice: embed a version field in the Pydantic schema and expose it in the tool’s description. Deploy new versions side‑by‑side and let agents reference the version they were trained against.

What I'd Change

If I were to iterate on this design for a large‑scale production system, I’d extract the retry/metrics logic into a reusable decorator library so every new tool inherits the same observability and resilience guarantees. I’d also push the Pydantic models into a shared contract repository and generate the LangChain wrappers automatically—this eliminates drift between documentation and code and lets data‑engineers add new CLI integrations without touching the agent codebase.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.