Skip to content
🎉 Launch sale:50% off everything over $22 — automatically applied at checkout· ends Aug 2Shop the sale →
DevOps AI ToolKit
Newsletter
All prompts
AI for Bash & Python Automation Difficulty: Advanced ClaudeChatGPT

Async Concurrent HTTP Poller with asyncio and httpx Prompt

Build a fast, bounded-concurrency async poller/fetcher that hits many endpoints with asyncio and httpx, with rate limiting, retries, timeouts, and structured results.

Target user
Python engineers fetching or polling many HTTP endpoints concurrently
Difficulty
Advanced
Tools
Claude, ChatGPT

The prompt

You are a senior Python engineer who writes high-throughput async I/O clients that stay polite to upstream services and never leak connections.

I will provide:
- The list/source of URLs (static list, paginated API, or a queue)
- Auth requirements (bearer, mTLS, query keys)
- Upstream rate limits and SLA expectations
- What "done" looks like (one-shot fetch vs continuous poll)

Produce a single, runnable module using `asyncio` + `httpx.AsyncClient`. Requirements:

1. **Bounded concurrency** — use an `asyncio.Semaphore` (configurable, default 10), never an unbounded `gather` over thousands of tasks. Reuse ONE `AsyncClient` with an explicit `limits=httpx.Limits(...)` connection pool.

2. **Timeouts** — set per-request `httpx.Timeout` (connect, read, write, pool). No request may hang forever.

3. **Retries with backoff** — retry only on 429, 5xx, and transport errors. Honor `Retry-After`. Exponential backoff with full jitter, capped attempts. Do NOT retry 4xx (except 429).

4. **Rate limiting** — token-bucket or simple delay to respect upstream RPS. Make it a reusable async limiter, not sleeps scattered in the code.

5. **Structured results** — return a dataclass per URL: url, status, elapsed_ms, attempt_count, ok flag, and either parsed body or error. Never let one failure abort the batch; gather with `return_exceptions` or per-task try/except.

6. **Polling mode** — when continuous, poll on an interval with graceful shutdown on SIGINT/SIGTERM (cancel tasks, drain, close client).

7. **Observability** — structured log lines (JSON optional) and a summary: total, ok, failed, p50/p95 latency.

8. **Backpressure** — if the source is a stream/queue, use an `asyncio.Queue` with worker tasks rather than loading everything into memory.

Output: (a) the full module with type hints, (b) a `main()` guarded by `if __name__ == "__main__"`, (c) a tiny `pytest` test using `respx`/`httpx.MockTransport`, (d) notes on tuning concurrency vs connection-pool size.

Bias toward: one shared client, explicit timeouts everywhere, idempotent retries only, clean cancellation. Avoid `requests` in async code and avoid `asyncio.run` inside libraries.

Run this prompt with AI

Test it, get an AI-improved version, or compare models — live in the Prompt Workspace. No copy-paste.

Related prompts

More Bash & Python Automation prompts & error guides

Browse every Bash & Python Automation prompt and troubleshooting guide in one place.

Free download · 368-page PDF

Reading prompts? Get all 500 in one free PDF

500 battle-tested, copy-paste AI prompts engineered by a senior systems engineer — every one with fill-in placeholders and safety/back-out notes. Drop your email and it's yours.

  • 500 prompts: Linux · Kubernetes · Terraform · OpenStack · GitLab · Docker · Monitoring · Incident Response
  • Instant PDF download — yours free, forever
  • Plus one practical AI-workflow email a week (no spam)

Single opt-in · unsubscribe anytime · no spam.