Combining LLMs with APIs and Databases

A logistics company gives its internal assistant one tool: run_sql, read-only, pointed at a replica of the warehouse database. The demo is excellent. "How many parcels are late in Leeds?" comes back correct in four seconds, and everyone agrees this is the future.

Nine days later an operations manager asks "which orders shipped last quarter?" The model writes exactly the SQL a human would write:

SELECT * FROM ordersWHERE shipped_at >= '2026-01-01' AND shipped_at < '2026-04-01';

That query returns 24,180 rows. The tool serialises them to JSON — about 1.9 MB — and hands the lot back to the model as a tool result. The next API call dies before it starts:

BadRequestError: 400 - This model's maximum context length is 200000tokens, however your messages resulted in 623441 tokens.

The conversation is now permanently broken: the oversized tool result is already in the message history, so every retry produces the same 400. Note what was not wrong. The reasoning was right, the SQL was right, the database was healthy. The tool was wrong — no row limit, no result shaping, no ceiling on what it would push into a context window.

Wiring a model to a real API or database is roughly ten per cent plumbing and ninety per cent boundary design. The ninety per cent is what follows.

Three shapes of integration

Almost every integration is one of three shapes, and they fail in genuinely different ways. Recognising which one you are in tells you what to worry about.

Shape 1: model decides, you call an external service

User      "Is it going to rain in Bristol tomorrow?"  vModel     emits get_forecast{location:"Bristol", days:1}  vHandler   validates args, adds the API key, sets a timeout  vUpstream  350 ms, 200 OK, 14 KB of JSON  vHandler   shapes it to 6 fields, ~90 tokens  vModel     reads the real numbers, writes the answer  vUser      "Yes - 8 mm expected between 14:00 and 18:00."

Shape 2: model decides, you touch your own data

Same skeleton, but the service is your database and some calls write. The difference is not technical: a bad weather lookup is embarrassing, a bad UPDATE is an incident. Reads and writes deserve separate treatment from line one.

Shape 3: multi-step workflow

User: "Order the blue mug and text me when it's confirmed."   |   vModel -> search_products{query:"blue mug"}          [read]   |   vModel -> create_order{sku:"MUG-BL-01", qty:1}       [write]   |   vModel -> charge_card{amount:12.50, order:"O-8871"}  [write, irreversible]   |   vModel -> send_sms{to:"+44...", body:"Confirmed"}    [write, external]   |   vFinal answer

Four steps, four chances to fail, and the failures are not symmetric. If step 4 fails you have a paid-for order and a silent customer. If step 3 fails after the card was charged but before your code saw the response, you have taken money for nothing.

The tool boundary is an API design problem

The useful mental shift: you are not "giving the model database access", you are publishing an API whose consumer happens to be a language model. That consumer has unusual properties.

• It reads only the tool name, description and argument schema. It cannot open your docs, see your table definitions, or ask a colleague.
• When it is unsure, it does not error — it produces something plausible. Ambiguity in your description becomes invented arguments.
• Every byte you return costs money and latency, on this turn and every subsequent turn, because tool results stay in the conversation history.
• It will call your tool with arguments no human would type, because a user asked.

Design the tool for the question the user is asking, not for the shape of your internal database. A tool is a product surface, not a thin RPC wrapper.

Calling an external API properly

The naive weather handler almost every tutorial shows:

def get_weather(location):    r = requests.get(URL, params={"q": location, "appid": KEY})    return r.json()          # 14 KB of upstream JSON, straight into the context

No timeout, so a hung upstream hangs your request thread indefinitely. No error handling, so a 404 becomes an exception mid-loop. No shaping, so you pay for fields nobody reads. And no validation of the argument you were handed. The hardened version:

1import os, requests2from typing import Any34WEATHER_URL = "https://api.openweathermap.org/data/2.5/weather"5API_KEY = os.environ["OPENWEATHER_API_KEY"]   # never a literal in source67def get_weather(location: str, units: str = "metric") -> dict[str, Any]:8    """Current conditions. Always returns a dict; never raises."""9    if not isinstance(location, str) or not 1 <= len(location) <= 80:10        return {"ok": False, "error": "invalid_location",11                "message": "location must be a city name, 1-80 characters"}12    try:13        r = requests.get(14            WEATHER_URL,15            params={"q": location, "units": units, "appid": API_KEY},16            timeout=(3.0, 5.0),        # (connect, read) - always a tuple17        )18    except requests.Timeout:19        return {"ok": False, "error": "upstream_timeout",20                "message": "weather service did not respond in 5s"}21    except requests.RequestException as e:22        return {"ok": False, "error": "upstream_unreachable",23                "message": type(e).__name__}2425    if r.status_code == 404:26        return {"ok": False, "error": "location_not_found",27                "message": f"no city matched {location!r}; try 'City, CC'"}28    if r.status_code >= 400:29        return {"ok": False, "error": "upstream_error",30                "message": f"HTTP {r.status_code}"}3132    d = r.json()33    return {                                  # shaped: 6 fields, not 4034        "ok": True, "location": f"{d['name']}, {d['sys']['country']}",35        "temperature_c": round(d["main"]["temp"], 1),36        "feels_like_c": round(d["main"]["feels_like"], 1),37        "condition": d["weather"][0]["main"],38        "humidity_pct": d["main"]["humidity"]}

The shaped result the model actually sees:

1{2  "ok": true,3  "location": "Bristol, GB",4  "temperature_c": 11.4,5  "feels_like_c": 9.8,6  "condition": "Rain",7  "humidity_pct": 878}

Roughly 45 tokens instead of 3,500 for the raw body. Over a 10-turn conversation, where the result is resent every turn, the raw version costs about 35,000 tokens of repeated context; the shaped one about 450.

Why the error must be a return value, not an exception

When get_weather("Atlantis") returns {"ok": false, "error": "location_not_found"} and you feed that back, the model says "I couldn't find Atlantis — did you mean Atlanta?" and the conversation survives. Raise instead and your loop crashes with the model never getting a chance to recover. Tool handlers are total functions: every input yields a JSON-serialisable output.

Several APIs at once, and what changes

Add a payment API and an SMS API and two problems appear that a single read-only service never has: double execution and partial completion.

Double execution and idempotency keys

A charge has a 5-second read timeout; the upstream takes 5.4 seconds. Your client raises Timeout at 5.0 s, the retry fires, the second attempt succeeds. The customer is charged 89.99 twice, because the first attempt also succeeded — you never saw the response.

The fix is an idempotency key: a value derived from the business intent, not from the attempt.

1import stripe23def charge_order(order_id: str, amount_gbp: float,4                 customer_id: str, payment_method_id: str) -> dict:5    try:6        intent = stripe.PaymentIntent.create(7            amount=int(round(amount_gbp * 100)),   # pence, integer - never floats8            currency="gbp",9            customer=customer_id,10            payment_method=payment_method_id,      # saved card11            off_session=True, confirm=True,        # charge now, no redirect12            metadata={"order_id": order_id},13            idempotency_key=f"charge:{order_id}",  # same key on every retry14        )15        return {"ok": True, "transaction_id": intent.id, "amount_gbp": amount_gbp}16    except stripe.CardError as e:17        return {"ok": False, "error": "card_declined", "message": e.user_message}18    except stripe.StripeError:19        return {"ok": False, "error": "payment_provider_error",20                "message": "payment could not be confirmed; not retried"}

With idempotency_key="charge:O-8871" the retry returns the original payment rather than creating a second. (The code uses Stripe's Payment Intents API. Stripe has deprecated creating card payments with the older Charges API, and new integrations cannot use it.) Note int(round(amount_gbp * 100)) too: 19.99 * 100 in floating point is 1998.9999999999998, and bare int() truncates it to 1998 pence. Money is integer pence at the boundary.

Partial completion

Payment succeeds, SMS fails. Refunding the card because Twilio was down is the wrong reaction. The right structure: irreversible steps commit, and everything downstream becomes a retriable background job.

Talking to a SQL database

There are two designs here, and choosing badly is expensive.

Injection, via arguments you did not write

String interpolation is unsafe even in the "fixed intent" design, because the arguments still come from the model, which got them from a user. A user types "look up the customer whose email is ' OR '1'='1"; the model passes it through, and this handler:

cursor.execute(f"SELECT id, name, email FROM customers WHERE email = '{email}'")

executes:

SELECT id, name, email FROM customers WHERE email = '' OR '1'='1'

Every customer row is returned, dumped into the model's context, and summarised back to whoever asked. That is a data breach with a polite tone of voice. The fix is the same as it has been for twenty-five years — bind parameters:

1import sqlite32from contextlib import closing34def find_customer(email: str) -> dict:5    if not isinstance(email, str) or "@" not in email or len(email) > 254:6        return {"ok": False, "error": "invalid_email"}7    with closing(sqlite3.connect(DB_PATH)) as conn:8        conn.row_factory = sqlite3.Row9        row = conn.execute(10            "SELECT id, name, email, phone FROM customers WHERE email = ? LIMIT 1",11            (email,),                       # bound, never interpolated12        ).fetchone()13    if row is None:14        return {"ok": False, "error": "customer_not_found",15                "message": f"no customer with email {email}"}16    return {"ok": True, "customer": dict(row)}

A model-supplied argument is user input that has been through a paraphrasing machine. Treat it exactly as you would treat a raw form field from the public internet.

If you do let the model write SQL

Sometimes the analytics value is worth it. Then the guardrails are not optional, and none of them is "ask nicely in the prompt":

1. A read-only database role. GRANT SELECT on named tables, nothing else. This is the only control that survives a clever prompt injection, because the database enforces it, not your Python.
2. A statement timeout (SET LOCAL statement_timeout = '5s'), or one accidental cross join pins a CPU for an hour.
3. A forced row cap: wrap it as SELECT * FROM (…user query…) AS q LIMIT 200, and tell the model when you truncated.
4. A byte cap on the serialised result, checked before it goes near the context.
5. One statement only. Reject anything with a second semicolon-terminated statement.

Shaping the result: the 1,550× difference

Return the answer, not the data. The opening failure returned 24,180 rows to a question whose useful answer is an aggregate:

1SELECT region, COUNT(*) AS orders, ROUND(SUM(total_gbp), 2) AS revenue_gbp2FROM orders3WHERE shipped_at >= '2026-01-01' AND shipped_at < '2026-04-01'4GROUP BY region5ORDER BY revenue_gbp DESC;

1{2  "ok": true,3  "row_count": 8,4  "truncated": false,5  "rows": [6    {"region": "London",     "orders": 6104, "revenue_gbp": 481203.55},7    {"region": "North West", "orders": 3877, "revenue_gbp": 298440.10},8    {"region": "Scotland",   "orders": 2210, "revenue_gbp": 176992.40}9  ]10}

That is a factor of 1,550 in tokens, and the aggregate is also more accurate, because arithmetic over thousands of rows is precisely the thing a language model does badly and a database does perfectly.

Document databases have the same holes, differently shaped

With MongoDB the injection vector is not concatenation but operator objects. If a tool accepts a filter document and a user coaxes the model into producing {"email": {"$ne": null}}, that filter matches every document in the collection. Never let the model supply a filter document: it supplies scalars, you build the filter.

1customers = MongoClient(MONGO_URI, serverSelectionTimeoutMS=3000)["shop"]["customers"]2customers.create_index([("email", 1)], unique=True)   # index, or it scans34def find_customer_mongo(email: str) -> dict:5    if not isinstance(email, str):          # blocks {"$ne": null} outright6        return {"ok": False, "error": "invalid_email"}7    doc = customers.find_one({"email": email},8                             projection={"_id": 1, "name": 1, "email": 1})9    if doc is None:10        return {"ok": False, "error": "customer_not_found"}11    doc["_id"] = str(doc["_id"])            # ObjectId is not JSON-serialisable12    return {"ok": True, "customer": doc}

The isinstance check is the actual security control here — an operator object is a dict, not a str. And ObjectId raises TypeError: Object of type ObjectId is not JSON serializable the moment you hand the result to the model, killing the loop.

Connections, pools, and a little queueing theory

Opening a database connection costs 5–50 ms and a chunk of server memory. Doing it inside every tool call is fine at one request per second and catastrophic at fifty.

How big should the pool be? Little's law answers it: connections in use equals arrival rate times holding time, $L = \lambda W$.

At 50 requests per second, each holding a connection for 120 ms:

Six connections are busy on average. A pool of 5 therefore queues permanently — arrivals outpace releases and the wait grows without bound. A pool of exactly 6 is critical, which means terrible tail latency. Allow headroom for bursts and slow queries: 10 to 12 here. Higher is not free either — every connection is memory and a scheduling slot, and 300 connections fighting over 8 cores is slower than 20.

The pool implementation people usually reach for first has three defects:

1class DatabasePool:               # do not ship this2    def __init__(self, db_path, pool_size=5):3        self.connections = [sqlite3.connect(db_path) for _ in range(pool_size)]45    @contextmanager6    def get_connection(self):7        conn = self.connections.pop()      # (1) IndexError when exhausted8        try:9            yield conn                     # (2) no rollback on exception10        finally:11            self.connections.append(conn)

1. Under load, pop() on an empty list raises IndexError: pop from empty list — a crash instead of a wait.
2. If the caller's transaction failed, the connection goes back into the pool mid-transaction and poisons the next borrower.
3. list.pop() / append() are not a synchronisation primitive; use a queue.Queue, which blocks with a timeout, or a real pooler.

Use your driver's pool (psycopg_pool.ConnectionPool, SQLAlchemy's QueuePool) rather than writing one. And a SQLite connection is bound to its creating thread unless you pass check_same_thread=False and serialise access yourself; sharing them naively raises ProgrammingError: SQLite objects created in a thread can only be used in that same thread.

Failing well: timeouts, retries, and a bug worth studying

Here is a retry decorator of the kind that circulates widely. It contains a bug that makes it fail on the first retry, every time.

1def retry_on_failure(max_retries=3, delay=1.0):2    def decorator(func):3        @wraps(func)4        def wrapper(*args, **kwargs):5            for attempt in range(max_retries):6                try:7                    return func(*args, **kwargs)8                except Exception as e:9                    if attempt < max_retries - 1:10                        time.sleep(delay)     # UnboundLocalError11                        delay *= 212                    else:13                        return {"ok": False, "error": str(e)}14        return wrapper15    return decorator

Because delay is assigned anywhere inside wrapper, Python treats it as local to wrapper for the whole body — an assignment on the line after the read is enough. So time.sleep(delay) raises UnboundLocalError: cannot access local variable 'delay' where it is not associated with a value, from inside an except block, so the traceback foregrounds the original error and buries the real one. The fix is one line: bind a local copy.

1RETRYABLE = (requests.Timeout, requests.ConnectionError)23def retry_on_failure(max_retries=3, base_delay=1.0):4    def decorator(func):5        @wraps(func)6        def wrapper(*args, **kwargs):7            wait = base_delay                       # local copy, no closure write8            for attempt in range(1, max_retries + 1):9                try:10                    return func(*args, **kwargs)11                except RETRYABLE as e:              # only retryable errors12                    if attempt == max_retries:13                        return {"ok": False, "error": "upstream_unavailable",14                                "message": f"{max_retries} attempts failed"}15                    time.sleep(wait + random.uniform(0, 0.3 * wait))  # jitter16                    wait *= 217        return wrapper18    return decorator

The retry budget is latency you are spending: three attempts at a 5-second read timeout plus 1 s + 2 s of backoff is a worst case of 3 × 5 + 3 = 18 seconds before the model starts writing. Decide whether the user would rather wait 18 seconds or hear "the weather service is down" after 6. Jitter matters too — without it every worker retries at the same instant, delivering a synchronised second wave to a struggling upstream.

When an upstream is properly down, retries stop helping and start hurting: every request now burns 18 seconds of a worker before failing. A circuit breaker fixes that — count consecutive failures, and after five, fail instantly for 30 seconds without calling out at all, then let one probe through. Users get a fast honest error instead of a slow one, and the upstream gets room to recover.

Error envelopes, logging, and what not to say out loud

Tool results have two audiences with opposite needs. The model needs enough detail to recover, your logs enough to debug, and neither should contain a database password.

1# Good - short, actionable, no internals; the model can recover from these2{"ok": False, "error": "customer_not_found",3 "message": "No customer with that email; ask the user to check the spelling."}4{"ok": False, "error": "rate_limited", "retry_after_s": 60,5 "message": "Quota exhausted; try again in 60 seconds."}67# Bad - leaks internals into a context the user may see, and is unactionable8{"ok": False, "error": "psycopg.OperationalError: connection to server at "9                       "10.0.4.19, port 5432 failed: FATAL: password auth "10                       "failed for user app_rw"}

That second envelope tells an attacker your internal IP, port, and service account name, and tells the model nothing it can act on. Log the full exception with a correlation id; return the id and a plain sentence.

Audit logging is the other half — one structured line per tool invocation, with redaction applied where it is written rather than hopefully later:

1REDACT = {"token", "card", "password", "api_key", "phone", "email"}23def redact(args: dict) -> dict:4    return {k: ("[redacted]" if k.lower() in REDACT else v) for k, v in args.items()}56log.info(json.dumps({7    "event": "tool_call", "trace_id": trace_id,   # ties call to conversation turn8    "user_id": user_id, "tool": name, "args": redact(args),9    "ok": result.get("ok"), "error": result.get("error"),10    "duration_ms": round(elapsed * 1000),11}))

The trace_id makes this usable. Without it you have a pile of tool calls; with it you can reconstruct which user turn produced which sequence of calls — the only way to debug an agent that did something strange three days ago.

If you cannot reconstruct, from logs alone, every external effect a conversation had, you cannot safely give that conversation write access to anything.

What to do before you wire up your first tool

What separates a demo from something you can leave running is a set of decisions made before the handler is written:

• Decide the result-size ceiling first. 200 rows, 8 KB, whatever fits your budget — enforced in code, with a truncated: true flag so the model knows it has a subset. That single decision would have prevented the failure that opened this lesson.
• Split read tools from write tools and give them different credentials. A read tool holding a read-only role cannot be talked into deleting anything, whatever appears in the context.
• Give every outbound call a timeout as you write it. Timeouts get retrofitted after the first outage, never before.
• Make every write idempotent with a key derived from the business intent. You will retry, whether you plan to or not.
• Return errors, never raise them, in words the model can act on.
• Shape results in the handler, not the prompt. "Please summarise the data" is a request; a projection is a guarantee.

Every one of those is a property of your code, not of the model. The model proposes actions in a well-defined format; whether those actions turn out fast, cheap, correct and safe is decided entirely on your side of the boundary. That is good news, because your side is the side you can test.