Python sibling of adcp-client#TBD. Ship a reference middleware Python adopters drop into their `comply_test_controller` implementation to populate the session-scoped recording buffer that `query_upstream_traffic` returns.
Why
PR adcontextprotocol/adcp#3816 ships the spec contract for `upstream_traffic` storyboard checks. Python-based adopters (sales-house adapters, signal marketplaces, governance tools written in FastAPI / Flask) opt in by exposing `query_upstream_traffic` from `comply_test_controller`. The recorder primitive lets them avoid hand-rolling the buffer + redaction + caller-scoping that the spec requires.
What to build
`adcp.testing.upstream_recorder`:
```python
from adcp.testing import UpstreamRecorder
recorder = UpstreamRecorder(
enabled=os.getenv("ADCP_SANDBOX") == "1",
redact_pattern=r"^(authorization|token|api_key|...)$",
buffer_size=1000,
ttl_seconds=3600,
)
Wire as httpx event hook / requests session adapter
import httpx
client = httpx.Client(event_hooks={"request": [recorder.on_request], "response": [recorder.on_response]})
Or for requests:
import requests
requests.Session().mount("https://", recorder.requests_adapter)
In your comply_test_controller handler for scenario: query_upstream_traffic:
def handle_query_upstream_traffic(req, principal):
result = recorder.query(
principal=principal,
since_timestamp=req.params.get("since_timestamp"),
endpoint_pattern=req.params.get("endpoint_pattern"),
limit=req.params.get("limit", 100),
)
return {
"success": True,
"recorded_calls": result.items,
"total_count": result.total,
"truncated": result.truncated,
"since_timestamp": req.params["since_timestamp"],
}
```
Same key behaviors as the Node sibling: caller scoping (per-principal buffer), redaction at recording time, sandbox-only gating, optional purpose tagging callback.
HTTP client coverage
Ship adapters for the common Python HTTP clients:
- `httpx` (event hooks — primary)
- `requests` (transport adapter)
- `aiohttp` (trace config)
- `urllib3` (lowest level — nice-to-have)
References
Python sibling of adcp-client#TBD. Ship a reference middleware Python adopters drop into their `comply_test_controller` implementation to populate the session-scoped recording buffer that `query_upstream_traffic` returns.
Why
PR adcontextprotocol/adcp#3816 ships the spec contract for `upstream_traffic` storyboard checks. Python-based adopters (sales-house adapters, signal marketplaces, governance tools written in FastAPI / Flask) opt in by exposing `query_upstream_traffic` from `comply_test_controller`. The recorder primitive lets them avoid hand-rolling the buffer + redaction + caller-scoping that the spec requires.
What to build
`adcp.testing.upstream_recorder`:
```python
from adcp.testing import UpstreamRecorder
recorder = UpstreamRecorder(
enabled=os.getenv("ADCP_SANDBOX") == "1",
redact_pattern=r"^(authorization|token|api_key|...)$",
buffer_size=1000,
ttl_seconds=3600,
)
Wire as httpx event hook / requests session adapter
import httpx
client = httpx.Client(event_hooks={"request": [recorder.on_request], "response": [recorder.on_response]})
Or for requests:
import requests
requests.Session().mount("https://", recorder.requests_adapter)
In your comply_test_controller handler for scenario: query_upstream_traffic:
def handle_query_upstream_traffic(req, principal):
result = recorder.query(
principal=principal,
since_timestamp=req.params.get("since_timestamp"),
endpoint_pattern=req.params.get("endpoint_pattern"),
limit=req.params.get("limit", 100),
)
return {
"success": True,
"recorded_calls": result.items,
"total_count": result.total,
"truncated": result.truncated,
"since_timestamp": req.params["since_timestamp"],
}
```
Same key behaviors as the Node sibling: caller scoping (per-principal buffer), redaction at recording time, sandbox-only gating, optional purpose tagging callback.
HTTP client coverage
Ship adapters for the common Python HTTP clients:
References