I kinda knew I could hotwire the API SDK, getting the API call JSON body it wants to send, to be used as your batch file input. Here’s a more eloquent way than just making it die with a bad endpoint URL…
Below is a complete example—with full documentation—that demonstrates how to capture the entire API request (including complex parameters built with BaseModel) when calling the beta parsing method. In this example the SDK’s validation is performed first; only if the input is valid will the SDK build an API request, which we then intercept and convert into a JSONL‐formatted line.
- This captured JSON line includes a required custom identifier saved to the batch file, so that you can later match responses when using the Batch API. You must improve the code so that every line added would get a unique ID.
Note:
• We set max_retries=0 to disable automatic retries.
• The custom HTTP transport intercepts the outgoing HTTP call (so that no actual network call or billable API call occurs).
• If the SDK validation (for example, of your response_format BaseModel) fails, no request is built and no batch file line is produced.
Intended Use:
Instead of directly calling
completion = client.beta.chat.completions.parse(...)
you call our helper function (here named batch_create_chat_completion) which, on success, returns a JSONL string of the full API request. You can then append that line to your batch file. The final captured JSON includes the HTTP method, URL (relative), and the complete JSON body built by the SDK.
Example Code
import httpx
import json
from pydantic import BaseModel, Field
from openai import Client, DefaultHttpxClient
# =============================================================================
# 1. Define Your Complex BaseModel for Response Validation
# =============================================================================
class Vegetables(BaseModel):
vegetable_name: str = Field(..., description="Extract all vegetables from text")
is_green_color: bool = Field(..., description="Only use if green vegetable color is true")
class AIResponse(BaseModel):
response_to_user: str = Field(..., description="What the user reads")
mentioned_vegetables: list[Vegetables] | None = Field(None, description="Produces UI info links")
primary_vegetable: Vegetables
# =============================================================================
# 2. Create a Custom HTTP Transport to Intercept and Capture Requests
# =============================================================================
class CapturingTransport(httpx.BaseTransport):
"""
A custom HTTP transport that intercepts every request, captures its details,
and then raises an exception to abort the network call.
"""
def __init__(self):
self.captured_request = None
def handle_request(self, request: httpx.Request) -> httpx.Response:
# Build a dictionary with the captured details.
captured = {
"method": request.method,
# Use the relative URL (path only); no host information.
"url": request.url.raw_path.decode(),
}
try:
if request.content:
captured["body"] = json.loads(request.content.decode("utf-8"))
else:
captured["body"] = None
except Exception:
captured["body"] = request.content.decode("utf-8") if request.content else None
self.captured_request = captured
# For user feedback, print a pretty version of the captured API request.
print("=== Captured Request ===")
print("Method:", captured["method"])
print("URL:", captured["url"])
print("Body:", json.dumps(captured["body"], indent=2))
print("========================")
# Abort the actual HTTP call.
raise Exception("Aborted request in CapturingTransport to capture payload.")
# =============================================================================
# 3. Instantiate the OpenAI Client with the Custom HTTP Client
# =============================================================================
# Create an instance of our custom transport.
capturing_transport = CapturingTransport()
# Build a custom HTTP client using the DefaultHttpxClient, providing our transport.
custom_http_client = DefaultHttpxClient(transport=capturing_transport)
# IMPORTANT: Set max_retries=0 to ensure that no internal retries mask validation errors.
client = Client(http_client=custom_http_client, max_retries=0)
# =============================================================================
# 4. Define the Batch API Request Capture Function
# =============================================================================
def batch_create_chat_completion(custom_id: str, **kwargs) -> str:
"""
Captures the full API request (as built by the SDK) when calling the beta chat
completions parsing method. The function returns a single-line JSON string (JSONL)
that contains:
- custom_id: A required identifier provided by the caller.
- method: HTTP method (e.g., "POST").
- url: The relative endpoint (e.g., "/v1/chat/completions").
- body: The full JSON payload built by the SDK.
If the SDK’s validation fails (for example, due to an invalid response_format),
no request is built and an error is raised.
"""
# Reset any previously captured request.
capturing_transport.captured_request = None
try:
# Use the beta parsing method (which employs BaseModel for response_format)
# Instead of calling the API directly, our transport intercepts the outgoing request.
_ = client.beta.chat.completions.parse(**kwargs)
except Exception as e:
# If no payload was captured, assume that SDK validation failed.
if capturing_transport.captured_request is None:
raise e
# Otherwise, extract the captured request details.
captured = capturing_transport.captured_request
# Build the batch API request JSON object.
batch_request = {
"custom_id": custom_id,
"method": captured["method"],
"url": captured["url"],
"body": captured["body"]
}
# Convert the object to a single-line JSON string.
json_line = json.dumps(batch_request)
# Present the captured request prettily on the console.
print("Captured API request to be added to the batch JSONL file:")
print(json.dumps(batch_request, indent=2))
return json_line
# If no exception occurred (unexpected), signal an error.
raise Exception("Expected interception did not occur; check your SDK validation.")
# =============================================================================
# 5. Example Usage and Explanation
# =============================================================================
# This is how you would normally construct a beta chat completion request using
# a BaseModel for response_format. The SDK will validate the input and build the
# proper JSON body behind the scenes.
api_call_params = {
"model": "gpt-4o-2024-08-06",
"messages": [
{
"role": "system",
"content": (
"You are a brief gardening expert. Avoid using non-required response fields."
)
},
{
"role": "user",
"content": "What is your specialty?"
}
],
"response_format": AIResponse # Using the pydantic BaseModel for validation.
}
# Instead of calling client.beta.chat.completions.parse() directly,
# call our batch API request capture function.
try:
captured_json_line = batch_create_chat_completion("request-1", **api_call_params)
except Exception as ex:
print("Error during batch API request creation:", ex)
raise
# Append the captured JSONL line to your batch file.
if captured_json_line:
with open("batch_requests.jsonl", "a") as f:
f.write(captured_json_line + "\n")
print("Captured API request written to batch_requests.jsonl")
Documentation and Usage Notes
- What This Code Does:
-
Validation & Request Building:
The call to client.beta.chat.completions.parse() is performed with a complex parameter (a BaseModel, AIResponse, in response_format). This allows the SDK to validate and convert your input into the proper JSON schema that would be sent to the API.
-
Interception via Custom Transport:
Instead of sending the request over the network, our custom HTTP transport (CapturingTransport) intercepts it. It extracts the HTTP method, the relative URL (e.g. /v1/chat/completions), and the fully formed JSON body.
-
Batch File JSONL Format:
The function batch_create_chat_completion() then wraps these details into a JSON object that includes a custom_id (which you provide) and returns a JSON string formatted as a single line. This is what you would send to the Batch API.
-
Error Handling:
If the SDK detects an error in your input (for example, an invalid response_format), the call will fail before the HTTP request is built—and nothing will be captured. This prevents writing an invalid batch entry.
-
Console Feedback:
The console prints a prettified version (using indent=2) of the captured API request so that you know exactly what is being added to your batch file.
Yes—it can be used in batch processing if you capture the exact JSON request that the SDK constructs internally and wants to send!
In the code above, we demonstrate how to use a custom HTTP transport to intercept the outgoing API request when calling client.beta.chat.completions.parse(). This way, you get the full, validated JSON (including any BaseModel conversion) and can write it into a batch file (JSONL) that the Batch API accepts. This approach ensures that only valid and fully formed API requests (as determined by the SDK’s own validations) are submitted to the batch endpoint.
Use this sample code as a template to adapt your own requests into batch entries. The pydantic classes at the top are your “response_format” (you probably don’t want vegetable responses!). The key benefit is that you see exactly what the SDK would send on the console when running this for one API call, which is not sent, and a line is added to a file in the execution directory.
You’d need to adapt this to a more automated use if you have many inputs you want to “ask” the batch API!