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

Ordinary unprotected failures return result.applied=False, allowing your original context. A record your workspace policy withholds comes back as result.reason == "blocked" with result.applied=True and an empty rendered_context: the verdict is "deliver nothing from these records", and that is what you install. Workspace guardrail failures (the service could not enforce, or returned a protected response that cannot be applied) raise NeedlepathGuardrailError; stop the model request instead of substituting the original context.

Workspace guardrails

The framework integrations (LangChain/LangGraph, Strands, LiteLLM and LlamaIndex) install a blocked verdict as a withheld notice in place of the record, or as no context, and let the agent continue; they propagate NeedlepathGuardrailError rather than falling back to the original. guardrails_required is a core-client option: pass a client you constructed with it to the integration.

For a workspace with guardrails, update the SDK and construct either the sync or async client with guardrails_required=True:

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

This local fallback setting is never sent to the server and cannot disable workspace policies. It stops on transport errors, timeouts, invalid requests, and missing or malformed enforcement evidence. An authoritative admin pause (enabled=False) or active configuration with zero policies is honored.

Regardless of this setting, invalid governance metadata or guardrails_unavailable raises a sanitized NeedlepathGuardrailError. A blocked response never raises: it is applied as an empty delivery (reason: "blocked"), in shadow mode too, and the receipt's governance block says what was withheld and under which policy revision. When the server forbids raw-context fallback, only its permitted selected context is applied, including in shadow mode and after engine recovery. result.declined(...) cannot restore the original context for a protected response. Do not catch a guardrail exception or a response rewrite error and continue with raw records. Existing clients must update and enable this setting to stop safely when an outage prevents the server from returning any enforcement evidence.

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
guardrails_required	—	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 for ordinary unprotected requests

With the default configuration and no workspace protection evidence, a timeout, 5xx, violated contract, empty selection, or stood-down gate returns a SelectionResult with applied=False, the exception on .error, and a machine-readable .reason. The workspace guardrail contract above takes precedence over these optimization fallbacks.

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.

An ordinary empty selection degrades to passthrough. A blocked verdict is applied as an empty delivery (reason: "blocked", applied=True), never a passthrough and never an exception. An empty selection under workspace protection that is not a blocked verdict raises and must stop the model request.

select_or_raise() exists for diagnostics and benchmarks that must not silently substitute a passthrough for a measurement. Framework adapters use select() with its explicit guardrail exception contract.

2. Pin the operating point

operating_point is required at construction; an invalid configuration raises. Failing open on an ordinary unprotected 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, with the same body, billing, and operating point. Ordinary unprotected selections are reported without being applied (reason: "shadow"). A blocked verdict is enforced in shadow mode too: shadow never restores a withheld record, so the result stays reason: "blocked", applied=True, with shadow: true, outcome and governance on the receipt. A protected delivery likewise takes precedence: only its permitted selected context may be used, or the request stops with NeedlepathGuardrailError.

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.