¶How do we make a complete API request without hard-coding credentials?
A study assistant that works only while an API key is pasted into a notebook is not yet an application component. This chapter turns that first model call into a Python boundary you can test, validate, and inspect without exposing credentials or paying for every test run.
I start with a non-streaming HTTP request rather than an SDK. For this first call, seeing the URL, headers, request body, and response makes failures easier to locate. An SDK becomes useful later; it does not remove your responsibility to understand what crosses the network.
¶Separate configuration from the request
The example uses OpenAI’s /v1/chat/completions endpoint. Its request contains a model identifier, a message list, and an output limit. The Authorization header carries the credential; it does not belong in the prompt. The Content-Type header tells the server that the body is JSON.
Save the following as first_call.py. It uses only Python’s standard library. Running it normally selects a deterministic fixture: a fixed response shaped like the relevant portion of the API response, with no network access and no bill. Passing --live explicitly permits a paid request.
import argparse
import json
import os
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
ENDPOINT = "https://api.openai.com/v1/chat/completions"
def required_environment(name):
value = os.environ.get(name, "").strip()
if not value:
raise RuntimeError(f"Missing environment variable: {name}")
return value
def ask_study_assistant(question, *, live=False):
if not live:
return {
"choices": [{
"message": {
"role": "assistant",
"content": (
"A deadlock occurs when tasks wait indefinitely "
"for resources held by one another."
),
},
"finish_reason": "stop",
}],
"usage": {
"prompt_tokens": 40,
"completion_tokens": 18,
"total_tokens": 58,
},
}
api_key = required_environment("OPENAI_API_KEY")
model = required_environment("OPENAI_MODEL")
payload = {
"model": model,
"messages": [
{
"role": "system",
"content": (
"You are a study assistant. Explain the requested "
"concept briefly. Do not claim to have read course notes."
),
},
{"role": "user", "content": question},
],
"max_tokens": 128,
}
request = Request(
ENDPOINT,
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urlopen(request, timeout=30) as response:
return json.load(response)
except HTTPError as exc:
raise RuntimeError(f"Model API returned HTTP {exc.code}") from None
except (URLError, TimeoutError):
raise RuntimeError("Model API connection failed or timed out") from None
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--live", action="store_true")
args = parser.parse_args()
response = ask_study_assistant("What is a deadlock?", live=args.live)
print(response["choices"][0]["message"]["content"])
print(json.dumps(response["usage"], sort_keys=True))
if __name__ == "__main__":
main()
Run python first_call.py first. The 40 input tokens and 18 output tokens are synthetic fixture values, not measured usage. This fixture tests local execution and response handling; it cannot prove that a credential, model, or remote endpoint works.
For a live run, create an API credential in your provider account, confirm that API billing is available, and configure OPENAI_API_KEY and OPENAI_MODEL in your process environment through your terminal or IDE’s private run configuration. A chat-product subscription does not necessarily include API credit. Use a model your account can access that supports this endpoint and max_tokens; gpt-4o-mini is one such model, provided it remains available to your account. Then run python first_call.py --live.
I keep the model explicit rather than silently choosing one. Model availability, supported parameters, and prices change independently of your Python code. If another model requires a different output-limit parameter, change the adapter deliberately instead of assuming API compatibility.
¶Know what the boundary does—and does not—guarantee
An environment variable keeps the key out of this source file; it is not a secret vault. Do not commit an IDE configuration containing it, print os.environ, or embed it in browser code. A .env file is not loaded automatically by this example. If a key reaches Git history, revoke it; deleting the visible line does not invalidate the credential.
A missing setting raises Missing environment variable: OPENAI_API_KEY before sending anything. HTTP 401 usually points to authentication; HTTP 429 can indicate throttling or exhausted quota. Neither means “rewrite the prompt.” The 30-second timeout bounds socket operations, not a guaranteed total request deadline. A timeout also does not prove that the provider stopped processing, so this example does not automatically retry.
The 128-token limit constrains generated output, not the entire bill. Input tokens are chargeable too; check current input and output rates before enabling live mode. Provider budget alerts may not be hard spending caps. The returned usage object is evidence for later cost accounting, whereas fixture counts must never enter a real spending report.
Try removing OPENAI_MODEL and running live mode: configuration should fail before network access. Then consider the interview question hidden in this exercise: if the server returns HTTP 200, what proves that the response contains an answer your application can actually use?
¶Which message roles, instructions, and examples improve the assistant's answers?
A successful HTTP request can still produce an unusable study answer: six paragraphs where the interface expects two sentences, a missing practice question, or a confident explanation of the wrong topic. The transport worked. The application contract did not. I start with a short, testable instruction rather than a theatrical persona such as “the world’s greatest professor.” A persona suggests a style; it does not specify what your interface needs.
¶Separate application instructions from the current question
In a chat-style request, each message has a role and content. The role identifies the message’s function, not just its speaker. A system message commonly carries application-wide instructions; a user message carries the current request; an assistant message represents an answer, including a demonstration you supply. Some APIs also expose a developer role for application instructions. Use the roles your selected endpoint documents: unsupported roles can cause request validation errors rather than better answers.
Do not assume every provider implements an identical instruction hierarchy. For an endpoint that supports it, put stable application requirements in its designated instruction channel, and keep the changing question separate. Writing “SYSTEM:” inside ordinary user text does not create a system message. Nor does placing instructions first guarantee compliance; these are model inputs, not executable assertions.
For the study assistant, I would start with three requirements: explain the concept briefly, include one check question, and request clarification when the subject is ambiguous. Avoid “always answer,” which conflicts with the clarification rule. Also distinguish your preferred length from a hard limit: “two sentences” is a generation instruction, not an enforced field constraint. Output validation belongs in the next section.
¶Demonstrate the behavior that wording leaves ambiguous
The following Python fixture builds a request payload without credentials, network access, or charges. It uses one demonstration: a user question followed by an assistant answer. This is the few-shot mechanism from Chapter 5 applied to an application contract, not a training step. Pass the resulting messages through the client established in the preceding section only if its endpoint accepts this message format.
import json
INSTRUCTIONS = (
"You are a study assistant for introductory computer science. "
"For a clear concept question, write an Explanation: label followed "
"by at most two explanatory sentences, then a Check: label followed "
"by one practice question. Use plain language. "
"If the subject is ambiguous, ask one clarification question instead "
"of using those labels."
)
def build_messages(question: str) -> list[dict[str, str]]:
if not question.strip():
raise ValueError("question must not be blank")
return [
{"role": "system", "content": INSTRUCTIONS},
{"role": "user", "content": "What is a stack in data structures?"},
{
"role": "assistant",
"content": (
"Explanation: A stack stores items so that the last item "
"added is the first removed. A stack of plates is a useful "
"analogy.\n"
"Check: If you push A and then B, which item pops first?"
),
},
{"role": "user", "content": question},
]
messages = build_messages("What is a queue in data structures?")
assert len(messages) == 4
assert messages[-1]["role"] == "user"
print(json.dumps(messages, indent=2))
The final user message asks about a queue, not a stack. That distinction matters: the demonstration should teach the answer pattern without accidentally supplying the answer to the test question. The fixture’s assertions verify four-message assembly only. They do not prove that a model will obey the instruction, explain queues correctly, or avoid copying the stack analogy.
I add examples only when a specific failure justifies their token cost. Start with 0 examples, compare against 1, and try 2 only if a second behavior needs demonstration. Examples occupy input tokens on every request unless an applicable provider mechanism changes their accounting. A long example can also teach unwanted verbosity. More demonstrations are not automatically better instructions.
¶Test the exceptions, not just the happy path
Use three fixed questions for a small experiment: “What is a queue in data structures?”, “Explain Java,” and “What is a stack in data structures?” Compare instructions alone against instructions plus the demonstration. Check correctness, the two-sentence limit, the practice question, and whether Java triggers clarification. These are illustrative test cases, not measured results; I cannot promise an improvement without testing your chosen model.
A common failure is format imitation without correct content: the labels appear, but the queue explanation describes last-in-first-out behavior. Another is over-clarification: the model asks what “queue” means despite the explicit subject. Record both as failures rather than rewarding neat formatting.
For interview practice, ask: “Why not put everything in one user message?” A useful answer is that separate roles express application intent and preserve message structure, while explicit instructions and examples make behavior testable; neither guarantees correctness. Now ask the engineering question that follows: what should your Python component reject when the response still violates that contract?
¶Why does valid JSON still need schema and application-level validation?
A model can return perfectly valid JSON that crashes your study assistant. {"answer": null} parses successfully, but calling answer.strip() on that value raises AttributeError. An equally valid response might contain an answer for the wrong course. That one may never throw an exception. It simply shows the student the wrong material.
I treat model output as an external service response, not as a Python object that deserves trust because the prompt requested a format. Instructions influence generation; validation determines what your application accepts. Keep those responsibilities separate, even when a provider offers schema-constrained output.
¶Three checks, three different promises
Parsing asks whether the response can be read as JSON. Python’s json.loads() raises JSONDecodeError for malformed syntax, such as an unfinished object. However, a JSON array, string, or number can also parse successfully. Successful parsing does not promise the object shape your interface expects.
Schema validation checks the declared structure: required fields, types, allowed values, and limits. For this illustrative study-assistant contract, require exactly 3 fields: answer, course_id, and status. Allow 2 status values, answered and needs_clarification, and cap the answer at 1,200 characters. This is a UI-oriented character limit, not a token budget.
Application-level validation checks whether that structurally acceptable object makes sense for the current request. The returned course_id must match the selected course. An answered response must contain non-whitespace text; a needs_clarification response must have an empty answer because this particular interface displays its own clarification control.
Some validators can express those cross-field rules inside a schema. The useful distinction is responsibility, not which library executes the check: the response shape belongs to the contract, while the selected course comes from application state.
¶A deterministic boundary you can test
I would start with this dependency-free fixture before connecting another live call. It implements a small schema directly rather than claiming to be a general JSON Schema validator. Save it as response_contract.py and run it with Python. All 4 inputs are local; no credentials or paid requests are involved.
import json
class ContractError(ValueError):
pass
def validate_response(raw: str, expected_course_id: str) -> dict:
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise ContractError("syntax: invalid JSON") from exc
required = {"answer", "course_id", "status"}
if not isinstance(data, dict) or set(data) != required:
raise ContractError("schema: expected exactly answer, course_id, status")
if any(not isinstance(data[key], str) for key in required):
raise ContractError("schema: all fields must be strings")
if data["status"] not in {"answered", "needs_clarification"}:
raise ContractError("schema: unsupported status")
if len(data["answer"]) > 1200:
raise ContractError("schema: answer exceeds 1200 characters")
if data["course_id"] != expected_course_id:
raise ContractError("application: course_id does not match request")
if data["status"] == "answered" and not data["answer"].strip():
raise ContractError("application: answered requires nonblank text")
if data["status"] == "needs_clarification" and data["answer"] != "":
raise ContractError("application: clarification requires empty answer")
return data
fixtures = [
("accepted", json.dumps({
"answer": "A stack removes the most recently added item first.",
"course_id": "CS201",
"status": "answered",
})),
("broken syntax", '{"answer":'),
("wrong type", json.dumps({
"answer": None, "course_id": "CS201", "status": "answered",
})),
("wrong course", json.dumps({
"answer": "A stack uses LIFO.",
"course_id": "CS301",
"status": "answered",
})),
]
for label, raw in fixtures:
try:
result = validate_response(raw, expected_course_id="CS201")
print(f"{label}: accepted ({result['status']})")
except ContractError as exc:
print(f"{label}: rejected: {exc}")
The wrong-type fixture fails with schema: all fields must be strings; it never reaches rendering. The wrong-course fixture passes the structural checks but fails with application: course_id does not match request. These separate failures tell you whether to inspect the response format or the request-specific rule.
I reject unexpected fields rather than silently discarding them here. Unless your contract explicitly permits extensions, accepting an extra field can conceal a misspelled field or an unreviewed change. For larger contracts, I would use a maintained validation library; hand-written checks become expensive once nested objects and optional branches multiply.
¶What validation cannot promise
A schema-constrained generation feature is worth using when the selected API and model support your required constraints. It reduces format failures. It does not establish factual correctness, and a schema containing course_id cannot establish course membership unless the allowed value is bound to the current request. Refusals and incomplete responses also need handling before you assume there is a complete object to validate.
Interview question: “Does validated JSON make an LLM response safe to display?” My answer is: it makes the response compatible with a defined contract, provided the validator covers that contract. It does not prove that the answer is true. Later retrieval and evaluation chapters address that separate obligation.
Exercise: add a fifth fixture with status set to answered and an answer containing only spaces. Then consider the operational question: once this boundary rejects a live response, should your client retry, ask for clarification, or report a failure?
¶How should Python handle timeouts, rate limits, retries, and streaming responses?
A request can fail after the provider has started generating—and possibly charging for—an answer. Your Python process sees a timeout, but that does not prove the server stopped. Retrying may create a second billable generation. I treat retries as a limited recovery budget, not a promise that every question eventually receives an answer.
For the study assistant, I would start with a 12-second overall deadline and at most 3 attempts, including the original request. These are illustrative application settings, not provider defaults. Increase the deadline if your chosen model and answer length require it. A connection timeout limits connection establishment; a read timeout limits waiting for incoming data. Neither necessarily limits the complete operation: a stream delivering small chunks regularly can remain open far longer. Use time.monotonic() for elapsed-time calculations because wall-clock adjustments must not extend the deadline.
¶Retry the failure, not every exception
I would normally retry connection failures, timeouts, HTTP 429 throttling, and selected transient server failures such as 502, 503, and 504, provided the deadline permits another attempt. Inspect the provider’s error body: a 429 caused by exhausted account credit may require intervention rather than waiting. Do not retry 401 authentication failures, malformed requests returning 400, or the previous section’s validation failures through this same loop. Waiting cannot repair a missing credential or an invalid request.
Exponential backoff increases the waiting window after each failure; jitter randomizes the actual wait so clients do not all retry together. With a 0.5-second base, the first 2 retry windows are 0–0.5 and 0–1.0 seconds. If the server supplies Retry-After, respect it unless it exceeds your remaining deadline; then fail rather than retry early. That header may contain seconds or an HTTP date, so the adapter must normalize it before calling the policy below.
This account-free fixture tests the policy without pretending to be a provider SDK. send is your adapter function: it receives the remaining budget, configures transport timeouts, and translates retryable provider exceptions into TransientError. Other exceptions escape unchanged. The virtual clock makes the test deterministic and avoids sleeping.
import random
import time
class TransientError(Exception):
def __init__(self, message, retry_after=None):
super().__init__(message)
self.retry_after = retry_after
def request_with_retry(
send, *, budget=12.0, attempts=3,
clock=time.monotonic, sleep=time.sleep, rng=None
):
if budget <= 0 or attempts < 1:
raise ValueError("positive budget and attempts required")
rng = rng if rng is not None else random.Random()
deadline = clock() + budget
for attempt in range(attempts):
remaining = deadline - clock()
if remaining <= 0:
raise TimeoutError("request deadline exceeded")
try:
result = send(remaining)
if clock() >= deadline:
raise TimeoutError("response arrived after deadline")
return result
except TransientError as error:
if attempt == attempts - 1:
raise
delay = rng.uniform(0.0, min(4.0, 0.5 * 2 ** attempt))
if error.retry_after is not None:
delay = max(delay, error.retry_after)
if delay >= deadline - clock():
raise TimeoutError("retry exceeds deadline") from error
sleep(delay)
class VirtualClock:
def __init__(self):
self.now = 0.0
self.waits = []
def monotonic(self):
return self.now
def sleep(self, seconds):
self.waits.append(seconds)
self.now += seconds
clock = VirtualClock()
outcomes = iter([
TransientError("HTTP 429: throttled", retry_after=2.0),
{"answer": "Attention weights combine value vectors."},
])
budgets = []
def fixture_send(remaining):
budgets.append(remaining)
outcome = next(outcomes)
if isinstance(outcome, Exception):
raise outcome
return outcome
answer = request_with_retry(
fixture_send,
clock=clock.monotonic,
sleep=clock.sleep,
rng=random.Random(7),
)
assert clock.waits == [2.0]
assert budgets == [12.0, 10.0]
assert answer["answer"].startswith("Attention")
print(answer)
The wrapper cannot interrupt a blocked send; the adapter must enforce its supplied budget, using cancellation where necessary. Rejecting a late result is not cancellation. I also disable SDK retries when this application loop owns retry policy. Otherwise, 3 application attempts multiplied by 3 SDK attempts can produce 9 network attempts. Check your pinned SDK’s settings rather than assuming its defaults.
¶Streaming changes what failure means
Streaming lets the student read before generation finishes; it does not inherently reduce total generation time. Treat text deltas as fragments, not complete tokens, JSON objects, or validated answers. Buffer them while showing a provisional preview. Only publish the study assistant’s structured result after the provider’s successful completion signal and the established validation checks.
A dropped connection after “The answer is” is a failed response, even if the iterator ends without a useful exception. Check the provider’s documented terminal event or finish status: end-of-file alone is not evidence of success. I would not automatically restart after displaying partial text, because the replacement may contradict the visible prefix. Mark the preview incomplete and offer an explicit retry. Close the stream on cancellation or failure, using its context manager or documented cleanup method, so abandoned connections do not linger.
Test two failures next: a Retry-After of 20 seconds against the 12-second budget, and a stream disconnected before completion. The first must stop without another attempt; the second must never reach validated-answer storage. Once those boundaries hold, what should the client record about attempts, elapsed time, and potentially billable work?
¶Can a prompt experiment improve quality without hiding latency and token cost?
A longer prompt can produce a better answer and still make the study assistant worse to use. If correctness rises from 1 passing case to 2, but the response takes longer and costs more, you need all three results on the same page. I would not accept “the answers look better” as evidence for changing a production prompt.
Start with one controlled change: add an instruction to check arithmetic before answering. Keep the model, questions, sampling settings, response schema, and output limit unchanged. Otherwise, you cannot tell which change caused the difference. Give each prompt a stable identifier, such as baseline and checked, and save the actual prompt text alongside the results—not just its nickname.
¶A fixture that tests the experiment itself
The following experiment runs without credentials or network access. Its responses, token counts, and durations are invented fixtures, not model measurements. The deliberately narrow questions let exact answers serve as checks; this is not a suitable scoring method for explanations of attention or gradient descent. I prefer this small harness first because a broken denominator can make any prompt look successful.
import json
from decimal import Decimal
PROMPTS = {
"baseline": 'Return JSON with one string field named "answer".',
"checked": (
'Check the arithmetic before answering. '
'Return JSON with one string field named "answer".'
),
}
CASES = [
("bytes", "How many bits are in 2 bytes?", "16"),
("groups", "What is 3 multiplied by 4?", "12"),
("halves", "What is 8 divided by 2?", "4"),
]
# Synthetic responses and accounting data, not recorded API results.
RESPONSES = {
"baseline": ['{"answer":"16"}', '{"answer":"7"}', 'not JSON'],
"checked": ['{"answer":"16"}', '{"answer":"12"}', 'not JSON'],
}
ACCOUNTING = {
"baseline": (100, 30, 500),
"checked": (180, 40, 800),
}
# Fictional USD prices per million tokens; not a provider price list.
INPUT_RATE = Decimal("0.10")
OUTPUT_RATE = Decimal("0.20")
def check_response(raw, expected):
try:
value = json.loads(raw)
except json.JSONDecodeError:
return False, "invalid_json"
if (
not isinstance(value, dict)
or set(value) != {"answer"}
or not isinstance(value["answer"], str)
):
return False, "invalid_schema"
if value["answer"] != expected:
return False, "wrong_answer"
return True, "pass"
rows = []
for variant in PROMPTS:
input_tokens, output_tokens, elapsed_ms = ACCOUNTING[variant]
for (case_id, question, expected), raw in zip(
CASES, RESPONSES[variant]
):
passed, status = check_response(raw, expected)
cost = (
input_tokens * INPUT_RATE + output_tokens * OUTPUT_RATE
) / Decimal(1_000_000)
rows.append({
"variant": variant,
"case_id": case_id,
"question": question,
"passed": passed,
"status": status,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"elapsed_ms": elapsed_ms,
"cost_usd": cost,
})
for variant in PROMPTS:
group = [row for row in rows if row["variant"] == variant]
passed = sum(row["passed"] for row in group)
mean_ms = sum(row["elapsed_ms"] for row in group) / len(group)
cost = sum((row["cost_usd"] for row in group), Decimal("0"))
print(
f"{variant}: {passed}/{len(group)} passed; "
f"mean={mean_ms:.0f}ms; total=${cost:.6f}"
)
for row in group:
print(f" {row['case_id']}: {row['status']}")
The output reports 1/3 passing for baseline and 2/3 for checked. Mean fixture latency rises from 500ms to 800ms; total fixture cost rises from $0.000048 to $0.000078. Neither malformed response disappears from the denominator. Dropping invalid_json rows would reward a prompt for failing to return usable output.
¶Replace fixtures without changing the question
For a live run, replace the fixture lookup with the validated client from the preceding sections. Measure elapsed time with time.perf_counter(), starting before the first attempt and stopping after validation or terminal failure. That boundary includes retry delays, because the student waits through them. If you measure only the successful attempt, a rate limit followed by a retry becomes an apparently fast answer.
Take input and output token counts from the provider’s usage data, not from character counts. Calculate each charge as tokens multiplied by the applicable price per token. The simple formula above applies only when those two billing categories cover the request; cached input or separately billed usage requires additional fields. Record the model identifier, run date, prices, and currency. I keep token counts even when storing a calculated charge, because prices can change while the original usage remains useful.
A timeout creates an uncomfortable accounting gap: the provider may have processed the request even though your client received no usage data. Record cost as unknown, not zero, unless billing records establish otherwise. Likewise, preserve invalid_schema, wrong_answer, and exhausted-retry failures as separate outcomes. A single “failed” total cannot tell you whether to change the prompt or repair the API integration.
For an initial live exercise, use 10 fixed questions and run each variant 3 times: 60 requests before retries. Alternate variant order so one prompt does not always encounter the earlier service conditions. This remains a small experiment, not proof of general improvement. I would inspect every changed answer before adopting the candidate, especially when one additional pass determines the result.
Interview check: “Why not choose the prompt with the highest pass rate?” A defensible answer is that quality must be compared under the same conditions, failures must remain visible, and the improvement must justify measured waiting time and spending. Explain which additional answers became correct; do not hide behind the percentage.
The next chapter needs this tested client boundary: explicit inputs, validated outputs, and accounting that survives failures. Keep those independent of the model backend so a different inference path does not require rewriting the study assistant.