Introduction to Function Calling at Scale with OpenAI Agents SDK
Function calling has evolved from a novel LLM feature into a foundational building block for production-grade AI systems. With the release of the OpenAI Agents SDK, developers now have a structured framework for orchestrating multiple tools, agents, and workflows — all powered by function calling. But when you move from a single-agent demo to a system that handles thousands of concurrent requests, dozens of tools, and complex multi-step reasoning, new challenges emerge: latency, cost, reliability, observability, and maintainability.
This guide walks through everything you need to know to build function-calling systems at scale using the OpenAI Agents SDK. We'll cover the fundamentals, architecture patterns, practical code, and the operational best practices that separate prototypes from production systems.
What Is Function Calling in the OpenAI Agents SDK?
Function calling allows a model to decide when to invoke external functions (tools) and with what arguments, rather than only generating text. The OpenAI Agents SDK builds on this primitive by providing a higher-level abstraction: Agents that have instructions, tools, guardrails, and handoff capabilities. Instead of manually parsing tool-call JSON and re-prompting the model, the SDK manages the tool execution loop for you.
At its core, the SDK handles the following loop:
- Send the conversation (including tool definitions) to the model.
- If the model returns tool calls, execute them in parallel when possible.
- Append tool outputs back into the conversation.
- Re-prompt the model until it produces a final response or hands off to another agent.
This abstraction is powerful, but at scale you need to understand what happens underneath to tune performance, control costs, and avoid silent failures.
Why Function Calling at Scale Matters
Building a single agent that calls one or two tools is straightforward. Scaling introduces a different class of problems:
- Tool explosion: As capabilities grow, you may have 20, 50, or 100+ tools. Token costs for tool schemas grow linearly, and model accuracy degrades when too many similar tools compete for selection.
- Concurrency: Handling hundreds of simultaneous user sessions means managing rate limits, connection pools, and async tool execution carefully.
- Reliability: Tools fail. APIs time out. The model occasionally produces malformed arguments. Your system must degrade gracefully.
- Cost: Each round-trip with a large tool schema is expensive. Inefficient loops can multiply costs 5-10x.
- Observability: When something goes wrong in a multi-agent, multi-tool trace, you need structured logging to find the root cause.
The Agents SDK addresses several of these concerns directly, but the architecture decisions you make around it are what determine whether your system scales.
Setting Up the OpenAI Agents SDK
First, install the SDK and configure your environment. The Agents SDK is available as a Python package.
pip install openai-agents python-dotenv
Create a .env file with your API key:
OPENAI_API_KEY=sk-your-key-here
Now let's set up a basic project structure:
from agents import Agent, Runner, function_tool
import os
from dotenv import load_dotenv
load_dotenv()
# Verify configuration
if not os.getenv("OPENAI_API_KEY"):
raise RuntimeError("OPENAI_API_KEY is not set")
Defining Tools with function_tool
The SDK provides the @function_tool decorator, which converts a regular Python function into a tool the agent can call. The function's docstring and type hints become the tool's schema automatically.
from agents import function_tool
@function_tool
def get_weather(city: str, units: str = "metric") -> dict:
"""Get the current weather for a given city.
Args:
city: The name of the city, e.g. "San Francisco".
units: Temperature units - "metric" for Celsius, "imperial" for Fahrenheit.
"""
# In production, call a real weather API here
return {
"city": city,
"temperature": 18 if units == "metric" else 64,
"conditions": "partly cloudy",
"units": units,
}
@function_tool
def search_knowledge_base(query: str, top_k: int = 5) -> list[dict]:
"""Search the internal knowledge base for relevant documents.
Args:
query: Natural language search query.
top_k: Maximum number of results to return.
"""
# Placeholder for vector search
return [
{"id": "doc-001", "title": "Return Policy", "snippet": "Items can be returned within 30 days..."},
{"id": "doc-002", "title": "Shipping FAQ", "snippet": "Standard shipping takes 3-5 business days..."},
]
The key principle here: write clear, specific docstrings. The model uses these descriptions to decide when and how to call each tool. Vague docstrings are the most common cause of poor tool selection.
Creating Your First Agent
With tools defined, creating an agent is concise:
from agents import Agent, Runner
customer_support_agent = Agent(
name="CustomerSupportAgent",
instructions=(
"You are a helpful customer support agent. "
"Use the available tools to answer questions about weather "
"and internal policies. Always cite which tool you used. "
"If a tool fails, apologize and suggest the user contact human support."
),
tools=[get_weather, search_knowledge_base],
model="gpt-4o-mini",
)
async def main():
result = await Runner.run(
customer_support_agent,
"What's the weather in Berlin and what's your return policy?"
)
print(result.final_output)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
The Runner handles the entire tool-call loop. When the model decides to call both get_weather and search_knowledge_base, the SDK executes them in parallel and feeds results back automatically.
Scaling Tools: The RAG-Pattern Alternative
When you have dozens of tools, passing all schemas to the model on every call becomes expensive and hurts accuracy. A common pattern at scale is tool retrieval: instead of giving the model all tools, dynamically select a relevant subset based on the user's query.
from agents import Agent, Runner, function_tool
from openai import OpenAI
client = OpenAI()
# A registry of all available tools, keyed by name
TOOL_REGISTRY = {
"get_weather": get_weather,
"search_knowledge_base": search_knowledge_base,
# ... potentially 50+ more tools
}
TOOL_DESCRIPTIONS = {
"get_weather": "Get current weather for a city.",
"search_knowledge_base": "Search internal documents by query.",
# ... one line per tool
}
def select_relevant_tools(user_query: str, max_tools: int = 5) -> list:
"""Use embeddings to pick the most relevant tools for a query."""
import numpy as np
# Pre-compute these embeddings offline in production
tool_names = list(TOOL_DESCRIPTIONS.keys())
tool_texts = [f"{name}: {desc}" for name, desc in TOOL_DESCRIPTIONS.items()]
response = client.embeddings.create(
input=tool_texts + [user_query],
model="text-embedding-3-small",
)
embeddings = [item.embedding for item in response.data]
tool_embeddings = np.array(embeddings[:-1])
query_embedding = np.array(embeddings[-1])
# Cosine similarity
scores = tool_embeddings @ query_embedding
top_indices = np.argsort(scores)[-max_tools:][::-1]
return [TOOL_REGISTRY[tool_names[i]] for i in top_indices]
async def run_with_dynamic_tools(query: str):
relevant_tools = select_relevant_tools(query, max_tools=5)
agent = Agent(
name="DynamicAgent",
instructions="You are a helpful assistant. Use the provided tools.",
tools=relevant_tools,
model="gpt-4o-mini",
)
return await Runner.run(agent, query)
This pattern keeps token usage low and tool-selection accuracy high, even with hundreds of available tools.
Multi-Agent Handoffs for Complex Workflows
Another scaling strategy is to split responsibilities across specialized agents and let them hand off to each other. The SDK supports this natively via the handoffs parameter.
from agents import Agent, Runner, function_tool
@function_tool
def process_refund(order_id: str, amount: float) -> dict:
"""Process a refund for a given order."""
return {"order_id": order_id, "refunded": amount, "status": "completed"}
@function_tool
def escalate_to_human(reason: str, priority: str = "normal") -> dict:
"""Escalate the conversation to a human agent."""
return {"ticket_id": "TKT-9921", "priority": priority, "reason": reason}
# Specialized agents
billing_agent = Agent(
name="BillingAgent",
instructions=(
"You handle billing and refund requests. "
"Use process_refund for approved refunds. "
"If the customer is upset or the issue is complex, escalate to human."
),
tools=[process_refund, escalate_to_human],
model="gpt-4o-mini",
)
general_agent = Agent(
name="GeneralSupportAgent",
instructions=(
"You are the first point of contact for customer support. "
"Answer general questions using available tools. "
"Hand off to BillingAgent for any billing or refund matters."
),
tools=[get_weather, search_knowledge_base],
handoffs=[billing_agent],
model="gpt-4o-mini",
)
async def handle_request(message: str):
result = await Runner.run(general_agent, message)
print(result.final_output)
return result
When the general agent detects a billing issue, it hands off the conversation to the billing agent, which then has its own specialized tools. This keeps each agent's context small and focused.
Handling Concurrency at Scale
For production workloads, you'll run many agent sessions concurrently. The SDK is async-native, which makes this straightforward, but you need to manage rate limits and resource pools.
import asyncio
from agents import Agent, Runner
from openai import RateLimitError
# Semaphore to limit concurrent API calls
MAX_CONCURRENT = 20
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def run_agent_with_backoff(agent: Agent, message: str, max_retries: int = 3):
"""Run an agent with exponential backoff on rate limit errors."""
for attempt in range(max_retries):
async with semaphore:
try:
result = await Runner.run(agent, message)
return result
except RateLimitError:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt
print(f"Rate limited, retrying in {wait_time}s...")
await asyncio.sleep(wait_time)
async def batch_process(messages: list[str], agent: Agent):
"""Process a batch of user messages concurrently."""
tasks = [run_agent_with_backoff(agent, msg) for msg in messages]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = []
failures = []
for msg, result in zip(messages, results):
if isinstance(result, Exception):
failures.append((msg, result))
else:
successes.append((msg, result.final_output))
print(f"Success: {len(successes)}, Failures: {len(failures)}")
return successes, failures
The semaphore caps concurrency to stay within rate limits, while the backoff logic handles transient failures gracefully.
Adding Guardrails for Safety
At scale, you cannot manually review every input and output. The Agents SDK supports guardrails — validation functions that run before or after the agent and can halt execution.
from agents import Agent, Runner, GuardrailFunctionOutput, input_guardrail
@input_guardrail
async def block_pii(agent: Agent, input_text: str) -> GuardrailFunctionOutput:
"""Block messages containing social security numbers or credit cards."""
import re
ssn_pattern = r"\b\d{3}-\d{2}-\d{4}\b"
cc_pattern = r"\b(?:\d[ -]*?){13,16}\b"
if re.search(ssn_pattern, input_text) or re.search(cc_pattern, input_text):
return GuardrailFunctionOutput(
output_info={"reason": "PII detected"},
tripwire_triggered=True,
)
return GuardrailFunctionOutput(
output_info={"reason": "clean"},
tripwire_triggered=False,
)
safe_agent = Agent(
name="SafeAgent",
instructions="You are a helpful assistant.",
tools=[get_weather],
input_guardrails=[block_pii],
model="gpt-4o-mini",
)
When the guardrail's tripwire is triggered, the SDK stops execution before the model is called, saving cost and preventing sensitive data from reaching the API.
Observability and Tracing
The Agents SDK includes built-in tracing. Every run produces a trace that captures each LLM call, tool invocation, and handoff. This is essential for debugging at scale.
from agents import Runner, trace
async def traced_run(agent, message: str):
with trace("Customer Support Session"):
result = await Runner.run(agent, message)
return result
# Traces are automatically sent to the OpenAI dashboard.
# You can also add custom spans for your own tool logic:
from agents import custom_span
@function_tool
async def slow_database_query(sql: str) -> dict:
"""Execute a database query."""
with custom_span("db_query", data={"sql": sql[:100]}):
# Simulate slow query
await asyncio.sleep(2)
return {"rows": 42, "status": "ok"}
For production, export traces to your existing observability stack (Datadog, Honeycomb, Langfuse) so you can correlate agent behavior with system metrics.
Best Practices for Production
1. Keep Tool Schemas Lean
Every tool definition consumes tokens on every call. Use concise but unambiguous names and descriptions. Avoid optional parameters unless truly needed — they increase the chance of the model making poor choices.
2. Validate Tool Arguments Server-Side
Never trust model-generated arguments blindly. The model can produce invalid values, especially under heavy load or with ambiguous prompts.
from pydantic import BaseModel, field_validator
class WeatherArgs(BaseModel):
city: str
units: str = "metric"
@field_validator("units")
@classmethod
def validate_units(cls, v):
if v not in ("metric", "imperial"):
raise ValueError("units must be 'metric' or 'imperial'")
return v
@function_tool
def get_weather_validated(city: str, units: str = "metric") -> dict:
"""Get current weather for a city."""
args = WeatherArgs(city=city, units=units) # Validates inputs
# ... proceed with validated args
return {"city": args.city, "units": args.units, "temperature": 18}
3. Use the Right Model for Each Agent
Not every agent needs GPT-4o. Use gpt-4o-mini for routing, classification, and simple tool selection. Reserve larger models for complex reasoning. This single change can cut costs by 70% or more.
4. Cache Tool Results
Many tool calls are idempotent. Cache results to avoid redundant API calls and reduce latency.
from functools import lru_cache
import time
# Cache weather for 10 minutes
@function_tool
def get_weather_cached(city: str, units: str = "metric") -> dict:
"""Get current weather for a city."""
return _fetch_weather(city, units)
@lru_cache(maxsize=1000)
def _fetch_weather(city: str, units: str) -> dict:
# Real API call here
return {"city": city, "temperature": 18, "units": units, "cached_at": time.time()}
5. Set Timeouts on Every Tool
A single slow tool can block an entire agent loop. Always enforce timeouts.
import asyncio
@function_tool
async def fetch_external_api(endpoint: str) -> dict:
"""Fetch data from an external API."""
try:
async with asyncio.timeout(5.0):
# Simulated HTTP call
await asyncio.sleep(1)
return {"data": "result"}
except asyncio.TimeoutError:
return {"error": "External API timed out after 5 seconds"}
6. Structure Your Tool Organization
For large systems, organize tools into modules and group them by domain. This makes maintenance easier and supports the dynamic tool-selection pattern described earlier.
# tools/billing.py
@function_tool
def process_refund(order_id: str, amount: float) -> dict: ...
@function_tool
def get_invoice(invoice_id: str) -> dict: ...
# tools/shipping.py
@function_tool
def track_shipment(tracking_number: str) -> dict: ...
@function_tool
def estimate_shipping(origin: str, destination: str) -> dict: ...
# tools/__init__.py
from .billing import process_refund, get_invoice
from .shipping import track_shipment, estimate_shipping
ALL_TOOLS = [process_refund, get_invoice, track_shipment, estimate_shipping]
Putting It All Together: A Complete Example
Here is a complete, production-oriented example that combines dynamic tool selection, multi-agent handoffs, guardrails, concurrency control, and error handling:
import asyncio
import os
from agents import (
Agent, Runner, function_tool,
input_guardrail, GuardrailFunctionOutput,
trace, custom_span,
)
from dotenv import load_dotenv
load_dotenv()
# ---- Tools ----
@function_tool
def get_weather(city: str, units: str = "metric") -> dict:
"""Get current weather for a city."""
return {"city": city, "temperature": 18, "units": units}
@function_tool
def search_kb(query: str, top_k: int = 5) -> list:
"""Search the internal knowledge base."""
return [{"id": "1", "snippet": f"Result for: {query}"}]
@function_tool
def process_refund(order_id: str, amount: float) -> dict:
"""Process a refund for an order."""
return {"order_id": order_id, "refunded": amount, "status": "done"}
@function_tool
def escalate_to_human(reason: str) -> dict:
"""Escalate to a human agent."""
return {"ticket": "TKT-100", "reason": reason}
# ---- Guardrails ----
@input_guardrail
async def no_pii(agent: Agent, text: str) -> GuardrailFunctionOutput:
import re
if re.search(r"\b\d{3}-\d{2}-\d{4}\b", text):
return GuardrailFunctionOutput(
output_info={"reason": "SSN detected"},
tripwire_triggered=True,
)
return GuardrailFunctionOutput(
output_info={"reason": "clean"},
tripwire_triggered=False,
)
# ---- Agents ----
billing_agent = Agent(
name="BillingAgent",
instructions="Handle billing and refunds. Escalate complex cases to human.",
tools=[process_refund, escalate_to_human],
model="gpt-4o-mini",
)
support_agent = Agent(
name="SupportAgent",
instructions=(
"You are a customer support agent. "
"Answer questions using tools. "
"Hand off billing matters to BillingAgent."
),
tools=[get_weather, search_kb],
handoffs=[billing_agent],
input_guardrails=[no_pii],
model="gpt-4o-mini",
)
# ---- Concurrency wrapper ----
MAX_CONCURRENT = 20
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def run_safe(agent: Agent, message: str, retries: int = 3):
for attempt in range(retries):
async with semaphore:
try:
with trace("support_session"):
result = await Runner.run(agent, message)
return {"ok": True, "output": result.final_output}
except Exception as e:
if attempt == retries - 1:
return {"ok": False, "error": str(e)}
await asyncio.sleep(2 ** attempt)
# ---- Batch runner ----
async def main():
messages = [
"What's the weather in Tokyo?",
"I need a refund for order #12345, amount $49.99",
"What is your return policy?",
"My SSN is 123-45-6789, can you help?", # Should be blocked
]
tasks = [run_safe(support_agent, msg) for msg in messages]
results = await asyncio.gather(*tasks)
for msg, res in zip(messages, results):
status = "OK" if res["ok"] else "FAIL"
print(f"[{status}] {msg[:50]}...")
if res["ok"]:
print(f" -> {res['output'][:100]}")
else:
print(f" -> Error: {res['error']}")
if __name__ == "__main__":
asyncio.run(main())
Conclusion
Function calling at scale with the OpenAI Agents SDK is about more than wiring up tools — it's about architecting systems that remain fast, affordable, and reliable as complexity grows. By leveraging dynamic tool selection to keep schemas lean, splitting responsibilities across specialized agents with handoffs, enforcing guardrails for safety, wrapping every tool in timeouts and validation, and using the built-in tracing for observability, you can build agent systems that handle real production traffic with confidence. The SDK gives you the primitives; the patterns in this guide give you the architecture. Start simple, measure relentlessly, and scale deliberately — the difference between a demo and a production system lives in the operational details.