needlepath — Python client

The framework-free core client for the Needlepath hosted context-selection service. HTTP only, zero required dependencies, one transport/auth seam.

Framework integrations are separate distributions and depend on this one:

Distribution	Seam
needlepath-langchain	AgentMiddleware (wrap_model_call, wrap_tool_call) + the LangGraph recipe
needlepath-litellm	LiteLLM proxy CustomGuardrail
llama-index-postprocessor-needlepath	BaseNodePostprocessor

Install

pip install needlepath
pip install "needlepath[httpx]"   # optional: httpx-backed async transport

Python 3.9+. No required runtime dependencies — the default transport is stdlib urllib, so installing this client can never move a version you already pinned.

Quickstart

from needlepath import NeedlepathClient, ContextRecord, TaskSpec

client = NeedlepathClient(
    api_key="np_live_...",  # or NEEDLEPATH_API_KEY
    operating_point="np-2026-08-r4",  # required — see "Pin the operating point"
)

result = client.select(
    records=[
        ContextRecord(text=tool_output, kind="tool_result", id="lookup-1"),
        ContextRecord(text=doc, kind="external_data", id="doc-7", title="SOP"),
    ],
    task=TaskSpec(prompt="what is the invoice total for A-1002?"),
    max_context_tokens=4000,
)

context = result.rendered_context if result.applied else my_original_context

select() never raises. Read result.applied; when it is False, use your original context. That is the whole integration contract.

Async

from needlepath import AsyncNeedlepathClient

async with AsyncNeedlepathClient(operating_point="np-2026-08-r4") as client:
    result = await client.select(records=..., task=..., max_context_tokens=4000)

Identical semantics — the decision logic is literally the same functions. Uses httpx.AsyncClient when httpx is importable, a worker thread otherwise; it never blocks the event loop either way.

R4 trace and outcome attribution

Pin np-2026-08-r4 and opt into the metadata-only trace when you need to inspect record dispositions:

result = client.select(
    records=records,
    task=TaskSpec(prompt=question),
    max_context_tokens=4000,
    return_trace=True,
)

if result.response and result.response.selection_trace:
    for item in result.response.selection_trace.records:
        print(item.record_id, item.disposition, item.reason)

The hosted response also supplies result.selection_id, a server-issued id separate from your request id. Report downstream quality after the model call:

if result.selection_id:
    client.report_outcome(
        result.selection_id,
        task_success=True,
        context_sufficient=True,
        evaluation_source="application",
    )

report_outcome() is idempotent and independently retryable. It can raise on a callback failure; that never changes select() or the context already returned. Do not send prompts, context, model output, answers, or free-form notes.

Configuration

Argument	Environment	Default
api_key	NEEDLEPATH_API_KEY	none (no Authorization header)
operating_point	NEEDLEPATH_OPERATING_POINT	required
base_url	NEEDLEPATH_BASE_URL	https://api.nextmoca.com
timeout	—	10.0 s, total, retries included
max_retries	—	2
max_request_bytes	—	6 * 1024 * 1024
shadow	—	False

The base URL is a documented default, not a hard-coded one: a durable custom domain replaces it, and swapping it is a config change rather than a release.

http:// is refused for any non-loopback host. TLS-only is an inherited binding constraint; loopback is exempt so the localhost benchmark adapter still works.

The four binding rules, and where they live

1. Fail open, always

Timeout, 5xx, a violated contract, an empty selection, a stood-down gate — every one of them returns a SelectionResult with applied=False, the exception on .error, and a machine-readable .reason. Nothing is raised on the selection path.

if not result.applied:
    log.info("needlepath passthrough", extra=result.metadata())
    context = my_original_context

result.reason is an open enum (needlepath.Reason). Today:

reason	applied	meaning
ok	yes	safe to apply
shadow	no	shadow mode — measured, deliberately not applied
disabled	no	the integration was constructed disabled
no_records	no	nothing to select over; no call was made
below_trigger	no	under the configured trigger; no call was made
escalated	no	the engine declined: full context over capacity. Charged at the full rate, unlike a free stand-down
empty_selection	no	zero records selected
engine_fallback	no	fallback_used — the engine wants the full context
selection_error	no	the response carried a selection_error
no_per_record_detail	no	per-record detail was required and absent
payload_too_large	no	over the client-side guard; no call was made
timeout / transport_error / http_error / contract_error	no	error paths
unexpected_error	no	a defect in this client — still fails open

Treat an unrecognized reason as "do not apply". result.failed separates the error paths from the benign declines, which is the distinction that decides whether a passthrough deserves a WARNING or a DEBUG.

The empty-selection rule is not a nicety. NEXPE-413 is the precedent: the engine returned an empty selection as a success, and at a framework boundary that has to degrade to passthrough, never to an empty context.

select_or_raise() exists for the two cases where fail-open is wrong — diagnostics, and a caller that is itself a benchmark and must not silently substitute a passthrough for a measurement. Do not use it in an adapter.

2. Pin the operating point

operating_point is required at construction, and this is the one failure that raises rather than fails open. Failing open on a bad request is degradation a customer absorbs; failing open on "no operating point was pinned" accepts exactly the behaviour drift the rule exists to prevent. It fails at wire-up, never under load.

Labels are immutable: np-2026-08-r4 means one frozen configuration forever, and a retune mints a new label. client.operating_points() lists what the endpoint serves — call it once at wire-up if you want the assertion, never per request.

3. Guard the 6 MB cap client-side

The serialized body is measured before it is sent. Over the guard, you get reason="payload_too_large" and no network call, rather than a platform error from the far side of the wire.

The guard measures the JSON body. Lambda's 6 MB limit applies to the whole invocation payload — the API Gateway event envelope sits on top — so a request that passes at 5.99 MB can still be rejected upstream. Lower max_request_bytes if you run near the cap.

4. Shadow mode is a one-line switch

client = NeedlepathClient(operating_point="np-2026-08-r4", shadow=True)

The call is made exactly as a live call would be — same body, same billing, same operating point — the full response comes back and is reported, and the selection is never applied. Measure what Needlepath would have saved you; change nothing.

budget.mode = "shadow" is not on the wire yet. BudgetSpec.mode on the published contract accepts fixed and adaptive only; adding shadow is a public-repo contract change (§0.1, sequenced in E7). The wire half is behind needlepath.features.WIRE_SHADOW_MODE; flip it in the same change that lands the field. Client-side behaviour does not change when you do — it is already correct.

Retries

Retried: 429, 502, 503, 504, and transport failures. Retry-After is honoured in delta-seconds form. Backoff is full-jitter exponential, and every retry is bounded by the timeout deadline — retries can never exceed the budget the caller gave you.

500 is not retried. A 500 from this endpoint is engine_error: a deterministic failure on this exact input. Retrying buys nothing, doubles the latency before you fail open, and spends a second billable invocation on a request that will fail identically.

POST is safe to retry here because the endpoint is stateless and idempotent by contract (ADAPTER_API_DESIGN.md invariant 1).

Metadata

result.metadata() is a flat, metadata-only dict built for a framework's metadata blob:

{"needlepath.applied": False, "needlepath.reason": "engine_fallback",
 "needlepath.tokens_before": 8400, "needlepath.tokens_saved": 0,
 "needlepath.gate_reason": "standdown:flat_gap", ...}

Nothing derived from record text is in it. One consequence is non-obvious: selection_error is built server-side as f"{type(exc).__name__}: {exc}" and can carry fragments of your input in an exception message — the Lambda deliberately declines to log it for that reason — so only its type appears here. The full string stays on result.response.selection_error.

System One models (TypeSafe Jev, Laya)

Small classifier models that take {state, questions} and answer the questions against the state. needlepath.integrations.system_one reshapes a selection into that request body:

from needlepath.integrations.system_one import selection_receipt, to_system_one_state

result = client.select(records=records, task=task, max_context_tokens=4000)
payload = to_system_one_state(
    # Fail open with your own records: a result that did not apply is refused.
    result if result.applied else records,
    questions={"refund": Noul(instructions="Is this a refund request?")},
)
# Sidecar metadata, not part of the body.
receipt = selection_receipt(result)

payload["state"] is a JSON array of {record_id, title, source, text}, one entry per record in response.selected, in the order the service returned them, with the text unmodified. The adapter is transport only: no threshold on score, no drop for an empty excerpt, no re-sort, no deduplication, and no question you did not pass. mode="rendered" returns state as a single string instead, records joined by blank lines under a one-line [record_id] title header, for a classifier whose state argument is text.

questions travels in the shape you pass it. langchain-typesafe reads a mapping of id to typed question (Noul, Choice, Score); a different {state, questions} service may read a list. Both are copied and sent unchanged, so a mapping never arrives as a list of its keys. A bare string is refused rather than iterated into single-character questions.

to_system_one_state accepts a SelectionResult only when it applied. A result that did not apply is a passthrough — your own records are what has to be sent — and its response can still carry a selection the client itself declined to use, so passing one raises NeedlepathConfigError naming the reason. To reshape that response anyway, for an audit or a shadow report, pass result.response.

selection_receipt() returns request_id, selection_id, policy_version, tokens_saved and outcome, prefixed the way result.metadata() prefixes its keys. Attribution travels beside the request; record text never travels in it.

Testing your integration

from needlepath.testing import FakeTransport, response_payload, ok, error

transport = FakeTransport([error(503), ok(response_payload("np-1"))])
client = NeedlepathClient(operating_point="np-2026-08-r4", transport=transport)

needlepath.testing is public API, not a private helper. Every rule above is a rule about what happens when the service misbehaves, and you cannot test that against a healthy service.

Wire contract

INTERFACE.md in context-selection-bench is the published contract; the dataclasses here are an independent implementation of it, deliberately copied rather than imported so a research harness never lands in a customer's dependency tree. tests/test_wire_compatibility.py asserts they have not drifted.

Unknown response fields are kept on response.extra, never rejected: /v1 is additive, and usage (E2) is already scheduled to arrive that way.

The gate outcome and reason are an open enum. Nothing in this SDK branches on them and nothing you build on it should — a third outcome is a planned additive change.