Agents
Python Memory + Consequence
Install the Python 3.10+ SDK:
python -m pip install neutron-ai-sdk
The SDK handles memory and agent context. The helper below calls industry-library and Consequence endpoints that are not yet exposed as typed Python methods.
Client and authenticated REST helper
from __future__ import annotations
import json
import os
from typing import Any
from urllib.request import Request, urlopen
from neutron_ai import NeutronAIClient
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("/")
NUCLEUS_ID = "checkout-platform"
def neutron_request(token: str, path: str, body: dict[str, Any]) -> dict[str, Any]:
request = Request(
f"{API_URL}{path}",
data=json.dumps(body).encode("utf-8"),
method="POST",
headers={
"authorization": f"Bearer {token}",
"content-type": "application/json",
},
)
with urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
def scoped(body: dict[str, Any]) -> dict[str, Any]:
return {"nucleusId": NUCLEUS_ID, **body}
memory = NeutronAIClient(
base_url=API_URL,
token=required_env("NEUTRON_API_TOKEN"),
nucleus_id=NUCLEUS_ID,
)
Keep each token in a server-side secret store. Do not pass workspace, approver, or internal credentials to the agent runtime.
Install knowledge and remember reviewed history
neutron_request(
required_env("NEUTRON_WORKSPACE_API_KEY"),
"/v1/platform/industry-library/plugins/ai-engineering-delivery/install",
{"nucleusId": NUCLEUS_ID},
)
memory.remember({
"scopeId": "history:checkout-releases",
"agentId": "agent:release-manager",
"type": "experience",
"privacyClass": "tenant",
"text": (
"Release 3.7 used a 5% cohort for 20 minutes before expansion. "
"The rollout paused when database lock time exceeded 250 ms. "
"Rollback succeeded. Verify lock duration before expanding similar changes."
),
"metadata": {
"source": "approved-release-review",
"release": "3.7",
"outcome": "rolled-back",
},
})
context = memory.agent_context({
"scopeIds": [
"industry:ai-engineering-delivery",
"service:checkout-api",
"history:checkout-releases",
"policy:production-change",
],
"agentId": "agent:release-manager",
"task": "Prepare a safe release recommendation for checkout API version 3.8.",
"privacyClasses": ["tenant"],
"tokenBudget": 1_600,
"cachePolicy": {
"mode": "prefer_cache",
"keyMode": "intent_profile",
"ttlSeconds": 300,
"includeDynamicRag": True,
},
})
Pass context["promptContext"] to the chosen model only as supporting context. Current deployment evidence and the current task remain primary.
Plan and record the decision
agent_token = required_env("NEUTRON_API_TOKEN")
decision_scopes = [
"industry:ai-engineering-delivery",
"service:checkout-api",
"history:checkout-releases",
"policy:production-change",
]
run = neutron_request(agent_token, "/v1/consequence/plan", scoped({
"scopeIds": decision_scopes,
"agentId": "agent:release-manager",
"task": (
"Compare delaying release 3.8, a 1% pilot, a 5% staged rollout, "
"and immediate release while preserving compatibility and the lock budget."
),
"domain": "engineering_ops",
"objectiveHints": [
"Reduce customer impact",
"Preserve rollback capability",
"Collect evidence before expansion",
],
"constraintHints": [
"Database lock time must remain below the approved threshold",
"Expansion requires the accountable release owner",
"Rollback must remain available",
],
"idempotencyKey": "checkout-3.8-release-review-v1",
"retentionDays": 90,
"contextPolicy": {
"includeMemory": True,
"includeKnowledge": True,
"includeContextCapsules": True,
"includePastDecisions": True,
"includeReflections": True,
},
"policy": {
"mode": "deep",
"depth": 10,
"maxScenarios": 8,
"maxRuntimeMs": 8_000,
"requireApprovalAboveRisk": "medium",
"allowExecution": False,
"storeSafeArtifactsOnly": True,
"includeCounterfactuals": True,
"includeReflections": True,
},
}))
scenario_id = run.get("recommendedScenarioId")
if not scenario_id:
raise RuntimeError("No scenario satisfied the current decision boundary")
decided = neutron_request(agent_token, "/v1/consequence/decide", scoped({
"runId": run["runId"],
"scopeIds": ["service:checkout-api", "policy:production-change"],
"selectedScenarioId": scenario_id,
"decisionSummary": "Selected the bounded pilot scenario for owner review.",
"rationaleSummary": (
"The pilot preserves rollback and gathers current lock-duration evidence. "
"Expansion remains conditional on the threshold and owner review."
),
}))
decision = decided["decisionRecord"]
if decision["status"] == "draft":
neutron_request(
required_env("NEUTRON_APPROVER_TOKEN"),
"/v1/consequence/approve",
scoped({
"decisionId": decision["decisionId"],
"scopeIds": ["service:checkout-api", "policy:production-change"],
"note": "Release owner approved the pilot only; expansion needs another review.",
}),
)
The application must verify the human approver before using NEUTRON_APPROVER_TOKEN. Approval records authorization; it does not execute the release.
Observe and reflect
After the separately authorized application records trusted pilot evidence:
neutron_request(agent_token, "/v1/consequence/observe", scoped({
"decisionId": decision["decisionId"],
"scopeIds": ["service:checkout-api", "history:checkout-releases"],
"summary": "The 1% pilot completed without customer errors and was held for review.",
"observedMetrics": {
"pilot_percent": 1,
"pilot_minutes": 30,
"database_lock_p95_ms": 180,
"customer_error_rate": 0,
},
"unexpectedConsequences": [
"Cache warm-up took ten minutes longer than estimated."
],
"sourceRefs": [
"release-run:checkout-3.8-pilot",
"dashboard:database-locks-2026-08-15",
],
}))
reflected = neutron_request(
required_env("NEUTRON_INTERNAL_TOKEN"),
"/v1/consequence/reflect",
scoped({
"decisionId": decision["decisionId"],
"scopeIds": ["service:checkout-api", "history:checkout-releases"],
}),
)
lessons = reflected.get("reflectionLessons", [])
if not lessons:
raise RuntimeError("Reflection did not return a lesson")
lesson = lessons[-1]
Review the lesson's prediction delta, regret score, confidence adjustment, future policy suggestion, and memory-write policy before promoting it into durable memory.
Return to the shared lifecycle and production checklist.