Anthropic Events & Async Processing
Overview
The Claude API does not use traditional webhooks. Instead it provides two event-driven patterns: Server-Sent Events (SSE) for real-time streaming and the Message Batches API for async bulk processing. This skill covers both.
SSE Streaming Events
import anthropic
client = anthropic.Anthropic()
# Process each SSE event type
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": "Explain microservices."}]
) as stream:
for event in stream:
match event.type:
case "message_start":
print(f"Started: {event.message.id}")
case "content_block_start":
if event.content_block.type == "tool_use":
print(f"Tool call: {event.content_block.name}")
case "content_block_delta":
if event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
elif event.delta.type == "input_json_delta":
print(event.delta.partial_json, end="")
case "message_delta":
print(f"\nStop: {event.delta.stop_reason}")
print(f"Output tokens: {event.usage.output_tokens}")
case "message_stop":
print("[Complete]")
SSE Event Reference
| Event |
When |
Key Data |
message_start |
Stream begins |
message.id, message.model, message.usage.input_tokens |
content_block_start |
New block begins |
content_block.type (text or tool_use), index |
content_block_delta |
Incremental content |
delta.text or delta.partial_json |
content_block_stop |
Block finishes |
index |
message_delta |
Message-level update |
delta.stop_reason, usage.output_tokens |
message_stop |
Stream complete |
(empty) |
ping |
Keepalive |
(empty) |
Async Batch Processing
# Submit batch (up to 100K requests, 50% cheaper)
batch = client.messages.batches.create(
requests=[
{
"custom_id": f"doc-{i}",
"params": {
"model": "claude-sonnet-4-20250514",
"max_tokens": 1024,
"messages": [{"role": "user", "content": f"Summarize: {doc}"}]
}
}
for i, doc in enumerate(documents)
]
)
# Poll for completion
import time
while True:
status = client.messages.batches.retrieve(batch.id)
if status.processing_status == "ended":
break
counts = status.request_counts
print(f"Processing: {counts.processing} | Done: {counts.succeeded} | Errors: {counts.errored}")
time.sleep(30)
# Stream results
for result in client.messages.batches.results(batch.id):
if result.result.type == "succeeded":
print(f"[{result.custom_id}]: {result.result.message.content[0].text[:100]}")
else:
print(f"[{result.custom_id}] ERROR: {result.result.error}")
Event-Driven Architecture Pattern
# Use queues to decouple Claude requests from user-facing endpoints
from redis import Redis
from rq import Queue
redis = Redis()
queue = Queue(connection=redis)
def process_with_claude(prompt: str, callback_url: str):
"""Background job for async Claude processing."""
client = anthropic.Anthropic()
msg = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
# Notify your system via internal callback
import requests
requests.post(callback_url, json={
"text": msg.content[0].text,
"usage": {"input": msg.usage.input_tokens, "output": msg.usage.output_tokens}
})
# Enqueue from your API handler
job = queue.enqueue(process_with_claude, prompt="...", callback_url="https://internal/callback")
Error Handling
| Issue |
Cause |
Fix |
| Stream disconnects |
Network timeout |
Reconnect and re-request (responses are not resumable) |
Batch expired |
Not processed in 24h |
Resubmit the batch |
errored results |
Individual request was invalid |
Check result.error.message per request |
Prerequisites
- Choose an approved sandbox workspace, synthetic documents, bounded batch size, queue with durable retry/dead-letter behavior, and an authenticated internal callback destination.
- Treat SSE as a provider stream and callbacks as application-owned events: Anthropic does not use traditional webhooks for the patterns described here.
- Define an event retention period and redaction policy. Do not log prompts, completions, document content, tool arguments, API keys, or callback secrets.
Instructions
- Validate the stream or batch request against an allowlist of model, source, destination, and maximum size before sending it. Use synthetic fixtures and assert
side_effects=0.
- For SSE, process known event types, preserve ordering by message/block index, and mark a response incomplete until
message_stop. Never assume a disconnected stream is resumable.
- For batches and queues, use stable custom IDs, authenticate internal callbacks, and make result handling idempotent. A duplicate event must not duplicate a write or notification.
- Enforce bounded polling, retries, and queue visibility timeouts. Quarantine errored or expired items for review instead of repeatedly resubmitting unknown data.
- Release from sandbox to a small canary, verify counts, suppression/data-scope checks, and retention cleanup, then roll back the consumer or producer configuration on regression.
Output
Produce an event-processing receipt with correlation/batch ID, event types and counts, succeeded/errored/expired counts, retry/dead-letter counts, callback authentication result, idempotency result, side_effects=0 for tests, canary status, rollback reference, and cleanup status. Include error classes, not raw payloads.
Examples
Submit two synthetic fixtures with custom IDs demo-001 and demo-002, consume each result twice, and assert one stored result per ID. A safe receipt can state batch=redacted; succeeded=2; duplicates_suppressed=2; callbacks_authenticated=true; side_effects=0; cleanup=verified without including document text or generated output.
Resources
Next Steps
For performance optimization, see anth-performance-tuning.
1---2name: anth-webhooks-events3description: Implement event-driven patterns with Claude API: streaming SSE events, Message Batches callbacks, and async processing architectures. Use when building real-time Claude integrations or processing batch results. Trigger with phrases like "anthropic events", "claude streaming events", "anthropic async processing", "claude batch callbacks".4license: MIT5---6# Anthropic Events & Async Processing
7
8## Overview
9
10The Claude API does not use traditional webhooks. Instead it provides two event-driven patterns: Server-Sent Events (SSE) for real-time streaming and the Message Batches API for async bulk processing. This skill covers both.
11
12## SSE Streaming Events
13
14```python
15import anthropic
16
17client = anthropic.Anthropic()
18
19# Process each SSE event type
20with client.messages.stream(
21 model="claude-sonnet-4-20250514",
22 max_tokens=1024,
23 messages=[{"role": "user", "content": "Explain microservices."}]
24) as stream:
25 for event in stream:
26 match event.type:
27 case "message_start":
28 print(f"Started: {event.message.id}")
29 case "content_block_start":
30 if event.content_block.type == "tool_use":
31 print(f"Tool call: {event.content_block.name}")
32 case "content_block_delta":
33 if event.delta.type == "text_delta":
34 print(event.delta.text, end="", flush=True)
35 elif event.delta.type == "input_json_delta":
36 print(event.delta.partial_json, end="")
37 case "message_delta":
38 print(f"\nStop: {event.delta.stop_reason}")
39 print(f"Output tokens: {event.usage.output_tokens}")
40 case "message_stop":
41 print("[Complete]")
42```
43
44## SSE Event Reference
45
46| Event | When | Key Data |
47|-------|------|----------|
48| `message_start` | Stream begins | `message.id`, `message.model`, `message.usage.input_tokens` |
49| `content_block_start` | New block begins | `content_block.type` (text or tool_use), `index` |
50| `content_block_delta` | Incremental content | `delta.text` or `delta.partial_json` |
51| `content_block_stop` | Block finishes | `index` |
52| `message_delta` | Message-level update | `delta.stop_reason`, `usage.output_tokens` |
53| `message_stop` | Stream complete | (empty) |
54| `ping` | Keepalive | (empty) |
55
56## Async Batch Processing
57
58```python
59# Submit batch (up to 100K requests, 50% cheaper)
60batch = client.messages.batches.create(
61 requests=[
62 {
63 "custom_id": f"doc-{i}",
64 "params": {
65 "model": "claude-sonnet-4-20250514",
66 "max_tokens": 1024,
67 "messages": [{"role": "user", "content": f"Summarize: {doc}"}]
68 }
69 }
70 for i, doc in enumerate(documents)
71 ]
72)
73
74# Poll for completion
75import time
76while True:
77 status = client.messages.batches.retrieve(batch.id)
78 if status.processing_status == "ended":
79 break
80 counts = status.request_counts
81 print(f"Processing: {counts.processing} | Done: {counts.succeeded} | Errors: {counts.errored}")
82 time.sleep(30)
83
84# Stream results
85for result in client.messages.batches.results(batch.id):
86 if result.result.type == "succeeded":
87 print(f"[{result.custom_id}]: {result.result.message.content[0].text[:100]}")
88 else:
89 print(f"[{result.custom_id}] ERROR: {result.result.error}")
90```
91
92## Event-Driven Architecture Pattern
93
94```python
95# Use queues to decouple Claude requests from user-facing endpoints
96from redis import Redis
97from rq import Queue
98
99redis = Redis()
100queue = Queue(connection=redis)
101
102def process_with_claude(prompt: str, callback_url: str):
103 """Background job for async Claude processing."""
104 client = anthropic.Anthropic()
105 msg = client.messages.create(
106 model="claude-sonnet-4-20250514",
107 max_tokens=1024,
108 messages=[{"role": "user", "content": prompt}]
109 )
110 # Notify your system via internal callback
111 import requests
112 requests.post(callback_url, json={
113 "text": msg.content[0].text,
114 "usage": {"input": msg.usage.input_tokens, "output": msg.usage.output_tokens}
115 })
116
117# Enqueue from your API handler
118job = queue.enqueue(process_with_claude, prompt="...", callback_url="https://internal/callback")
119```
120
121## Error Handling
122
123| Issue | Cause | Fix |
124|-------|-------|-----|
125| Stream disconnects | Network timeout | Reconnect and re-request (responses are not resumable) |
126| Batch `expired` | Not processed in 24h | Resubmit the batch |
127| `errored` results | Individual request was invalid | Check `result.error.message` per request |
128
129## Prerequisites
130
131- Choose an approved sandbox workspace, synthetic documents, bounded batch size, queue with durable retry/dead-letter behavior, and an authenticated internal callback destination.
132- Treat SSE as a provider stream and callbacks as application-owned events: Anthropic does not use traditional webhooks for the patterns described here.
133- Define an event retention period and redaction policy. Do not log prompts, completions, document content, tool arguments, API keys, or callback secrets.
134
135## Instructions
136
1371. Validate the stream or batch request against an allowlist of model, source, destination, and maximum size before sending it. Use synthetic fixtures and assert `side_effects=0`.
1382. For SSE, process known event types, preserve ordering by message/block index, and mark a response incomplete until `message_stop`. Never assume a disconnected stream is resumable.
1393. For batches and queues, use stable custom IDs, authenticate internal callbacks, and make result handling idempotent. A duplicate event must not duplicate a write or notification.
1404. Enforce bounded polling, retries, and queue visibility timeouts. Quarantine errored or expired items for review instead of repeatedly resubmitting unknown data.
1415. Release from sandbox to a small canary, verify counts, suppression/data-scope checks, and retention cleanup, then roll back the consumer or producer configuration on regression.
142
143## Output
144
145Produce an event-processing receipt with correlation/batch ID, event types and counts, succeeded/errored/expired counts, retry/dead-letter counts, callback authentication result, idempotency result, `side_effects=0` for tests, canary status, rollback reference, and cleanup status. Include error classes, not raw payloads.
146
147## Examples
148
149Submit two synthetic fixtures with custom IDs `demo-001` and `demo-002`, consume each result twice, and assert one stored result per ID. A safe receipt can state `batch=redacted; succeeded=2; duplicates_suppressed=2; callbacks_authenticated=true; side_effects=0; cleanup=verified` without including document text or generated output.
150
151## Resources
152
153- [Streaming API](https://docs.anthropic.com/en/api/messages-streaming)
154- [Message Batches API](https://docs.anthropic.com/en/api/creating-message-batches)
155
156## Next Steps
157
158For performance optimization, see `anth-performance-tuning`.