csrd-delegate
HTTP client delegate base class with retry support for FastAPI microservices.
Package: csrd.delegate · Import: from csrd.delegate import BaseDelegate
What's included
BaseDelegate— async HTTP client with header forwarding, retry via tenacity, and lifecycle (close()/async with)- Response parsing via
csrd.models.model_parser - Configurable retry profiles (
conservative,aggressive,resilient) - httpx-specific response types (
ResponseHandler,ResponseHandlerMap)
Stateless client & reverse-proxy helpers
from csrd.delegate import (
stateless_async_client, # httpx.AsyncClient with no cookie persistence
no_cookie_jar, # CookieJar(policy=NoCookiesPolicy())
proxy_request_headers, # inbound -> upstream header filter (+ x-internal-* strip)
proxy_response_headers, # upstream -> client raw header pairs (multi-value)
has_request_body, # stream request.stream() only when a body exists
)
client = stateless_async_client(timeout=30.0)
Any httpx.AsyncClient shared across inbound requests must not keep a cookie
jar: httpx stores every Set-Cookie and replays it on later requests, leaking
one user's session to another. The csrd compose gateway scaffold uses these
helpers.
Installation
uv pip install "csrd-delegate @ git+ssh://git@github.com/csrd-api/fastapi-common.git#subdirectory=packages/delegate"
Dependencies
csrd-models,csrd-context(Tier 2)
Consumer Feature Matrix (BaseDelegate)
| Capability area | Common expectation | Actual support | How to use / notes |
|---|---|---|---|
| HTTP methods | Standard verb helpers exist | ✅ Built-in | get, post, put, patch, delete, head, options |
| Retry support | Automatic retries for transient failures | ✅ Built-in + configurable | Set default in constructor (retry_profile / retry_*) or per call |
| Retry presets | Named retry policies | ✅ Built-in | conservative, aggressive, resilient |
| Auth forwarding | Bearer token from incoming request is forwarded | ✅ Built-in | Uses request context headers; can be filtered explicitly |
| Header filtering | Hop-by-hop headers are removed safely | ✅ Built-in | RFC-hop headers filtered by default; body requests additionally filter content-length |
| Response parsing | Auto-parse JSON to typed model | ✅ Built-in + configurable | Use response_model and/or model_handler |
| Status handling | Non-2xx statuses raise exceptions | ✅ Built-in + overridable | Default raises HTTPException; override specific statuses with response_handlers |
| Per-status custom behavior | Custom handling for 404/409/etc | ✅ Built-in | Pass response_handlers={404: ...} style map |
| Client lifecycle | Safe async with usage and close management |
✅ Built-in | Delegate-owned client closes automatically; injected client is not closed |
| Per-call transport overrides | Override timeout/headers/auth per request | ✅ Built-in | timeout, headers, auth, follow_redirects, etc. on each call |
| Circuit breaker | Open/half-open breaker semantics | ❌ Not built-in | Add externally (middleware/proxy/custom wrapper) |
| Rate limiting | Outbound request throttling | ❌ Not built-in | Add externally (gateway/proxy/custom transport) |
| Service discovery | Dynamic upstream resolution/registry | ❌ Not built-in | Provide concrete service_host at wiring time |
Inter-Service Auth Propagation
In multi-service clusters, delegates automatically forward authentication headers from the inbound request to upstream service calls. This enables transparent claim propagation through delegate chains.
How It Works
- Client → Service A: Client sends JWT via
Authorization: Bearer <token>header - Service A extracts and verifies claims using
VerifiedTokenDep(or similar) - Service A → Service B (via
BaseDelegate): Delegate automatically includesAuthorizationheader from inbound request - Service B receives and verifies same JWT from Service A's outbound call
- Service B → Service C (if needed): Claim propagation continues transitively
Example: Multi-Hop Orchestration
# quote-service orchestrates inventory + pricing calls
class QuoteDelegate(BaseDelegate):
def __init__(self) -> None:
super().__init__(settings.quote_service_url, retry_profile="conservative")
async def compose_quote(self, item_id: str, quantity: int):
# All three calls automatically include Authorization header
inventory_data = await self.get(f"/api/inventory/{item_id}")
pricing_data = await self.get(f"/api/pricing/{item_id}")
return {
"item_id": item_id,
"quantity": quantity,
"stock_available": inventory_data["available_qty"],
"price": pricing_data["base_price"],
}
# On inbound request with JWT:
@router.post("/api/quotes")
async def create_quote(
item_id: str,
quantity: int,
verified: VerifiedTokenDep, # JWT verified here
quote_delegate: QuoteDelegate,
) -> dict:
# QuoteDelegate.compose_quote() calls use verified user's JWT
return await quote_delegate.compose_quote(item_id, quantity)
Header Filtering Behavior
- Hop-by-hop headers (RFC 7230): Automatically filtered (connection, keep-alive, etc.)
- Authorization header: Always forwarded unless explicitly filtered
- Content-Length: Re-computed by httpx for each request
- Custom headers: Passed through by default; filter via
header_filter_listif needed
Cookie isolation
A delegate's own client is shared by every inbound request, so it is built
with no_cookie_jar(): response Set-Cookie headers are never stored and
never replayed onto later calls. Pass cookies explicitly per call if needed.
If you inject your own client=, build it with stateless_async_client().
Pattern Constraints
- No auth bypass: Each hop validates the JWT independently
- Claim consistency: All services see the same
sub,roles, etc. - No credential injection: Delegates forward inbound claims; the only credential they add is the opt-in internal token (below)
- No header injection: Custom headers must be passed explicitly; inbound headers only forwarded if present
Service-to-Service Internal Token
Pass internal_token to send a shared token on every call, alongside the
forwarded Authorization header. The receiving service guards the route with
csrd.auth.require_internal_token.
class CatalogDelegate(BaseDelegate):
def __init__(self) -> None:
super().__init__(
settings.catalog_url,
internal_token=settings.internal_token,
internal_token_header="X-Internal-Token", # default; configurable
)
Noneor""disables it (no header is sent).- An inbound value of the internal header is never relayed downstream, whether or not a token is configured.
- A per-call
headers=value for the same header still takes precedence.
When to Use Manual Header Control
Override the automatic forwarding only in specific cases:
class InternalDelegate(BaseDelegate):
def __init__(self) -> None:
super().__init__(
settings.internal_service_url,
header_filter_list=["authorization", "x-tenant-id"], # Don't forward these
ignore_incoming_headers=False, # Still use get_headers() for others
)
1 maintainer
Metadata
1 maintainer
Release files for csrd-delegate 0.7.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| csrd_delegate-0.7.0.tar.gz | 14.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| csrd_delegate-0.7.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 26.1 kB
Release files / csrd_delegate-0.7.0.tar.gz
| Download URL | csrd_delegate-0.7.0.tar.gz |
|---|---|
| Size | 14.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
21707c41a83a49630c4af37c949ae7383e058baecc1f8af756e9ab0a766715be
|
|
BLAKE2b-256 checksum How to use checksums |
33ccc3dfb602d504dec8156508bb9f7a76bc687e734a229d3ce2478db013c5a6
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency logRelease files / csrd_delegate-0.7.0-py3-none-any.whl
| Download URL | csrd_delegate-0.7.0-py3-none-any.whl |
|---|---|
| Size | 11.9 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
5f79870ab44665bf7e4d2fd85fcdfd85ef0db6fd0ea7adef084ec91a0538f992
|
|
BLAKE2b-256 checksum How to use checksums |
e569e6ce5e3cb653c76cb7e3579837c82f727b7367acb27dfdb91a6f90a38ab8
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency log
