Have you ever tried to build an AI agent that felt like it hit a wall after just one action? You give it a complex request, expecting a multi-step solution, but it only ever executes a single tool, or worse, gets stuck in an endless loop, unable to connect the dots between capabilities. In production, real-world tasks are rarely linear; they demand a sophisticated sequence of operations, where the output of one step doesn't just inform, but actively *drives* the next. This isn't just about calling a function; it's about intelligent, adaptive decision-making at each stage of a complex query. This post is for you if you're a developer grappling with orchestrating multiple domain-specific tools, managing intermediate results, and gracefully handling dynamic, real-time data from external APIs. I'll show you how to move beyond static tool definitions to create truly dynamic, adaptive workflows by architecting a robust Tool Orchestrator that empowers your agents to tackle complex, multi-step queries, using the Cloudflare blog's RSS feed as our live, external data source.
Key Takeaways
- Implementing a `Tool Orchestrator` pattern for dynamic tool selection and sequential execution.
- Leveraging LLM function calling capabilities to infer and execute chained operations based on user intent.
- Strategies for effectively managing and passing intermediate results between tools in a multi-step agent dialogue.
- Designing robust, domain-specific tools that abstract complex API interactions for seamless agent consumption.
- Techniques for handling real-time API data freshness, potential failures, and self-correction within production agent systems.
The Problem: Beyond Single-Shot Agent Actions
Our previous explorations into AI agents touched on defining tools and basic function calling. But what happens when a user asks, "Find the latest Cloudflare post about security and summarize it, highlighting any specific vulnerabilities mentioned"? This isn't a single API call. It requires: 1) fetching recent posts, 2) filtering them by a keyword, and then 3) summarizing the relevant content, potentially extracting specific details. Each step depends on the success and output of the previous one. The challenge lies in building an agent that can intelligently determine this sequence, execute it, and pass context seamlessly, all while maintaining resilience against real-world API flakiness.
Data and Sources
For this walkthrough, we'll be interacting with a live RSS feed and leveraging the power of large language models for orchestration and summarization. This provides a tangible, real-time data source to demonstrate the dynamic nature of our agent.
- Cloudflare Blog RSS feed: https://blog.cloudflare.com/rss/
- `feedparser` Python library documentation: https://pythonhosted.org/feedparser/
- OpenAI Function Calling documentation: https://platform.openai.com/docs/guides/function-calling
Data accessed on 2024-07-29
Step 1 — Defining Atomic, Composable Tools for API Interaction
The first sub-problem in building a sophisticated agent is ensuring it has well-defined, granular capabilities to interact with external systems. Think of these as the building blocks. Each tool should perform a single, focused task. This step outlines how to encapsulate API calls and data processing into atomic Python functions, making them ready for consumption by an LLM.
I design these functions with clear docstrings and type hints, which are crucial for automatically generating the function schemas that the LLM will use to understand and call them. By keeping them atomic, we maximize their reusability and simplify the LLM's task of composing them.
import feedparser
import requests
import json
from datetime import datetime, timedelta
import os
import time
# --- Tool Definitions ---
def get_cloudflare_rss_feed(feed_url: str = "https://blog.cloudflare.com/rss/") -> dict:
"""
Fetches the latest entries from the Cloudflare blog RSS feed.
Args:
feed_url: The URL of the RSS feed to fetch. Defaults to Cloudflare's blog.
Returns:
A dictionary containing the feed entries and a freshness timestamp.
Returns an empty dict and error message on failure.
"""
try:
response = requests.get(feed_url, timeout=10)
response.raise_for_status() # Raise an exception for bad status codes
feed = feedparser.parse(response.content)
if feed.bozo:
# feedparser.bozo indicates a parsing error
raise ValueError(f"RSS feed parsing error: {feed.bozo_exception}")
# Add a freshness timestamp to the data
return {
"entries": feed.entries,
"freshness_timestamp": datetime.now().isoformat()
}
except requests.exceptions.RequestException as e:
return {"error": f"Network or HTTP error fetching RSS feed: {e}", "entries": []}
except ValueError as e:
return {"error": f"Data parsing error: {e}", "entries": []}
except Exception as e:
return {"error": f"An unexpected error occurred: {e}", "entries": []}
def filter_rss_posts_by_keyword(feed_data: dict, keyword: str) -> list:
"""
Filters RSS feed entries by a specific keyword found in the title or summary.
Args:
feed_data: A dictionary containing 'entries' (list of feedparser entries)
and 'freshness_timestamp'.
keyword: The keyword to search for.
Returns:
A list of filtered feed entries.
"""
if not feed_data or "entries" not in feed_data or feed_data["error"]:
return []
filtered_entries = []
lower_keyword = keyword.lower()
for entry in feed_data["entries"]:
title = getattr(entry, 'title', '').lower()
summary = getattr(entry, 'summary', '').lower()
if lower_keyword in title or lower_keyword in summary:
filtered_entries.append({
"title": entry.title,
"link": entry.link,
"published": getattr(entry, 'published', 'N/A'),
"summary": getattr(entry, 'summary', '')
})
return filtered_entries
def get_post_summary_from_entry(entry: dict) -> str:
"""
Extracts and returns the summary of a single RSS feed entry.
Args:
entry: A dictionary representing a single filtered RSS entry.
Returns:
The summary text of the entry.
"""
return entry.get("summary", "No summary available.")
Step 2 — Building the LLM-Powered Tool Selector
With our atomic tools defined, the next sub-problem is enabling the agent to intelligently choose the *first* appropriate tool based on the initial user query. This is where the LLM's function calling capabilities shine. Instead of hardcoding decision trees, we present the LLM with a list of available tool schemas, letting it infer the user's intent and select the starting action.
I convert our Python functions into a format the LLM understands – typically a JSON schema. The LLM's response will then include a `tool_calls` object if it decides to use a tool, specifying the function name and its arguments. This is the crucial hand-off from natural language to executable code.
from openai import OpenAI
# Initialize OpenAI client (ensure OPENAI_API_KEY is set in environment)
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# Function to generate OpenAI tool schemas from Python functions
def get_tool_schemas():
tools = [
{
"type": "function",
"function": {
"name": get_cloudflare_rss_feed.__name__,
"description": get_cloudflare_rss_feed.__doc__.strip(),
"parameters": {
"type": "object",
"properties": {
"feed_url": {"type": "string", "description": "The URL of the RSS feed"}
},
"required": []
}
}
},
{
"type":
