Calling the Claude API: REST, JSON and sync vs async patterns

1.The Messages API as a REST endpoint that speaks JSON

Calling Claude is an ordinary REST call, so the conventions your other services already follow carry over. The API is served from https://api.anthropic.com, and a conversational request is POST /v1/messages. Every request has a JSON body with content-type: application/json, an anthropic-version header such as 2023-06-01, and credentials in Authorization: Bearer <token>. The older x-api-key header still works as a fallback.

The official SDKs set the authentication, version and content-type headers for you. They also add typed requests and responses, built-in retries, streaming support and request timeouts. With a raw HTTP client, all of that is your code to write. The one header you still pass yourself is anthropic-workspace-id, and only when your key covers more than one workspace.

{
  "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
  "type": "message",
  "role": "assistant",
  "content": [
    {
      "type": "text",
      "text": "Hello!"
    }
  ],
  "model": "claude-opus-5-5",
  "stop_reason": "end_turn",
  "stop_sequence": null,
  "usage": {
    "input_tokens": 12,
    "output_tokens": 6
  }
}

Because the API keeps no conversation state, your service has to store it. The client sends the whole history with every request, as covered in [Stateless Messages API (client holds the conversation)](lesson:2.3). In practice, keep the messages array as plain JSON, a list of {"role", "content"} objects, in your own database or session store keyed by conversation. Load it, append the user's new turn, send it, then append the assistant's content and save it again. Earlier turns don't have to come from Claude, so a stored history can include synthetic assistant messages you wrote yourself.

That stored history grows with every turn, and request size is capped. Messages and Token Counting requests can be at most 32 MB, and anything larger returns 413 request_too_large. A long-lived chat that saves every turn forever will eventually hit that limit, so decide how history is trimmed before it does.

A REST call spends almost all of its time waiting on the network, and asynchronous programming is how a program uses that wait for other work instead of freezing. With the async client, await client.messages.create(...) suspends only the current coroutine; the event loop runs other coroutines until the response arrives. The request and the returned Message are identical to a blocking call, so the change is in how the waiting is scheduled, not in the JSON.

Interleaving brings a hazard that blocking code doesn't have. Coroutines can interleave, so several in-flight API calls that append to one shared list or dict can corrupt it, the same shared-mutable-state race that threads cause. Have each coroutine return its own result and combine them afterwards, rather than mutating shared state from inside the call. The API also offers asynchronous work at the endpoint level: the Message Batches API accepts requests now and processes them later, so the HTTP response only acknowledges the submission.

Sources123

2.Blocking calls and the async client

The simplest pattern is a blocking call. client.messages.create(...) on the synchronous Anthropic client waits until Claude has finished, then returns the full Message. That suits scripts, CLI tools and worker processes that handle one request at a time, where nothing else needs the thread while the request is in flight.

A service built on an event loop needs something else, because a blocking call there holds up every other coroutine while one request waits on the network. The Python SDK's AsyncAnthropic client has the same messages.create method, and you await it inside an async function. The request body and the returned Message are unchanged. The only difference is that your code yields while it waits. The async client can also use an aiohttp HTTP backend instead of the default one when you need better async throughput.

async def main() -> None:
    message = await client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Hello, Claude",
            }
        ],
        model="claude-opus-5-5",
    )
    print(message.content)


asyncio.run(main())

async def main() -> None:
    message =  ?  client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Hello, Claude",
            }
        ],
        model="claude-opus-5-5",
    )
    print(message.content)


asyncio.run(main())

Sources4

3.Streaming when output is long or a user is watching

Streaming sits between a blocking call and a background job. The response arrives in pieces over one open HTTP connection, as covered in [Streaming over server-sent events](lesson:5.2). It wins in two situations. In the first, a person is waiting, so printing text as it arrives makes the response feel fast. In the second, the response is long enough that one silent connection would hit an HTTP timeout. For that second case the SDK requires streaming, even when you don't want partial text.

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=128000,
    messages=[{"role": "user", "content": "Write a detailed analysis..."}],
    model="claude-opus-5-5",
) as stream:
    message = stream.get_final_message()

Streaming also changes how failures reach you. A problem that a blocking call reports as an HTTP status can arrive mid-stream as an error event. Overload is the common case: an overloaded_error event where a non-streaming call would get HTTP 529. Error handling written only around the initial HTTP response will miss it. The Python SDK offers both sync and async streams, so streaming works with either client from the previous section.

Sources5

4.Submit-and-poll for work nobody is waiting on

Submit-and-poll decouples sending work from collecting results. You submit a job, get back an ID, and check its status later. The Message Batches API, which handles large volumes of requests asynchronously at lower cost, is the standard example. Its pricing and limits are covered in [Message Batches API and realtime vs batch](lesson:2.3). For the code you write, what matters is the shape: a batch starts with processing_status set to in_progress and moves to ended once every request has finished and results are ready.

message_batch = None
while True:
    message_batch = client.messages.batches.retrieve(MESSAGE_BATCH_ID)
    if message_batch.processing_status == "ended":
        break

    print(f"Batch {MESSAGE_BATCH_ID} is still processing...")
    time.sleep(60)
print(message_batch)

Polling has a cost: the job ID has to outlive the process that submitted it. Store the batch id along with whatever you need to match results back to your records. The poller can then run as a separate scheduled job, and a restart doesn't lose the work. Choose this pattern when throughput and cost matter more than latency, for example nightly classification or evaluation runs. Don't choose it when a user is waiting for the answer.

Sources6

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Streaming is only for UIs. A backend job that wants the whole answer should always use a plain blocking call.Why is that wrong?

   With large max_tokens values, the SDK requires streaming to avoid HTTP timeouts. Calling get_final_message() on the stream still returns the complete Message.

   Covered in Streaming when output is long or a user is watching

2. 2.The API remembers earlier turns, so each request only needs the newest user message.Why is that wrong?

   The Messages API is stateless. Your service has to store the history and send all of it on every request.

   Covered in The Messages API as a REST endpoint that speaks JSON