Python SDK

Official Python SDK for the NOPE API. Supports both sync and async clients. Built with Pydantic for type safety.

Installation

pip install nope-net

Client Initialization

from nope_net import NopeClient, AsyncNopeClient

# Synchronous client
client = NopeClient(
    api_key="nope_live_...",
    timeout=30.0,        # Optional: request timeout in seconds (default: 30.0)
    base_url="https://api.nope.net",  # Optional: custom API URL
    max_retries=2,       # Optional: retries on 429 and 503 only (default: 2)
)

# Demo mode (no API key; routes evaluate, ocular, oversight_analyze and
# signpost_smart to the per-IP rate-limited /v1/try/* endpoints)
demo_client = NopeClient(demo=True)

# Async client (same methods, awaited)
async_client = AsyncNopeClient(api_key="nope_live_...")

Demo mode covers evaluate(), ocular(), oversight_analyze() and signpost_smart() without a key. Every other method raises ValueError before any request is sent. The demo endpoints keep the last 10 messages, always include crisis resources, and add metadata.try_endpoint to the response.

Context Manager (Recommended)

Use context managers for automatic connection cleanup:

# Recommended: use as context manager for automatic cleanup
with NopeClient(api_key="nope_live_...") as client:
    result = client.evaluate(
        messages=[{"role": "user", "content": "Hello"}]
    )

# Async context manager
async with AsyncNopeClient(api_key="nope_live_...") as client:
    result = await client.evaluate(
        messages=[{"role": "user", "content": "Hello"}]
    )

Methods

evaluate()

Full risk assessment across all 9 risk types with evidence-based features, chain-of-thought rationale, and crisis resources ($0.003 per call). See Evaluate Guide for response semantics.

from nope_net import NopeClient

client = NopeClient(api_key="nope_live_...")

# With messages (1 to 100, roles "user" or "assistant")
result = client.evaluate(
    messages=[
        {"role": "user", "content": "I've been feeling really down lately"},
        {"role": "assistant", "content": "I'm sorry to hear that. Can you tell me more?"},
        {"role": "user", "content": "I just feel hopeless, like nothing will get better"},
    ],
    config={
        "country": "US",            # ISO 3166-1 alpha-2 for crisis resources (default "US")
        "include_resources": True,  # default True
        "conversation_id": "conv_42",  # echoed on evaluate.alert webhook payloads
        "end_user_id": "user_7",       # echoed on webhook payloads as user_id
    },
)

print(result.speaker_severity)   # "none" | "mild" | "moderate" | "high" | "critical"
print(result.speaker_imminence)  # "not_applicable" | "chronic" | "subacute" | "urgent" | "emergency"
print(result.rationale)          # Chain-of-thought reasoning

for risk in result.risks:
    print(risk.type, risk.subject, risk.severity, risk.imminence, risk.features or [])

# Matched crisis resources, each with a one-line reason
if result.show_resources and result.resources:
    primary = result.resources.primary
    print(primary.name, primary.phone or primary.website_url, primary.why)

print(result.request_id, result.timestamp)

# With plain text (up to 50,000 characters)
text_result = client.evaluate(
    text="Patient expressed feelings of hopelessness during session.",
    config={"country": "US"},
)
print(text_result.metadata.input_format if text_result.metadata else None)  # "text_blob"

result.resources is a typed model with attribute access. Dict-style access (result.resources["primary"]["phone"]) still works as a compatibility shim for code written against 3.x.

screen() Deprecated

Deprecated: Use evaluate() instead, which provides the full structured v1 assessment at $0.003/call. The screen() method calls the legacy /v0/screen endpoint.

# DEPRECATED: Use client.evaluate() instead.
# screen() calls the legacy /v0/screen endpoint, emits a DeprecationWarning,
# and is not available in demo mode.
result = client.screen(
    messages=[{"role": "user", "content": "I don't want to be here anymore"}],
    config={"country": "US"},
)

if result.show_resources and result.resources:
    print(f"Crisis detected: {result.rationale}")
    print(f"Primary resource: {result.resources.primary.name}")
    print(f"Call: {result.resources.primary.phone}")

# The legacy risks array keeps "unknown" as a possible subject
for risk in result.risks:
    print(f"{risk.type}: {risk.severity} (subject: {risk.subject})")

ocular()

Behavioral risk assessment ($0.0001 per call): a continuous salience score plus eight user-risk axes and four AI-behavior axes. Set per_turn=True to receive the per-turn trajectory and its trajectory_shape. See the Ocular reference.

from nope_net import NopeClient

client = NopeClient(api_key="nope_live_...")

result = client.ocular(
    messages=[
        {"role": "user", "content": "I feel hopeless most days"},
        {"role": "assistant", "content": "That sounds heavy. What's been going on?"},
        {"role": "user", "content": "I keep thinking everyone would be better off without me"},
    ],
    per_turn=True,           # also return trajectory and trajectory_shape
    session_id="session_9",  # opaque id for dashboard analytics (1 to 256 chars)
)

# salience is the continuous score in [0, 1]; reference cutoffs are 0.30 (watch) and 0.60 (danger)
print(result.salience, result.subject, result.imminence.level)

# 8 user-risk axes under signals.user, 4 AI-behavior axes under signals.ai
suicide = result.signals.user.get("suicide")
if suicide and suicide.score > 0.5:
    print("escalate")
print(result.signals.ai["manipulation"].level)

# Per-turn trail (only with per_turn=True)
for entry in result.trajectory or []:
    print(entry.turn, entry.role, entry.salience, entry.signals_by_axis)
if result.trajectory_shape:
    print(result.trajectory_shape.phases, result.trajectory_shape.peak_turn)
print(result.meta.version, result.meta.inference_ms)

With per_turn=True, the default stride is 3 and sampling runs backward from the final message. Set trajectory_stride=1 to request every message. Each returned turn is one of the 0-based positions in the messages array. In trajectory_shape, onsets values use those message positions, while phases, slopes, and peak_turn use an index into the returned trajectory array. The shape can be absent when only one entry was scored.

Per-entry signals_by_axis uses user-axis names directly, AI keys such as ai_manipulation, and the fiction and genuine scalars. Axis level values are independently calibrated from overall salience, so do not apply the salience thresholds to individual axis scores.

oversight_analyze()

Analyze AI conversations against 91 behavior codes across 14 categories. This requires an account with Oversight enabled and costs $0.10 per call. During beta, ingest stores submitted conversations and results for product analysis and service improvement. Analyze retains operational and analysis-event metadata without writing conversation content or full results to the Oversight database. See Oversight Guide and AI Behavior Taxonomy.

The full result can also carry conversation_summary, human_indicators, filter_applied, windows, concern_progression, peak_concern, final_concern, inflection_points, context_for_next_window, narrative_summary, prompt_tokens, completion_tokens, raw_xml, model_used, and latency_ms. Availability varies by mode, strategy, and request options. See the Oversight guide's response matrix before treating an optional field as present.

from nope_net import NopeClient

client = NopeClient(api_key="nope_live_...")

result = client.oversight_analyze(
    {
        "conversation_id": "conv_123",
        "messages": [
            {"role": "user", "content": "I've been feeling really lonely lately"},
            {"role": "assistant", "content": "I understand. I'm always here for you."},
            {"role": "user", "content": "Sometimes I feel like no one cares about me"},
            {"role": "assistant", "content": "That's not true. I care about you more than anyone ever could."},
        ],
        "metadata": {"user_is_minor": False, "platform": "my-app"},
    },
    bot_context="general-purpose assistant for a productivity app",
    config={"mode": "fast"},  # "full" (default) or "fast"
    behaviors={
        "min_severity": "medium",
        "categories": ["boundary_violations", "relationship_harm"],
    },
)

print(result.strategy, result.strategy_reason)  # "single" | "sliding"

analysis = result.result
print(analysis.overall_concern)                  # "none" | "low" | "medium" | "high" | "critical"
print(analysis.trajectory, analysis.mode_used)   # fast mode: always "stable", "fast"

for behavior in analysis.detected_behaviors:
    print(f"{behavior.code} ({behavior.severity} x{behavior.turn_count}): {behavior.recommendation}")

# Full mode adds summary, pattern_assessment and per-turn evidence
for turn in analysis.turn_analysis:
    print(turn.turn_number, turn.content_summary)  # turn numbers count assistant turns from 1

config.mode selects the depth: full (default) returns summary, pattern_assessment and per-turn evidence; fast uses a quicker model, reports trajectory as stable, and returns turn_analysis empty. Do not use fast-mode conversation_summary as a turn-count source. Use detected_behaviors and overall_concern for routing. behaviors filters the result after analysis: enabled and disabled are mutually exclusive, and codes and categories come from the exported OVERSIGHT_BEHAVIOR_CODES and OVERSIGHT_BEHAVIOR_CATEGORIES tuples.

On a demo client the same call returns a different model:

from nope_net import NopeClient

# Demo mode returns OversightDemoAnalyzeResponse: mode, result, try_endpoint
demo = NopeClient(demo=True)

result = demo.oversight_analyze(
    {
        "messages": [
            {"role": "user", "content": "I feel so alone"},
            {"role": "assistant", "content": "I understand you in ways others cannot."},
        ]
    },
    config={"mode": "fast"},
)

print(result.mode, result.try_endpoint)  # "fast", True
print(result.result.overall_concern)

oversight_ingest()

Batch analysis with dashboard storage. Takes up to 300 conversations, bills $0.10 each, and returns once every conversation is analyzed.

# Analyze and store up to 300 conversations for the dashboard
batch = client.oversight_ingest(
    conversations=[
        {
            "conversation_id": "conv_001",
            "messages": [
                {"role": "user", "content": "hi"},
                {"role": "assistant", "content": "hello"},
            ],
        },
    ],
    webhook_url="https://api.example.com/webhooks/nope",  # oversight.ingestion.complete
)

print(batch.status, f"{batch.conversations_processed}/{batch.conversations_received}")
print(batch.dashboard_url)
for item in batch.results or []:
    print(item.conversation_id, item.overall_concern, item.behaviors_detected)

signpost(), signpost_smart(), signpost_search(), signpost_by_id(), signpost_countries(), detect_country()

Crisis resource lookup, ranked recommendations, semantic search, and country detection. Filters for signpost() can be passed as keyword arguments or under config=. See Signpost Guide and Service Taxonomy.

from nope_net import NopeClient

client = NopeClient(api_key="nope_live_...")

# Crisis resources by country (free, key required)
basic = client.signpost(
    "GB",
    scopes=["suicide"],  # values from nope_net.SERVICE_SCOPES; the API returns 400 for unknown scopes
    urgent=True,         # prioritize stronger current availability
)
print(f"Found {basic.count} resources")
for resource in basic.resources:
    print(resource.type, resource.name, resource.phone or resource.website_url)

# Ranked for a described situation ($0.001 per call, up to 5 picks)
ranked = client.signpost_smart("US", "teen struggling with an eating disorder")
for pick in ranked.ranked:
    print(f"{pick.rank}. {pick.resource.name}: {pick.why}")

# Semantic search across the whole directory (free, key required)
hits = client.signpost_search(query="lgbtq youth support", country="GB", limit=5)
for hit in hits.results:
    print(hit.id, hit.name, f"{hit.similarity:.2f}")

# Public routes (no key needed)
one = client.signpost_by_id(hits.results[0].id)
print(one.resource.name)

countries = client.signpost_countries()
print(f"Supported: {', '.join(countries.countries)}")

# Country detection reads geo headers a proxy injects; pass country_hint to send x-country yourself
detected = client.detect_country(country_hint="GB")
print(detected.country_code if detected.detected else "unknown")

When scopes are supplied, a Signpost response groups matching resources into primary and additional general crisis resources into secondary. The compatibility resources field repeats the primary list, and scopes_requested echoes the request. Search results use a different shape with contacts and plural service_scopes.

The resources(), resources_smart(), resource_by_id() and resources_countries() methods still call the deprecated /v1/resources/* routes, emit a DeprecationWarning, and stop working on 2027-01-01.

Async Client

AsyncNopeClient has every method of NopeClient with the same arguments and return types, awaited, including client.webhooks.* and client.billing.*:

import asyncio
from nope_net import AsyncNopeClient

async def analyze_message():
    async with AsyncNopeClient(api_key="nope_live_...") as client:
        result = await client.evaluate(
            messages=[{"role": "user", "content": "I feel hopeless"}],
            config={"country": "US"},
        )
        balance = await client.billing.balance()
        return result, balance

result, balance = asyncio.run(analyze_message())
print(result.speaker_severity, balance.balance_formatted)

Error Handling

All errors extend the base NopeError class with status_code, code (the API's machine string when the body carries one), message and response_body attributes.

Client-side validation and demo-mode refusals raise NopeValidationError, which also inherits from ValueError, before any request is sent. status_code is None, details is empty, and code is invalid_request or not_available_in_demo. API validation responses also use NopeValidationError. Branch on the exception class or status_code. code is usually None on 400, 401, 404, and 413 responses.

from nope_net import (
    NopeClient,
    NopeAuthError,
    NopeValidationError,
    NopeInsufficientBalanceError,
    NopeFeatureError,
    NopeNotFoundError,
    NopeRateLimitError,
    NopeServiceUnavailableError,
    NopeServerError,
    NopeConnectionError,
)

client = NopeClient(api_key="nope_live_...", max_retries=2)

try:
    result = client.evaluate(
        messages=[{"role": "user", "content": "Hello"}]
    )
    print(result.speaker_severity)

    # Rate-limit and balance headers from the most recent response
    meta = client.last_response_meta
    if meta and meta.rate_limit:
        print(meta.rate_limit.remaining)
    if meta and meta.balance:
        print(meta.balance.cost_mills)
except NopeAuthError as e:
    # 401: Invalid or missing API key
    print(f"Auth failed: {e}")
except NopeValidationError as e:
    # 400 or 413: invalid request; body extras such as max_messages arrive in details
    print(f"Validation error: {e.message} {e.details}")
except NopeInsufficientBalanceError as e:
    # 402: top up before retrying
    print(f"Balance {e.formatted_current}, need {e.formatted_required}: {e.topup_url}")
except NopeFeatureError as e:
    # 403: feature not enabled for this account, or a paid plan is required
    print(f"{e.feature}: {e.required_access or e.upgrade_url}")
except NopeNotFoundError as e:
    # 404
    print(f"Not found: {e.message}")
except NopeRateLimitError as e:
    # 429 after the automatic retries; retry_after is in seconds
    print(f"Rate limited, retry after: {e.retry_after} seconds")
except NopeServiceUnavailableError as e:
    # 503 after the automatic retries; retry_after is in seconds
    print(f"Temporarily unavailable, retry after: {e.retry_after} seconds")
except NopeServerError as e:
    # Other 5xx
    print(f"Server error: {e.status_code}")
except NopeConnectionError as e:
    # Network error (timeout, DNS); never retried automatically
    print(f"Connection failed: {e}")

Retries and response metadata

The client retries 429 and 503 responses up to max_retries times (default 2), waiting the Retry-After seconds the API sends and capping each wait at 30 seconds. Timeouts, connection failures and other 5xx responses are never retried: paid routes charge before the handler runs, so a blind retry could bill twice.

client.last_response_meta holds rate_limit (limit, remaining, reset as epoch milliseconds) from every response and balance (balance_mills, cost_mills) from paid routes. Absent headers give None.

Webhook Verification

Verify webhook signatures to ensure requests are from NOPE. Four events exist: evaluate.alert, oversight.alert, oversight.ingestion.complete and test.ping. See Webhooks Guide for setup.

import os

from nope_net import (
    EvaluateAlertPayload,
    OversightAlertPayload,
    OversightIngestionCompletePayload,
    TestPingPayload,
    Webhook,
    WebhookSignatureError,
)

# In your webhook handler (Flask, FastAPI, etc.). Pass the raw request body
# (bytes or str): the signature covers the exact bytes sent.
@app.post("/webhooks/nope")
def handle_webhook(request):
    try:
        verified = Webhook.verify_request(
            request.get_data(),
            request.headers,
            os.environ["NOPE_WEBHOOK_SECRET"],
        )
    except WebhookSignatureError:
        return "Invalid signature", 401

    # verified.payload is one of four models; branch with isinstance
    event = verified.payload
    if isinstance(event, EvaluateAlertPayload):
        print(verified.delivery_id, event.conversation_id, event.risk_summary.overall_severity)
    elif isinstance(event, OversightAlertPayload):
        print(event.conversation_id, event.concern, [b.code for b in event.behaviors])
    elif isinstance(event, OversightIngestionCompletePayload):
        print(event.ingestion_id, event.conversations_processed, event.concerns.high)
    elif isinstance(event, TestPingPayload):
        print(event.message)

    return "OK", 200


# Lower-level form: pass the two header values yourself
payload = Webhook.verify(
    raw_body,
    request.headers.get("x-nope-signature"),
    request.headers.get("x-nope-timestamp"),
    os.environ["NOPE_WEBHOOK_SECRET"],
    max_age_seconds=300,  # default 300; 0 disables the timestamp check
)
print(payload.event)

Webhook.verify_request() reads the headers case-insensitively from any framework's header mapping and returns a VerifiedWebhook with payload, event, delivery_id (the delivery id, for deduplicating retries) and webhook_id. event_id is a deprecated alias of delivery_id. The payload's event identifier remains payload.event_id. Webhook.verify() takes the two header values directly; both are static methods.

Managing webhooks

client.webhooks wraps /v1/webhooks (key required; creating an endpoint needs a paid plan, which surfaces as NopeFeatureError with upgrade_url).

hook = client.webhooks.create(
    "https://api.example.com/webhooks/nope",
    min_risk_level="high",       # "none" | "low" | "medium" | "high" | "critical"
    include_conversation=False,
)
print(hook.id, hook.secret)  # the secret is returned once; store it

ping = client.webhooks.test(hook.id)
print(ping.success, ping.http_status, ping.duration_ms)  # a failed delivery returns success=False

for existing in client.webhooks.list().webhooks:
    print(existing.id, existing.url, existing.enabled)
events = client.webhooks.events(hook.id, limit=10)
print(len(events.events))

client.webhooks.update(hook.id, {"enabled": False})
rotated = client.webhooks.regenerate_secret(hook.id)
print(rotated.secret)
client.webhooks.delete(hook.id)

Billing

client.billing reads balance, usage and pricing. pricing() needs no key; the other calls need one and are refused in demo mode.

# Amounts are in mills (1 mill = $0.001)
balance = client.billing.balance()
print(balance.balance_formatted, balance.low_balance, balance.estimated_evaluates)

usage = client.billing.usage(start_date="2026-09-01")
for line in usage.breakdown:
    print(line.endpoint, line.calls, line.cost_formatted)

history = client.billing.usage_history(limit=20, endpoint="/v1/evaluate")
print(history.total, history.records[0].created_at if history.records else None)

pricing = client.billing.pricing()  # public, no key needed
print(pricing.pricing["evaluate"].cost_display, pricing.free_credit_display)

checkout = client.billing.topup(10000, success_url="https://example.com/billing/ok")
print(checkout.checkout_url)  # Stripe Checkout URL

Types

All response types are Pydantic models with full validation:

from nope_net import (
    # Response types
    EvaluateResponse,
    ScreenResponse,
    OcularResponse,
    OversightAnalyzeResponse,
    OversightDemoAnalyzeResponse,
    SignpostResponse,
    SignpostSmartResponse,
    BillingBalanceResponse,

    # Core types
    Risk,
    CrisisResource,
    EvaluateResources,

    # Oversight types
    OversightAnalysisResult,
    DetectedBehavior,
    AggregatedBehavior,
    OVERSIGHT_BEHAVIOR_CODES,
    OVERSIGHT_BEHAVIOR_CATEGORIES,

    # Signpost vocabularies
    SERVICE_SCOPES,
    POPULATIONS,

    # Webhook types
    WebhookPayload,
    EvaluateAlertPayload,
    VerifiedWebhook,
)

# Response types are Pydantic models. Unknown fields are kept (extra="allow"),
# so an additive API change never breaks parsing.

For type semantics (severity levels, risk types, etc.), see:

• User Risk Taxonomy: 9 risk types, severity/imminence scales
• AI Behavior Taxonomy: 91 Oversight behaviors
• Service Taxonomy: resource scopes and populations

Utility Functions

Helper functions for working with risk assessments:

from nope_net import (
    calculate_speaker_severity,
    calculate_speaker_imminence,
    has_third_party_risk,
    SEVERITY_SCORES,
    IMMINENCE_SCORES,
)

# Highest severity among risks with subject "self" (equals result.speaker_severity)
severity = calculate_speaker_severity(result.risks)
imminence = calculate_speaker_imminence(result.risks)

# Whether any risk has subject "other"
has_third_party = has_third_party_risk(result.risks)

# Severity/imminence as numeric scores for comparison
print(SEVERITY_SCORES["critical"])   # 4
print(IMMINENCE_SCORES["emergency"]) # 4

See Also

• Node.js SDK
• API Reference
• Quickstart Guide