Concurrent Cached Fetch

Concurrent + Cached External Fetching

Sequential, uncached external calls are the single most common avoidable performance sink in agent-written code. Ten lookups that each take 300 ms run in 3 seconds when serialized; fanned out they finish in ~300 ms. And on the next run — a re-analysis, a follow-up question, a re-run after an unrelated bug fix — uncached code pays that cost all over again, plus hammers a public endpoint that may rate-limit or ban you.

This skill exists to make two properties the default for any code that touches the network in bulk:

1. Concurrency — independent calls run in parallel, at least ~10 in flight at once. 2. Disk caching — every response is persisted, keyed by request content, so an identical call is served from disk and never re-issued unless explicitly refreshed.

These are code patterns you write into the product, not something you do by hand. The goal is that the code stays fast and cache-warm every time it runs, including in follow-up work.

When this applies

Reach for this skill whenever the work involves N independent external calls where N is more than a few and the calls don't depend on each other's results. Concrete tells:

• A loop body contains requests.get/post, httpx, urllib, fetch, axios, an SDK

client call, or any HTTP/RPC to a service you don't control.

• "Enrich / look up / resolve / annotate each of these <items>."
• Refactoring code that already loops calls sequentially and feels slow.
• A two-step API (find an ID, then fetch details per ID) — the inner per-ID fetches are

the fan-out.

If the calls are genuinely dependent (call B needs call A's result), you can't parallelize those two, but you can usually still parallelize across the outer items.

If only one or two calls happen total, don't over-engineer — a single cached call is fine, concurrency buys nothing.

Apply this even when the user never says "slow" or "cache". A prompt like "look up the details for each ID in this list" or "get the price for each of these 80 SKUs" is a bulk-fetch task — the fan-out and the disk cache are the right default, not an optimization to bolt on later when someone complains. The model will usually parallelize on its own once a task is framed as slow; the durable win this skill adds is the persistent, content-keyed disk cache plus the hygiene around it (don't cache failures, atomic writes, gitignore), which survives across runs and follow-up analysis. That payoff only happens if you build it in from the first version, so reach for this skill the moment you see "for each <item>: call <external service>", regardless of how the request is phrased.

The two non-negotiables (and the why)

1. Fan out — at least ~10 concurrent

Serialized network I/O wastes wall-clock time doing nothing but waiting on sockets. The fix is to issue independent requests concurrently with a bounded worker pool. Bounded matters: unbounded concurrency over a public API gets you throttled or blocked, and exhausts local file descriptors. A cap of ~10–20 is the sweet spot for most public endpoints — enough to collapse the wait, polite enough not to trip rate limits.

Pick the idiom that fits the codebase's language and existing style — see references/patterns.md for ready-to-adapt implementations in Python (threads for the common requests-style blocking client; asyncio for httpx/aiohttp), JS/TS, Go, and Java. Match what the project already uses rather than introducing a new async stack.

2. Cache every response to disk, keyed by request content

The cache key is a hash of everything that determines the response — method, full URL, query params, and (for POST) the body. Identical request → same key → served from disk. This makes re-runs and follow-up analysis instant and keeps you off the wire.

Default cache policy: content-keyed, no expiry. Entries do not auto-expire. They are reused indefinitely until explicitly invalidated, because the data these calls return (reference data, code systems, catalogs, documentation) is typically stable on the timescale of a work session and re-fetching it buys nothing. Provide a single escape hatch — a refresh=True argument or a CACHE_BYPASS=1 env var — that forces a live call and overwrites the cached entry, for the rare case where you know the upstream changed. Don't build TTL/expiry machinery unless the data is genuinely time-sensitive; for stable reference data it's complexity you don't need.

Cache location and hygiene:

• Store under a project-local dir such as .cache/api/ (or honor an existing project

cache convention if one exists).

• Add the cache dir to `.gitignore` — cached responses are derived data, never

committed.

• Write atomically (temp file + rename) so a crashed run can't leave a half-written

entry that later parses as valid.

• Only cache successful responses. Caching an error (a 500, a timeout) poisons the cache

— a later run would replay the failure forever. On failure, don't write; let the next run retry live.

See references/patterns.md for a drop-in disk-cache wrapper in each language.

How to apply it

1. Spot the fan-out. Identify the collection being iterated and confirm the calls are independent. That collection is what you parallelize over. 2. Wrap the single call in a cached fetch. Factor the one-item network call into a function fetch(request) -> response that checks disk first, calls live on a miss, and writes the result on success. This keeps caching in one place. 3. Run the collection through a bounded pool. Map the cached fetch over all items with a concurrency cap (~10–20). Preserve input order in the results if downstream code expects it. 4. Keep the existing error contract. If the surrounding code has a convention for failures (e.g. tools that always return a structured result and never raise), preserve it per item. One item's failure must not abort the whole batch or corrupt the cache. 5. Make the cache visible and bypassable. Ensure the cache dir is gitignored and the refresh/CACHE_BYPASS escape hatch works.

Worked trigger: a two-step lookup service

The canonical case is a "find, then fetch details per result" tool. Picture a function that resolves a search term to a list of IDs, then issues one `requests.get` per ID sequentially in a loop to fetch each ID's details — and nothing anywhere is cached, so every run re-hits the upstream service for terms it already resolved seconds ago.

The fix has two parts:

• A shared cached-GET helper used by both the term→IDs call and every per-ID details

call, so identical requests are served from disk.

• A bounded thread pool over the per-ID inner calls (and over batches of terms when

several are looked up at once), so the fan-out runs ~10–20 in flight instead of one at a time.

Keep the tool's existing return shape and error encoding — wrap the network helpers, don't change the public contract. See references/patterns.md → "Python: threaded + cached requests" for the shape to apply.

Anti-patterns to avoid

• Sequential loop of network calls with no concurrency — the thing this skill

replaces.

• Unbounded concurrency — spawning a task per item with no cap; trips rate limits and

exhausts fds.

• In-memory-only caching (a dict that dies with the process) — doesn't survive across

runs or follow-up sessions, which is the whole point here.

• Caching failures — persisting error responses, which replays them forever.
• TTL/expiry scaffolding for stable data — for reference data that doesn't change on

a session timescale, cache without expiry and expose a manual refresh instead.

• Committing the cache — derived data doesn't belong in version control.

Reference

references/patterns.md — ready-to-adapt concurrent + disk-cache implementations per language (Python threads, Python asyncio, JS/TS, Go, Java), plus the atomic-write and content-key-hashing snippets. Read it when you're about to write the actual code so you match the project's language and HTTP client.

Concurrent + Cached Fetch — Implementation Patterns

Ready-to-adapt building blocks. Pick the language/HTTP-client pair that matches the project. Each pattern combines a bounded concurrency primitive with a content-keyed disk cache (no expiry, manual bypass). Adapt names and error handling to the surrounding code; don't paste verbatim if the project has its own conventions.

Table of contents

• Cache key + atomic write (language-agnostic recipe)
• Python: threaded + cached `requests` ← most common
• Python: asyncio + cached `httpx`
• JS/TS: `Promise` pool + cached `fetch`
• Go: bounded goroutines + cached `net/http`
• Java: `ExecutorService` + cached `HttpClient`

---

Cache key + atomic write

The key is a stable hash of everything that determines the response. The write is atomic so a crash never leaves a corrupt entry that later reads as valid.

• Key inputs: HTTP method, full URL, sorted query params, and request body (for

POST/PUT). Serialize them canonically (sorted keys), then SHA-256.

• Filename: <cache_dir>/<hexdigest>.json.
• Atomic write: write to <file>.tmp.<pid>, then rename() onto the final path —

rename is atomic on POSIX, so readers see either the old file or the complete new one, never a partial.

• No expiry: if the file exists, use it. The only bypass is an explicit

refresh/CACHE_BYPASS flag, which forces a live call and overwrites.

• Never cache failures: only write after a confirmed-success response.

---

Python: threaded + cached requests

The default for code already using the blocking requests library. Threads are the right tool — these calls are I/O-bound, so the GIL is released during the socket wait and threads give real concurrency.

import os, json, hashlib, threading
from concurrent.futures import ThreadPoolExecutor
import requests

CACHE_DIR = os.path.join(os.path.dirname(__file__), ".cache", "api")
os.makedirs(CACHE_DIR, exist_ok=True)
_MAX_WORKERS = 12  # bounded: enough to collapse wait, polite to public APIs

def _cache_key(method, url, params=None, body=None):
    blob = json.dumps(
        {"m": method.upper(), "u": url, "p": params or {}, "b": body},
        sort_keys=True, separators=(",", ":"),
    )
    return hashlib.sha256(blob.encode()).hexdigest()

def cached_get(url, params=None, *, refresh=False, timeout=10):
    """GET with content-keyed disk cache (no expiry). Returns parsed JSON or text.
    Only successful responses are cached; failures raise so the caller can apply
    its own error contract and so the failure is never persisted."""
    bypass = refresh or os.environ.get("CACHE_BYPASS") == "1"
    key = _cache_key("GET", url, params)
    path = os.path.join(CACHE_DIR, f"{key}.json")

    if not bypass and os.path.exists(path):
        with open(path) as f:
            return json.load(f)

    resp = requests.get(url, params=params, timeout=timeout)
    resp.raise_for_status()  # don't cache errors — let the caller retry live next run
    try:
        data = resp.json()
    except ValueError:
        data = {"_text": resp.text}

    tmp = f"{path}.tmp.{os.getpid()}.{threading.get_ident()}"
    with open(tmp, "w") as f:
        json.dump(data, f)
    os.replace(tmp, path)  # atomic
    return data

def fetch_all(items, fetch_one, max_workers=_MAX_WORKERS):
    """Map fetch_one over items concurrently, preserving input order.
    fetch_one(item) should return a result or raise; exceptions are captured
    per-item so one failure never aborts the batch."""
    results = [None] * len(items)
    def work(i_item):
        i, item = i_item
        try:
            results[i] = fetch_one(item)
        except Exception as e:
            results[i] = {"error": str(e), "item": item}
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        list(pool.map(work, enumerate(items)))
    return results

Applied to a two-step lookup tool: the per-ID details loop becomes fetch_all(id_list, lambda i: cached_get(DETAILS_API_URL.format(id=i))), and the top-level term→IDs lookup uses cached_get. Keep the tool's existing return shape and error encoding — wrap the helpers, don't change the public contract.

---

Python: asyncio + cached httpx

When the project is already async (FastAPI, httpx.AsyncClient, aiohttp). A Semaphore provides the bound; asyncio.gather fans out.

import os, json, hashlib, asyncio, httpx

CACHE_DIR = os.path.join(os.path.dirname(__file__), ".cache", "api")
os.makedirs(CACHE_DIR, exist_ok=True)
_SEM = asyncio.Semaphore(12)  # bounded concurrency

def _cache_key(method, url, params=None, body=None):
    blob = json.dumps({"m": method.upper(), "u": url, "p": params or {}, "b": body},
                      sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(blob.encode()).hexdigest()

async def cached_get(client, url, params=None, *, refresh=False):
    bypass = refresh or os.environ.get("CACHE_BYPASS") == "1"
    key = _cache_key("GET", url, params)
    path = os.path.join(CACHE_DIR, f"{key}.json")
    if not bypass and os.path.exists(path):
        return json.load(open(path))
    async with _SEM:  # cap in-flight requests
        resp = await client.get(url, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    tmp = f"{path}.tmp.{os.getpid()}"
    json.dump(data, open(tmp, "w"))
    os.replace(tmp, path)
    return data

async def fetch_all(items, build_call):
    """build_call(client, item) -> coroutine. Returns results in input order;
    failures become {'error': ...} instead of aborting the batch."""
    async with httpx.AsyncClient() as client:
        async def guarded(item):
            try:
                return await build_call(client, item)
            except Exception as e:
                return {"error": str(e), "item": item}
        return await asyncio.gather(*(guarded(it) for it in items))

---

JS/TS: Promise pool + cached fetch

No native bounded-pool primitive, so cap manually (a small worker-drain loop, or a library like p-limit if it's already a dependency). Cache to disk with fs.

import { createHash } from "node:crypto";
import { promises as fs } from "node:fs";
import * as path from "node:path";

const CACHE_DIR = path.join(process.cwd(), ".cache", "api");
const MAX_CONCURRENCY = 12;

const cacheKey = (method: string, url: string, params?: object, body?: unknown) =>
  createHash("sha256")
    .update(JSON.stringify({ m: method.toUpperCase(), u: url, p: params ?? {}, b: body ?? null }))
    .digest("hex");

export async function cachedGet(url: string, refresh = false): Promise<any> {
  await fs.mkdir(CACHE_DIR, { recursive: true });
  const bypass = refresh || process.env.CACHE_BYPASS === "1";
  const file = path.join(CACHE_DIR, `${cacheKey("GET", url)}.json`);
  if (!bypass) {
    try { return JSON.parse(await fs.readFile(file, "utf8")); } catch { /* miss */ }
  }
  const resp = await fetch(url);
  if (!resp.ok) throw new Error(`HTTP ${resp.status}`); // don't cache failures
  const data = await resp.json();
  const tmp = `${file}.tmp.${process.pid}`;
  await fs.writeFile(tmp, JSON.stringify(data));
  await fs.rename(tmp, file); // atomic
  return data;
}

// Bounded fan-out preserving input order.
export async function fetchAll<T, R>(items: T[], one: (x: T) => Promise<R>,
                                     limit = MAX_CONCURRENCY): Promise<(R | { error: string })[]> {
  const results = new Array(items.length);
  let next = 0;
  const worker = async () => {
    while (next < items.length) {
      const i = next++;
      try { results[i] = await one(items[i]); }
      catch (e: any) { results[i] = { error: String(e?.message ?? e) }; }
    }
  };
  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker));
  return results;
}

---

Go: bounded goroutines + cached net/http

A buffered channel is the idiomatic semaphore; sync.WaitGroup joins.

package fetch

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"sync"
)

const maxConcurrency = 12

var cacheDir = filepath.Join(".cache", "api")

func cacheKey(method, url string) string {
	sum := sha256.Sum256([]byte(method + " " + url))
	return hex.EncodeToString(sum[:])
}

// CachedGet returns the raw body, served from disk on a hit. Failures are not cached.
func CachedGet(url string, refresh bool) ([]byte, error) {
	_ = os.MkdirAll(cacheDir, 0o755)
	path := filepath.Join(cacheDir, cacheKey("GET", url)+".json")
	if !refresh && os.Getenv("CACHE_BYPASS") != "1" {
		if b, err := os.ReadFile(path); err == nil {
			return b, nil
		}
	}
	resp, err := http.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("HTTP %d", resp.StatusCode) // don't cache errors
	}
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	tmp := fmt.Sprintf("%s.tmp.%d", path, os.Getpid())
	if err := os.WriteFile(tmp, body, 0o644); err == nil {
		_ = os.Rename(tmp, path) // atomic
	}
	return body, nil
}

// FetchAll maps one() over items with bounded concurrency, preserving order.
func FetchAll[T any, R any](items []T, one func(T) (R, error)) []R {
	results := make([]R, len(items))
	sem := make(chan struct{}, maxConcurrency)
	var wg sync.WaitGroup
	for i, item := range items {
		wg.Add(1)
		go func(i int, item T) {
			defer wg.Done()
			sem <- struct{}{}        // acquire
			defer func() { <-sem }() // release
			r, _ := one(item)        // apply your own per-item error handling
			results[i] = r
		}(i, item)
	}
	wg.Wait()
	return results
}

(For body-encoded JSON output, marshal the cached bytes back via json.Unmarshal.)

---

Java: ExecutorService + cached HttpClient

A fixed thread pool bounds concurrency; invokeAll fans out and preserves order.

import java.net.URI;
import java.net.http.*;
import java.nio.file.*;
import java.security.MessageDigest;
import java.util.*;
import java.util.concurrent.*;

public final class CachedFetch {
    private static final Path CACHE_DIR = Path.of(".cache", "api");
    private static final int MAX_CONCURRENCY = 12;
    private static final HttpClient CLIENT = HttpClient.newHttpClient();

    private static String cacheKey(String method, String url) throws Exception {
        var md = MessageDigest.getInstance("SHA-256");
        byte[] d = md.digest((method + " " + url).getBytes());
        var sb = new StringBuilder();
        for (byte b : d) sb.append(String.format("%02x", b));
        return sb.toString();
    }

    /** GET with content-keyed disk cache (no expiry). Failures are not cached. */
    public static String cachedGet(String url, boolean refresh) throws Exception {
        Files.createDirectories(CACHE_DIR);
        Path path = CACHE_DIR.resolve(cacheKey("GET", url) + ".json");
        boolean bypass = refresh || "1".equals(System.getenv("CACHE_BYPASS"));
        if (!bypass && Files.exists(path)) return Files.readString(path);

        var req = HttpRequest.newBuilder(URI.create(url)).GET().build();
        HttpResponse<String> resp = CLIENT.send(req, HttpResponse.BodyHandlers.ofString());
        if (resp.statusCode() != 200) throw new RuntimeException("HTTP " + resp.statusCode());

        Path tmp = CACHE_DIR.resolve(path.getFileName() + ".tmp." + ProcessHandle.current().pid());
        Files.writeString(tmp, resp.body());
        Files.move(tmp, path, StandardCopyOption.ATOMIC_MOVE);
        return resp.body();
    }

    /** Map one() over items with bounded concurrency, preserving input order. */
    public static <T, R> List<R> fetchAll(List<T> items, java.util.function.Function<T, R> one)
            throws InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(MAX_CONCURRENCY);
        try {
            List<Callable<R>> tasks = new ArrayList<>();
            for (T item : items) tasks.add(() -> one.apply(item));
            List<Future<R>> futures = pool.invokeAll(tasks); // order preserved
            List<R> out = new ArrayList<>();
            for (Future<R> f : futures) {
                try { out.add(f.get()); } catch (ExecutionException e) { out.add(null); }
            }
            return out;
        } finally {
            pool.shutdown();
        }
    }
}

---

Checklist before you call the fetch code done

• [ ] Independent calls run concurrently with a bounded pool (~10–20).
• [ ] Every successful response is written to a disk cache, keyed by request content.
• [ ] Cache has no expiry; a refresh arg / CACHE_BYPASS=1 env var forces a live call.
• [ ] Failures are not cached; one item's error doesn't abort the batch.
• [ ] Cache writes are atomic (temp + rename).
• [ ] Cache dir is in `.gitignore`.
• [ ] The surrounding code's error/return contract is preserved per item.