Programmatic Tool Calling with the Claude SDK

Every morning somebody on the team runs the same check. Thirty tables in the staging schema, and the only thing anyone wants to know is which ones did not get loaded overnight. It is a five minute job, it is boring, and it is exactly the sort of thing you would hand to an agent. So you do. And then you watch it take thirty separate trips to the model, one table at a time, dragging every result set through the context window on the way. The answer was three table names. You paid for thirty round trips to get them.

The LLMCompiler paper put a name on this in 2024. Treat the tool calls like a compiler treats instructions: work out what depends on what, then run everything else at the same time. Two years later the idea is in the API, and it arrived in a shape nobody quite predicted. There is no plan format to parse. Claude just writes the Python.

1. One field changes everything

Add allowed_callers to a tool and switch on code execution. Your tool stops being something the model asks for one call at a time and becomes an async function sitting inside a sandbox, taking a dict, returning a string. From there, loops and conditionals and asyncio.gather come free. They are just Python.

import anthropic

client = anthropic.Anthropic()

TOOLS = [
    {"type": "code_execution_20260120",
     "name": "code_execution"},
    {
        "name": "run_sql",
        "description": ("Run a read-only query on the warehouse. "
                        "Returns a JSON list of row objects."),
        "input_schema": {
            "type": "object",
            "properties": {"sql": {"type": "string"}},
            "required": ["sql"],
        },
        "allowed_callers": ["code_execution_20260120"],
    },
]

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=4096,
    tools=TOOLS,
    messages=[{"role": "user", "content":
        "For every table in the stage schema, get the latest "
        "load timestamp and list only those older than 6 hours."}],
)
  

Sonnet 5 is the right call here. Plan, fan out, summarize, nothing that needs the expensive model, which is the same reasoning behind matching the model to the task rather than the app. The floor is the code execution version, not the tier, so any Sonnet from 4.5 forward will do.

2. Somebody else writes the loop

Back comes a server_tool_use block with the generated script inside it. The first time you read one of these it is a little uncanny, because it is roughly the code you would have written yourself on a Tuesday afternoon.

# written in the container, not by you
import json, asyncio

tables = json.loads(await run_sql({"sql": LIST_SQL}))
rows = await asyncio.gather(*[
    run_sql({"sql": f"select max(load_ts) ts from {t['name']}"})
    for t in tables
])

stale = [(t["name"], r) for t, r in zip(tables, rows)
         if hours_since(r) > 6]
print(stale)
  

One query to find the tables, then all thirty timestamps at once, then a filter. Three names get printed. Three names is all the model ever sees. The other twenty seven results lived and died in the sandbox, which is the whole point.

3. You still answer the phone

Your database is not in that container, so every time the script calls the tool, everything stops and waits for you. The response shows up with stop_reason of tool_use, a container id, and one block per pending call. Run the query, hand back the rows, the script picks up mid-line.

while resp.stop_reason == "tool_use":
    results = [
        {"type": "tool_result",
         "tool_use_id": b.id,
         "content": execute(b.name, b.input)}
        for b in resp.content if b.type == "tool_use"
    ]
    # tool_result blocks only on this turn, nothing else
    messages += [{"role": "assistant", "content": resp.content},
                 {"role": "user", "content": results}]
    resp = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=4096,
        container=resp.container.id,  # required while calls pend
        tools=TOOLS,
        messages=messages,
    )
  

Two things will trip you on the first attempt. That user turn takes tool results and absolutely nothing else, not even a stray line of text. And the full tools array has to ride along on every continuation, code execution included, otherwise the paused script has nothing to wake up into.

4. Is it worth it

Depends entirely on the shape of your work, and the documentation is refreshingly blunt about it. On a 75-tool agent benchmark, about 38 percent fewer billed input tokens with accuracy unchanged. Across production traffic declaring 10 to 49 tools, savings of 20 to 40 percent. And on a benchmark where each turn made one or two sequential calls, it cost 8 percent more. Fan-out wins. Short chains lose.

Before you point this at production

Keep an eye on the clock. A pending call gives up after about four minutes and throws a timeout inside the running script, and idle containers get reclaimed after about five, so a query that takes ten minutes needs chunking, not optimism. Availability is uneven at the moment: Claude API, Claude Platform on AWS and Microsoft Foundry yes, Bedrock and Google Cloud not yet. And the obvious one, which is worth saying anyway. Every query in that fan-out is generated text. Give the tool a read-only connection and let the database enforce it, because allowed_callers is a hint to Claude about how to use the tool, not a wall around it.

Generate Embeddings in SQL with Aurora and Bedrock

Most embedding pipelines on AWS have the same shape: a job reads rows out of the database, calls Amazon Bedrock, and writes the vectors back. That is a second service to deploy, monitor, and pay for, and it exists only to move text a few hundred milliseconds away and bring numbers back. Aurora PostgreSQL can do the whole thing in one statement. The Aurora machine learning extension calls Bedrock from inside SQL, and pgvector stores and searches the result in the same table.

1. Wire the cluster to Bedrock

Aurora ML needs an IAM role carrying bedrock:InvokeModel, attached to the cluster under the Bedrock feature in the Connectivity and security tab. Model access has to be requested in Bedrock separately. After that, two extensions in the target database. The Aurora machine learning documentation walks through the console steps.

-- aws_ml pulls in aws_commons and creates the aws_bedrock schema
CREATE EXTENSION IF NOT EXISTS aws_ml CASCADE;
CREATE EXTENSION IF NOT EXISTS vector;

SELECT extname, extversion FROM pg_extension
WHERE extname IN ('aws_ml', 'vector');
  

You want aws_ml at 2.0, which is the version that adds the Bedrock functions, and vector at 0.8.0. Version 2.0 of the extension ships with Aurora PostgreSQL 16.1, 15.5, and 14.10 and higher. pgvector 0.8.0 needs 17.4, 16.8, 15.12, 14.17, or 13.20 and higher.

2. Call Bedrock from a SELECT

The function aws_bedrock.invoke_model_get_embeddings takes the model payload as JSON and pulls the vector out by key. Titan Text Embeddings V2 returns 1,024 dimensions by default and puts them under embedding. Aurora hands the result back as float8[], so cast it.

SELECT aws_bedrock.invoke_model_get_embeddings(
    model_id     := 'amazon.titan-embed-text-v2:0',
    content_type := 'application/json',
    json_key     := 'embedding',
    model_input  := '{"inputText": "tempdb contention on a heap"}'
)::vector AS embedding;
  

3. Materialize the vector, do not compute it at query time

Every invocation is billed, so the embedding belongs in a stored column that you fill once. Backfill in bounded batches rather than one enormous UPDATE, because each row is a separate synchronous call and a single failure rolls the whole statement back.

CREATE TABLE documents (
    id        bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    tenant_id text NOT NULL,
    content   text NOT NULL,
    embedding vector(1024)
);

-- backfill 500 rows at a time
UPDATE documents d
SET embedding = aws_bedrock.invoke_model_get_embeddings(
        'amazon.titan-embed-text-v2:0',
        'application/json',
        'embedding',
        json_build_object('inputText', d.content)::text
    )::vector
WHERE d.id IN (
    SELECT id FROM documents
    WHERE embedding IS NULL ORDER BY id LIMIT 500
);
  

4. Index it, then filter without losing rows

Titan V2 normalizes its output by default, which means inner product ranks identically to cosine while skipping the norm division on every comparison. Pair the <#> operator with vector_ip_ops. The m and ef_construction values below are the AWS starting points from the pgvector production guidance.

CREATE INDEX documents_embedding_ip_idx
ON documents USING hnsw (embedding vector_ip_ops)
WITH (m = 16, ef_construction = 128);

BEGIN;
SET LOCAL hnsw.iterative_scan = relaxed_order;
SET LOCAL hnsw.ef_search = 100;

SELECT id, content
FROM documents
WHERE tenant_id = 'acme'
ORDER BY embedding <#> '<query embedding>'
LIMIT 10;
COMMIT;
  

Those two settings are the reason to be on pgvector 0.8.0. Before it, a WHERE clause combined with a vector search regularly returned fewer rows than the LIMIT asked for, because the filter ran after the index had already handed back its top candidates. Iterative scans keep pulling from the index until the query is satisfied, and relaxed_order is the mode to reach for in production. The ef_search default of 40 is usually too low once real traffic arrives. Worth confirming with EXPLAIN that the planner actually picks the HNSW index here and has not quietly fallen back to a sequential scan, which is the kind of thing PlanTrace makes obvious at a glance.

Before you point this at a large table

The appeal here is the missing infrastructure. No Lambda, no queue, no separate deployment, and the vector never leaves the transaction that produced it. The limit is that Bedrock calls through Aurora ML are one row at a time: the UDFs support neither batching nor parallel execution, unlike the Comprehend and SageMaker functions in the same extension. That makes this a good fit for scheduled backfills through pg_cron and for low-volume writes, and a poor fit for an INSERT trigger on a high-throughput table. Size the instance so the HNSW graph stays in memory, watch BufferCacheHitRatio, and treat a drop there as the first sign the index has outgrown the box.

Match the Model to the Task, Not the App: An Orchestrator-Worker Pattern in Python

Most apps that call an LLM send every request to the same model. That works until the bill arrives. A classification job that returns one word costs the same as a multi-step reasoning job, because you picked one model at design time and never revisited it. The fix is an orchestrator-worker setup: a small central router reads the incoming task, decides which tier of model it needs, and delegates to a worker. The router sits at the hub, each model tier is a spoke, and no worker talks to another. This post builds that in Python using both the Anthropic and OpenAI SDKs, so you can mix providers in the same pipeline.

1. Define the tiers before writing any code

Three tiers cover almost everything. Fast handles extraction, classification, tagging, and format conversion. Balanced handles summarization, SQL generation, and short analysis. Deep handles multi-step reasoning, code review, and anything where a wrong answer is expensive. Map each tier to a model from each provider so you can fail over between them.

# tiers.py
TIERS = {
    "fast":     {"anthropic": "claude-haiku-4-5", "openai": "gpt-5.6-luna"},
    "balanced": {"anthropic": "claude-sonnet-5",  "openai": "gpt-5.6-terra"},
    "deep":     {"anthropic": "claude-opus-4-8",  "openai": "gpt-5.6-sol"},
}
  

2. The orchestrator itself belongs on the cheap tier

The orchestrator runs on every request, before any real work happens. Whatever it costs gets multiplied across your entire traffic, and its latency lands in front of every response. Putting a deep model there means paying premium prices to decide what to do. Routing is short classification against a closed label set, which is precisely what the small models handle well.

The risk is that router errors are asymmetric. Sending a fast task to the deep tier wastes money and you see it in the bill. Sending a deep task to the fast tier produces a plausible wrong answer that flows downstream unnoticed. Three controls keep that in check: a closed label set instead of free-form output, a hard cap on max_tokens so the router cannot drift into explanation, and the validator gate covered further down, which catches the cases where the cheap tier was not enough.

import anthropic

client = anthropic.Anthropic()

ROUTER_PROMPT = """Classify the task into exactly one tier.
fast: extraction, classification, tagging, formatting
balanced: summarization, query generation, short analysis
deep: multi-step reasoning, code review, planning
Answer with the tier name only."""

def route(task_text):
    msg = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=8,
        system=ROUTER_PROMPT,
        messages=[{"role": "user", "content": task_text}],
    )
    tier = msg.content[0].text.strip().lower()
    return tier if tier in TIERS else "balanced"
  

The fallback on the last line matters. Routers misfire, and an unrecognized label should land on a safe middle tier rather than raise. And when your app already knows the task type because the user clicked a specific button or the pipeline stage is fixed, skip the router entirely. A dictionary lookup costs nothing and never misclassifies.

3. Wrap each worker behind one function signature

The two SDKs differ in shape. Anthropic uses messages.create with a separate system argument and a required max_tokens. OpenAI uses responses.create with an input list and a reasoning effort setting. Normalize both to take a prompt and return a string, and the orchestrator stops caring which vendor answered. The Anthropic Python SDK and the OpenAI model catalog document the current identifiers for each family.

import anthropic
from openai import OpenAI

anthropic_client = anthropic.Anthropic()
openai_client = OpenAI()

def call_anthropic(model, system, prompt, max_tokens=1024):
    msg = anthropic_client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    return msg.content[0].text

def call_openai(model, system, prompt, effort="low"):
    resp = openai_client.responses.create(
        model=model,
        instructions=system,
        input=prompt,
        reasoning={"effort": effort},
    )
    return resp.output_text
  

4. Dispatch, and allow one escalation

The orchestrator is now small. Pick the tier, pick the provider, run the call. Add a single escalation path: if the fast tier returns something your validator rejects, retry once on the next tier up. That ladder is a model cascade, and it should stay short. One retry, not a loop. Unbounded escalation turns a cost optimization into a cost multiplier.

LADDER = ["fast", "balanced", "deep"]

def run_task(prompt, system, validator, provider="anthropic"):
    tier = route(prompt)
    idx = LADDER.index(tier)

    for attempt_tier in LADDER[idx:idx + 2]:
        model = TIERS[attempt_tier][provider]
        if provider == "anthropic":
            out = call_anthropic(model, system, prompt)
        else:
            effort = "low" if attempt_tier == "fast" else "high"
            out = call_openai(model, system, prompt, effort)

        if validator(out):
            return {"tier": attempt_tier, "model": model, "output": out}

    raise ValueError("no tier produced a valid result")
  

The validator is the piece people skip, and it is what makes the whole design safe. For extraction, check that the JSON parses and the required keys exist. For SQL generation, run EXPLAIN against the target database. Without a validator you have no signal that the cheap tier was good enough, and the routing decision becomes a guess.

5. Log the tier, not just the answer

Store the tier, model, token counts, and validator result for every call. After a week you will see which task types actually escalate and which ones you over-provisioned from the start. That log is the only honest way to tune the routing prompt, and it usually reveals that a large share of traffic never needed anything above the fast tier.

6. A real pipeline: invoice ingestion

Consider a service that receives scanned invoices and has to produce clean rows in a database. Four distinct jobs happen per document, and only one of them is hard. Classifying the document type is a label from a short list. Field extraction is bounded and schema-driven. Totals reconciliation is arithmetic plus judgment about rounding and tax lines. Resolving an ambiguous line item against a product catalog is where reasoning actually pays for itself. Hardcode the first three, and let the orchestrator decide only on the last one.

# pipeline.py

STAGE_TIERS = {
    "classify_doc":   "fast",
    "extract_fields": "fast",
    "reconcile":      "balanced",
}

def run_stage(stage, system, prompt, validator, provider="anthropic"):
    # known stage, no router call needed
    tier = STAGE_TIERS[stage]
    model = TIERS[tier][provider]
    ...  # same call plus escalation logic as run_task

def process_invoice(ocr_text):
    doc_type = run_stage(
        "classify_doc",
        system="Answer with one of: invoice, receipt, credit_note, other.",
        prompt=ocr_text[:2000],
        validator=is_known_doc_type,
    )
    if doc_type != "invoice":
        return route_to_manual_queue(ocr_text, doc_type)

    fields = run_stage(
        "extract_fields",
        system=EXTRACTION_SCHEMA_PROMPT,
        prompt=ocr_text,
        validator=parses_as_invoice_json,
    )

    if not totals_match(fields):
        fields = run_stage(
            "reconcile",
            system=RECONCILE_PROMPT,
            prompt=describe_mismatch(fields),
            validator=totals_match,
        )

    # only unresolved items reach the orchestrator and the expensive tiers
    for item in unmatched_items(fields):
        item["sku"] = run_task(
            prompt=catalog_match_prompt(item),
            system=CATALOG_PROMPT,
            validator=sku_exists,
        )["output"]

    return persist(fields)
  

Notice what the structure buys you. Every invoice pays for two fast calls. The balanced tier only runs when the arithmetic disagrees, and the deep tier only touches line items the catalog lookup could not resolve on its own. On a clean batch, most documents never leave the cheap tier, and the expensive models are reserved for the handful of cases that genuinely earn them.

Wrapping Up

Orchestrator-worker does not have to mean an elaborate framework. A tier table, a cheap orchestrator, one wrapper per SDK, and a validator gate get you most of the benefit in under a hundred lines. Both providers expose a low-cost tier that handles far more than people expect once the prompt is narrow enough.

One practical note before you ship: pin the versioned model identifiers in production rather than the family aliases. Aliases move when a vendor releases a new generation, and a routing table that silently changes tiers under you is a debugging session nobody enjoys.

Moving On-Prem PostgreSQL to the Cloud: Picking the Right Path for Big Tables

Every PostgreSQL migration eventually hits the same fork in the road. The database is small enough to dump and restore in a maintenance window, or it isn't. Once you cross into terabyte territory, or the business can't afford more than a few minutes of downtime, the simple approach stops being an option and you have to think harder. Here are the three paths that actually work, and when each one makes sense.

1. Dump and restore, the reliable baseline

For anything under a couple hundred GB, pg_dump piped into pg_restore is still the cleanest way to move a database. It carries the full schema, data, indexes, and sequences, and on decent hardware a mid-size database finishes in well under an hour. The one flag people forget is parallelism.

# dump in the directory format so restore can run in parallel
pg_dump -Fd -j 8 -f /dump/appdb sourcedb

# restore with the same worker count on the target
pg_restore -Fd -j 8 -d appdb /dump/appdb
  

The catch shows up with one dominant table. Parallelism here is per-table, so a single 2 TB table becomes one long serial stream while the other seven workers sit idle. That is the moment dump and restore starts feeling slow, and the reason the next two options exist.

2. Native logical replication for a live cutover

When downtime is the real constraint, PostgreSQL's built-in logical replication lets the target keep pace with the source while the application stays online. You publish on the old server, subscribe on the new one, and Postgres copies the initial snapshot and then streams every change until you decide to switch over.

-- on the source (wal_level must be 'logical')
CREATE PUBLICATION migration_pub FOR ALL TABLES;

-- on the target
CREATE SUBSCRIPTION migration_sub
  CONNECTION 'host=old-db port=5432 dbname=appdb user=repl password=...'
  PUBLICATION migration_pub;
  

Two things save you real pain on big datasets. First, tables without a primary key need a REPLICA IDENTITY FULL or a unique index, otherwise updates and deletes won't replicate. Second, that same large table problem returns. Logical replication syncs each table with its own worker, so your biggest table still copies as a single stream. A common workaround is to pre-load the huge table with pg_restore, then attach it to the subscription with copy_data = false so replication only handles the incremental changes.

3. Managed DMS when you want the plumbing handled

Both AWS and Google Cloud offer a managed migration service that wraps logical replication for PostgreSQL, including from on-premises sources. Google's Database Migration Service supports CDC via pglogical and can open multiple parallel subscriptions across tables during the initial load and the ongoing CDC phase. AWS DMS does the same using logical replication slots with the test_decoding or pglogical plugin. Both handle slot management and give you a controlled cutover step.

The shared limitation is worth knowing before you commit: for tables without a primary key, both services replicate the initial snapshot and INSERT operations only. UPDATE and DELETE changes during CDC are dropped and need to be handled separately. Beyond that, parallelism is still table-level on both platforms, so the single dominant large table remains the bottleneck either way.

How to actually choose

Small database and a maintenance window you can spare? Dump and restore, done. Need the source online with minimal downtime and comfortable managing slots yourself? Native logical replication. Want the same low-downtime result without babysitting the replication setup? Reach for the managed DMS on your target cloud. In all three cases the single dominant table is what decides your timeline, so measure it first and plan to pre-seed it on its own rather than hoping a migration tool will parallelize what the engine simply won't.