Python Data Placement

python -m pip install neutron-ai-sdk

from __future__ import annotations

import json
import os
from typing import Any
from urllib.request import Request, urlopen


def required_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


API_URL = required_env("NEUTRON_API_URL").rstrip("/")


def neutron_request(
    token: str,
    method: str,
    path: str,
    body: dict[str, Any] | None = None,
) -> Any:
    payload = json.dumps(body).encode("utf-8") if body is not None else None
    request = Request(
        f"{API_URL}{path}",
        data=payload,
        method=method,
        headers={
            "authorization": f"Bearer {token}",
            "content-type": "application/json",
        },
    )
    with urlopen(request, timeout=30) as response:
        raw_body = response.read().decode("utf-8")
        return json.loads(raw_body) if raw_body else None


workspace_key = required_env("NEUTRON_WORKSPACE_API_KEY")
profiles = [
    {
        "nucleusId": "product-knowledge-global",
        "name": "Global product knowledge",
        "placement": {
            "mode": "global",
            "guarantee": "none",
            "notes": ["Approved global knowledge; no jurisdictional storage guarantee."],
        },
    },
    {
        "nucleusId": "support-operations-weur",
        "name": "Western Europe support operations",
        "placement": {
            "mode": "regional",
            "regionHint": "weur",
            "r2LocationHint": "weur",
            "guarantee": "best_effort_regional",
            "notes": ["Latency preference only; not a legal residency boundary."],
        },
    },
    {
        "nucleusId": "regulated-cases-eu",
        "name": "EU regulated case memory",
        "placement": {
            "mode": "jurisdictional",
            "jurisdiction": "eu",
            "regionalServicesRequired": True,
            "metadataBoundaryRequired": True,
            "geoKeyManagerRequired": False,
            "guarantee": "jurisdictional_storage",
            "notes": [
                "Requires the EU Durable Object jurisdiction and matching EU archive binding.",
                "Processing, logs, metadata, keys, retention, and contracts require separate verification.",
            ],
        },
    },
]

for profile in profiles:
    neutron_request(workspace_key, "POST", "/v1/nuclei", profile)

from neutron_ai import NeutronAIClient


global_memory = NeutronAIClient(
    base_url=API_URL,
    token=required_env("NEUTRON_GLOBAL_TOKEN"),
    nucleus_id="product-knowledge-global",
)
western_europe_memory = NeutronAIClient(
    base_url=API_URL,
    token=required_env("NEUTRON_WEUR_TOKEN"),
    nucleus_id="support-operations-weur",
)
eu_regulated_memory = NeutronAIClient(
    base_url=API_URL,
    token=required_env("NEUTRON_EU_TOKEN"),
    nucleus_id="regulated-cases-eu",
)


def store_global_product_knowledge(text: str) -> Any:
    return global_memory.remember({
        "scopeId": "kb:approved-products",
        "type": "knowledge",
        "privacyClass": "public",
        "text": text,
    })


def store_western_europe_support_lesson(text: str) -> Any:
    return western_europe_memory.remember({
        "scopeId": "history:support-resolutions",
        "type": "tool_lesson",
        "privacyClass": "tenant",
        "text": text,
    })


def store_eu_regulated_case(case_id: str, minimized_summary: str) -> Any:
    return eu_regulated_memory.remember({
        "scopeId": f"case:{case_id}",
        "type": "experience",
        "privacyClass": "user_private",
        "text": minimized_summary,
        "metadata": {
            "dataClass": "eu-regulated-case",
            "source": "approved-case-summary",
        },
    })

eu_case_context = eu_regulated_memory.agent_context({
    "scopeIds": [
        "case:case-4821",
        "policy:eu-case-handling",
        "kb:approved-products",
    ],
    "agentId": "agent:eu-case-support",
    "task": "Prepare the next authorized case-support step.",
    "privacyClasses": ["tenant", "user_private"],
    "tokenBudget": 1_200,
    "cachePolicy": {
        "mode": "prefer_cache",
        "keyMode": "intent_profile",
        "ttlSeconds": 120,
        "includeDynamicRag": True,
        "allowSensitive": False,
    },
})

eu_token = required_env("NEUTRON_EU_TOKEN")
placement = neutron_request(
    eu_token,
    "GET",
    "/v1/nuclei/regulated-cases-eu/placement",
)
health = neutron_request(
    eu_token,
    "GET",
    "/v1/nuclei/regulated-cases-eu/health",
)

blocking_warnings = {
    "jurisdictional_do_subnamespace_unavailable",
    "jurisdictional_archive_bucket_missing",
}
active_warnings = set(health.get("warnings", []))
if active_warnings & blocking_warnings:
    raise RuntimeError(f"EU Nucleus is not ready: {sorted(active_warnings)}")