When LLMs call external tools, how to ensure the input to those external functions is valid?
Structured responses from LLM APIs and tool calling are the two key use cases for Pydantic in the context of LLMs.
Tool calling (or function calling) is a way of connecting an external LLM to other APIs - a piece of functionality we may give the model access to in order to generate a response to a prompt.
We could give the model access to tools that:
- Get today's weather for a location (external API)
- Access account details for a given user ID
- Issue refunds for a lost orderor any other source that we would like the model to pull additional context information from when responding to a prompt.
The way this process works is illustrated on the diagram below:
When tools are available, the model examines a prompt, and then may determine that in order to follow the instructions in the prompt, it needs to call one of the tools made available to it.
When calling these tools (making an API call) data needs to be in the correct, valid format.
I’ll use Pydantic to validate data at the following stages: obtaining user input, classifying user input and dteremining if an external tool needs to be used (a function searching through the Q&A database and a function searching through the order details), obtaining a response from the tool and obtaining the final response from a second LLM API call.
# Import packages
from pydantic import BaseModel, Field, EmailStr, field_validator
from pydantic_ai import Agent
from typing import Literal, List, Optional
from datetime import datetime, date
import json
from openai import OpenAI
import anthropic
import instructor
from dotenv import load_dotenv
load_dotenv()
import nest_asyncio
nest_asyncio.apply()Define a Pydantic model using the typing library. Here, I’ll test a special field_validator function to check the format of the ‘order_id’ field using regular expressions. The order_id should be three letters followed by a hyphen and five digits.
# Define your UserInput model
class UserInput(BaseModel):
name: str = Field(..., description="User's name")
email: EmailStr = Field(..., description="User's email address")
query: str = Field(..., description="User's query")
order_id: Optional[str] = Field(
None,
description="Order ID if available (format: ABC-12345)"
)
# Validate order_id format (e.g., ABC-12345)
@field_validator("order_id")
def validate_order_id(cls, order_id):
import re
if order_id is None:
return order_id
pattern = r"^[A-Z]{3}-\d{5}$"
if not re.match(pattern, order_id):
raise ValueError(
"order_id must be in format ABC-12345 "
"(3 uppercase letters, dash, 5 digits)"
)
return order_id
purchase_date: Optional[date] = NoneDefine a CustomerQuery Pydantic model that inherits from the UserInput model - ie. it contains all the fields in the UserInput model plus some new ones.
# Define your CustomerQuery model
class CustomerQuery(UserInput):
priority: str = Field(
..., description="Priority level: low, medium, high"
)
category: Literal[
'refund_request', 'information_request', 'other'
] = Field(..., description="Query category")
is_complaint: bool = Field(
..., description="Whether this is a complaint"
)
tags: List[str] = Field(..., description="Relevant keyword tags")Data validation steps in a project where user input (question, comment, feedback) is passed to an LLM given access to tools. 
# Define a function to validate user input (1st step in the diagram above) - no LLMs involved...
def validate_user_input(user_json: str):
"""Validate user input from a JSON string and return a UserInput
instance if valid."""
try:
user_input = (
UserInput.model_validate_json(user_json)
)
print("user input validated...")
return user_input
except Exception as e:
print(f" Unexpected error: {e}")
return None# Define a function to call an LLM using Pydantic AI to create an instance of CustomerQuery. It takes user input in valid JSON format as input. It outputs an instance of CustomerQuery.
# LLM populates the additional fields in CustomerQuery based on the initial user input.
def create_customer_query(valid_user_json: str) -> CustomerQuery:
customer_query_agent = Agent(
model="google-gla:gemini-2.0-flash",
output_type=CustomerQuery,
)
response = customer_query_agent.run_sync(valid_user_json)
print("CustomerQuery generated...")
return response.outputTry out user input validation on sample data
```{python}
#| label: ex1-code
#| eval: false
# Define user input JSON data
user_input_json = '''
{
"name": "Aga Mucha",
"email": "aga.mucha@interia.pl",
"query": "When can I expect delivery of the headphones I ordered?",
"order_id": "ABC-12345",
"purchase_date": "2025-02-01"
}
'''
# Validate user input and create a CustomerQuery
valid_data = validate_user_input(user_input_json).model_dump_json()
customer_query = create_customer_query(valid_data)
print(type(customer_query))
print(customer_query.model_dump_json(indent=2))
```user input validated...
CustomerQuery generated...
<class '__main__.CustomerQuery'>
{
"name": "Aga Mucha",
"email": "aga.mucha@interia.pl",
"query": "When can I expect delivery of the headphones I ordered?",
"order_id": "ABC-12345",
"purchase_date": "2025-02-01",
"priority": "medium",
"category": "information_request",
"is_complaint": false,
"tags": [
"delivery",
"headphones"
]
}I’ll be using a Q&A search tool. It will take two arguments: query and tags (a list of tags). I’ll define a Pydantic model representing the expected format of the data.
# Define FAQ Lookup tool input as a Pydantic model
class FAQLookupArgs(BaseModel):
query: str = Field(..., description="User's query")
tags: List[str] = Field(
..., description="Relevant keyword tags from the customer query"
)I’ll also use a second tool that checks the status of an order. I’ll define a Pydantic model for that as well.
# Define Check Order Status tool input as a Pydantic model
class CheckOrderStatusArgs(BaseModel):
order_id: str = Field(
..., description="Customer's order ID (format: ABC-12345)"
)
email: EmailStr = Field(..., description="Customer's email address")
#Custom validation in a Pydantic model
@field_validator("order_id")
#This method validates the order_id format (e.g., ABC-12345) - it can be used for other things - like preventing SQL injections or XSS attacks
def validate_order_id(order_id):
import re
pattern = r"^[A-Z]{3}-\d{5}$"
if not re.match(pattern, order_id):
raise ValueError(
"order_id must be in format ABC-12345 "
"(3 uppercase letters, dash, 5 digits)"
)
return order_idI’ll create a sample Q&A database and a sample order database - a very simple Python dictionary.
# Create a sample Q&A list with the question, answer and keywords fields
sample_faq = [
{
"question": "What is the return policy?",
"answer": (
"You can return most items within 30 days of delivery "
"for a full refund."
),
"keywords": ["return", "refund", "policy"]
},
{
"question": "How long does shipping take?",
"answer": (
"Standard shipping usually takes 5-7 business days. "
"Expedited options are available at checkout."
),
"keywords": ["shipping", "delivery", "time"]
},
{
"question": "How can I track my order?",
"answer": (
"Once your order ships, you'll receive an email with "
"tracking information."
),
"keywords": ["track", "order", "tracking"]
},
{
"question": "Can I change my shipping address?",
"answer": (
"You can change your shipping address before your order "
"ships by contacting customer support."
),
"keywords": ["change", "shipping", "address"]
},
{
"question": "What payment methods do you accept?",
"answer": (
"We accept Visa, MasterCard, American Express, PayPal, "
"and Apple Pay."
),
"keywords": ["payment", "methods", "accept"]
}
]# Create a sample order database with order_id as key and status, purchase_date, estimated_delivery, email as values
sample_orders = {
"ABC-12345": {
"status": "Shipped",
"purchase_date": "2025-02-01",
"estimated_delivery": "2025-02-08",
"email": "aga.mucha@interia.pl"
},
"DEF-67890": {
"status": "Processing",
"purchase_date": "2025-02-10",
"estimated_delivery": "2025-02-15",
"email": "john.doe@example.com"
},
"GHI-54321": {
"status": "Delivered",
"purchase_date": "2025-01-25",
"estimated_delivery": "2025-01-30",
"email": "jane.smith@example.com"
},
"JKL-09876": {
"status": "Cancelled",
"purchase_date": "2025-02-05",
"estimated_delivery": "2025-02-12",
"email": "bob.johnson@example.com"
}
}Next, I’ll define the two functions that will serve as tools that the LLM will be able to call when responsing to a request (user query).
|# code-summary: Define a Q&A search tool function
# The function takes FAQLookupArgs as input (query string and tags list of strings) and returns the best matching FAQ answer as a string.
def lookup_faq_answer(args: FAQLookupArgs) -> str:
"""Look up an FAQ answer by matching tags and words in query
to FAQ entry keywords."""
#I will use a simple keyword matching approach for this example.
# query_words, tag_set, keywords are sets of lowercase words. The score will be calculated based on the length of their intersection (elements in common).
query_words = set(word.lower() for word in args.query.split())
tag_set = set(tag.lower() for tag in args.tags)
best_match = None
best_score = 0
for faq in sample_faq:
keywords = set(k.lower() for k in faq["keywords"])
score = len(keywords & tag_set) + len(keywords & query_words)
if score > best_score:
best_score = score
best_match = faq
if best_match and best_score > 0:
return best_match["answer"]
return "Sorry, I couldn't find an FAQ answer for your question."# The function takes CheckOrderStatusArgs as input (order_id string and email string) and returns a dictionary with order status information.
def check_order_status(args: CheckOrderStatusArgs):
"""Simulate checking the status of a customer's order by
order_id and email."""
order = sample_orders.get(args.order_id)
if not order:
return {
"order_id": args.order_id,
"status": "not found",
"estimated_delivery": None,
"note": "order_id not found"
}
if args.email.lower() != order.get("email", "").lower():
return {
"order_id": args.order_id,
"status": order["status"],
"estimated_delivery": order["estimated_delivery"],
"note": "order_id found but email mismatch"
}
return {
"order_id": args.order_id,
"status": order["status"],
"estimated_delivery": order["estimated_delivery"],
"note": "order_id and email match"
}In this step, I’ll define the tools for the OpenAi tool API call The documentation for the tool calling API is here
# Parameters should be passed as JSON schema. In our case, we can use the model_json_schema() method from Pydantic models defined above.
tool_definitions = [
{
"type": "function",
"function": {
"name": "lookup_faq_answer",
"description": "Look up an FAQ answer by matching tags to FAQ entry keywords.",
"parameters": FAQLookupArgs.model_json_schema()
}
},
{
"type": "function",
"function": {
"name": "check_order_status",
"description": "Check the status of a customer's order.",
"parameters": CheckOrderStatusArgs.model_json_schema()
}
}
]Next, I’ll define the format in which order details should be returned by the check_order_status function and the model for the final LLM response. The function lookup_faq_answer returns a simple string so there is no need to define a model.
# Here I'll define the format in which data should be returned by the LLM when calling a tool
#Final output Pydantic models
class OrderDetails(BaseModel):
status: str
estimated_delivery: str
note: str
#There is no model for the FAQ response - it's just a string
#This is the format of the expected response from the LLM to the initial customer query after calling a tool
class SupportTicket(CustomerQuery):
recommended_next_action: Literal[
'escalate_to_agent', 'send_faq_response',
'send_order_status', 'no_action_needed'
] = Field(
..., description="LLM's recommended next action for support"
)
order_details: Optional[OrderDetails] = Field(
None, description="Order details if action is send_order_status"
)
faq_response: Optional[str] = Field(
None, description="FAQ response if action is send_faq_response"
)
creation_date: datetime = Field(
..., description="Date and time the ticket was created"
)In the next step I create a function that accepts the structured initial customer data as a parameter. It returns a response in the format specified by the SupportTicket Pydantic model. That model contains a ‘tool_calls’ field.
This is the first LLM call. It’s response will be passed to the second LLM call together with the data returned by the ‘tools’.
# Initialize OpenAI client
client = OpenAI()
# Define a function to call OpenAI with tools
def decide_next_action_with_tools(customer_query: CustomerQuery):
#This call will not return a complete support ticket but we'll pass the schema for context so that the LLM knows what fields are expected in the final output
support_ticket_schema = json.dumps(
SupportTicket.model_json_schema(), indent=2
)
system_prompt = f"""
You are a helpful customer support agent. Your job is to
determine what support action should be taken for the customer,
based on the customer query and the expected fields in the
SupportTicket schema below. If more information on a particular
order_id or FAQ response would be helpful in responding to the
user query and can be obtained by calling a tool, call the
appropriate tool to get that information. If an order_id is
present in the query, always look up the order status to get
more information on the order.
Here is the JSON schema for the SupportTicket model you must
use as context for what information is expected:
{support_ticket_schema}
"""
messages = [
#The system prompt defines the context and expected output format
{"role": "system", "content": system_prompt},
#I'll pass the structured customer query as the main body of the prompt
{"role": "user", "content": str(customer_query.model_dump())}
]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
#Here I pass the tool definitions and specify that the model should decide when to call a tool
tools=tool_definitions,
tool_choice="auto"
)
message = response.choices[0].message
# If a tool was called, it will be in the tool_calls attribute of the message
tool_calls = getattr(message, "tool_calls", None)
#Return the full response, any tool calls made, and the full prompt
return message, tool_calls, messagesLet’s run a trial using a sample customer query!
```{python}
#| label: ex2-code
#| eval: false
# Call the decide_next_action_with_tools function
message, tool_calls, messages = decide_next_action_with_tools(
customer_query
)
# Investigate the LLM's outputs before proceeding
print("LLM message:\n", json.dumps(message.model_dump(), indent=2))
print(
"\nTool calls:\n",
json.dumps([call.model_dump() for call in tool_calls], indent=2)
)
```LLM message:
{
"content": null,
"refusal": null,
"role": "assistant",
"annotations": [],
"audio": null,
"function_call": null,
"tool_calls": [
{
"id": "call_7J3qR2lk1SpZHlqShz54k1yO",
"function": {
"arguments": "{\"order_id\":\"ABC-12345\",\"email\":\"aga.mucha@interia.pl\"}",
"name": "check_order_status"
},
"type": "function"
}
]
}
Tool calls:
[
{
"id": "call_7J3qR2lk1SpZHlqShz54k1yO",
"function": {
"arguments": "{\"order_id\":\"ABC-12345\",\"email\":\"aga.mucha@interia.pl\"}",
"name": "check_order_status"
},
"type": "function"
}
]The LLM returns a message with no content but with an attribute tool_calls which indicates which tool (function), with which parameters it would like to use.
I’ll define a function to get tool outputs - no LLMs involved here. The function just inspects existing databases.
# Define a function to get tool outputs - no LLMs involved here. The function just inspects existing databases.
def get_tool_outputs(tool_calls):
tool_outputs = []
if tool_calls:
for tool_call in tool_calls:
if tool_call.function.name == "lookup_faq_answer":
print("Agent requested a call to the Lookup FAQ tool...")
args = FAQLookupArgs.model_validate_json(
tool_call.function.arguments
)
result = lookup_faq_answer(args)
tool_outputs.append({
"tool_call_id": tool_call.id, "output": result
})
print(f"Lookup FAQ tool returned {result}")
elif tool_call.function.name == "check_order_status":
print("Agent requested a call to Check Order Status tool...")
args = CheckOrderStatusArgs.model_validate_json(
tool_call.function.arguments
)
result = check_order_status(args)
tool_outputs.append({
"tool_call_id": tool_call.id, "output": result
})
print(f"Check Order Status tool returned {result}")
return tool_outputs
tool_outputs = get_tool_outputs(tool_calls)
# Print tool outputs for inspection
print("Tool outputs:\n", json.dumps(tool_outputs, indent=2))Agent requested a call to Check Order Status tool...
Check Order Status tool returned {'order_id': 'ABC-12345', 'status': 'Shipped', 'estimated_delivery': '2025-02-08', 'note': 'order_id and email match'}
Tool outputs:
[
{
"tool_call_id": "call_7J3qR2lk1SpZHlqShz54k1yO",
"output": {
"order_id": "ABC-12345",
"status": "Shipped",
"estimated_delivery": "2025-02-08",
"note": "order_id and email match"
}
}
]Next, I’ll generate a structured support ticket using Anthropic containing relevant data from the user and from our internal databases.
# Create the Anthropic client with Instructor
anthropic_client = instructor.from_anthropic(
anthropic.Anthropic()
)
# Define a function to call Anthropic to generate a support ticket
def generate_structured_support_ticket(
#It takes as arguments a structured customer query (output of the first LLM call), the message returned by the first LLM call, and any tool outputs (from the external functions that I defined above)
customer_query: CustomerQuery, message, tool_outputs: list
):
tool_results_str = "\n".join([
f"Tool: {out['tool_call_id']} Output: {json.dumps(out['output'])}"
for out in tool_outputs
]) if tool_outputs else "No tool calls were made."
# Concatenate prompt parts into a single string for Anthropic
prompt = f"""
You are a support agent. Use all information below to
generate a support ticket as a validated Pydantic model.
Customer query: {customer_query.model_dump_json(indent=2)}
LLM message: {str(message.content)}
Tool results: {tool_results_str}
"""
# Create the message with structured output
response = anthropic_client.messages.create(
model="claude-3-7-sonnet-latest",
max_tokens=1024,
messages=[
{
"role": "user",
"content": prompt
}
],
response_model=SupportTicket
)
support_ticket = response
support_ticket.creation_date = datetime.now()
return support_ticketLet’s check if it works…
# Run the final step of generating a support ticket and print output
support_ticket = generate_structured_support_ticket(
customer_query, message, tool_outputs
)
print(support_ticket.model_dump_json(indent=2)){
"name": "Aga Mucha",
"email": "aga.mucha@interia.pl",
"query": "When can I expect delivery of the headphones I ordered?",
"order_id": "ABC-12345",
"purchase_date": "2025-02-01",
"priority": "medium",
"category": "information_request",
"is_complaint": false,
"tags": [
"delivery",
"headphones"
],
"recommended_next_action": "send_order_status",
"order_details": {
"status": "Shipped",
"estimated_delivery": "2025-02-08",
"note": "order_id and email match"
},
"faq_response": null,
"creation_date": "2025-08-27T18:17:45.646028"
}Great! This output could be passed to another agent to prepare a draft email.
Few more tests…
#Generate a few more test cases
#Test case 1 - FAQ only
user_input_json_2 = '''
{
"name": "John Doe",
"email": " john.doe@example.com",
"query": "What is your return policy?",
"order_id": null,
"purchase_date": null
}
'''
valid_data_2 = validate_user_input(user_input_json_2).model_dump_json()
customer_query_2 = create_customer_query(valid_data_2)
message_2, tool_calls_2, messages_2 = decide_next_action_with_tools(
customer_query_2
)
tool_outputs_2 = get_tool_outputs(tool_calls_2)
support_ticket_2 = generate_structured_support_ticket(
customer_query_2, message_2, tool_outputs_2
)
print(support_ticket_2.model_dump_json(indent=2))user input validated...
CustomerQuery generated...
Agent requested a call to the Lookup FAQ tool...
Lookup FAQ tool returned You can return most items within 30 days of delivery for a full refund.
{
"name": "John Doe",
"email": "john.doe@example.com",
"query": "What is your return policy?",
"order_id": null,
"purchase_date": null,
"priority": "low",
"category": "information_request",
"is_complaint": false,
"tags": [
"return policy"
],
"recommended_next_action": "send_faq_response",
"order_details": null,
"faq_response": "You can return most items within 30 days of delivery for a full refund.",
"creation_date": "2025-08-27T18:17:54.119271"
}#Test case 2 - Order status only
user_input_json_3 = '''
{
"name": "Jane Smith",
"email": " jane.smith@example.com",
"query": "What is the status of my order?",
"order_id": "DEF-67890",
"purchase_date": "2023-01-01"
}
'''
valid_data_3 = validate_user_input(user_input_json_3).model_dump_json()
customer_query_3 = create_customer_query(valid_data_3)
message_3, tool_calls_3, messages_3 = decide_next_action_with_tools(
customer_query_3
)
tool_outputs_3 = get_tool_outputs(tool_calls_3)
support_ticket_3 = generate_structured_support_ticket(
customer_query_3, message_3, tool_outputs_3
)
print(support_ticket_3.model_dump_json(indent=2))user input validated...
CustomerQuery generated...
Agent requested a call to Check Order Status tool...
Check Order Status tool returned {'order_id': 'DEF-67890', 'status': 'Processing', 'estimated_delivery': '2025-02-15', 'note': 'order_id found but email mismatch'}
{
"name": "Jane Smith",
"email": "jane.smith@example.com",
"query": "What is the status of my order?",
"order_id": "DEF-67890",
"purchase_date": "2023-01-01",
"priority": "medium",
"category": "information_request",
"is_complaint": false,
"tags": [
"order status"
],
"recommended_next_action": "send_order_status",
"order_details": {
"status": "Processing",
"estimated_delivery": "2025-02-15",
"note": "order_id found but email mismatch"
},
"faq_response": null,
"creation_date": "2025-08-27T18:19:32.519645"
}