Safety API¶
The client.safety namespace exposes the safety control plane, rule registry, and checkpoint mechanism. Use it to inspect, tune, and trigger human-in-the-loop review.
SDK proxy¶
crp.sdk.proxies._SafetyProxy ¶
Safety Control Plane proxy (SPEC-033, SPEC-034).
Provides access to the safety registry, manifest, coverage map, checkpoints, and the active safety profile.
profile property ¶
Return the current safety profile name or dict.
Returns:
| Type | Description |
|---|---|
Any | Profile name (str) or settings dict. |
control_plane() ¶
Return the orchestrator's SafetyControlPlane instance.
Instantiates a new control plane using the orchestrator's security manager settings if one is not already attached.
Returns:
| Type | Description |
|---|---|
Any | SafetyControlPlane instance. |
manifest() ¶
Return the current SafetyManifest.
Returns:
| Type | Description |
|---|---|
Any | SafetyManifest instance. |
coverage_map() ¶
Return the current SafetyCoverageMap.
Returns:
| Type | Description |
|---|---|
Any | SafetyCoverageMap instance. |
add_rule(name, check_fn=None, **kwargs) ¶
Register a custom safety rule on the control plane.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Unique rule name. | required |
check_fn | Any | None | Callable invoked by the rule (optional). | None |
**kwargs | Any | Additional metadata forwarded to | {} |
Returns:
| Type | Description |
|---|---|
Any | The registered |
checkpoint(trigger='RISK_HIGH', timeout_action='ESCALATE', reject_action='HALT', timeout=300) ¶
Create a human-in-the-loop checkpoint (SPEC-033 §3).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
trigger | str | Condition that fires the checkpoint. | 'RISK_HIGH' |
timeout_action | str | Action on timeout ( | 'ESCALATE' |
reject_action | str | Action on reject ( | 'HALT' |
timeout | int | Seconds before auto-resolution. | 300 |
Returns:
| Type | Description |
|---|---|
Any | Checkpoint instance configured with the orchestrator's security |
Any | manager. |
set_profile(profile) ¶
Apply a named safety profile (balanced/strict/medical/financial).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
profile | str | One of the supported profile names. | required |
set(**kwargs) ¶
Tune one or more manifest safety settings by key.
Example::
client.safety.set(require_grounding=0.8, hallucination_halt="HIGH")
show() ¶
Return a human-readable summary of the full safety surface.
registry() ¶
Return the safety capability registry as a list of dicts.
explain(name) ¶
Explain a single safety capability or setting.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Capability or setting name. | required |
Returns:
| Type | Description |
|---|---|
str | Human-readable explanation, or a refusal if unknown. |
Safety Control Plane¶
crp.security.control_plane.SafetyControlPlane dataclass ¶
Single place from which all CRP safety is seen, tuned, and extended (SPEC-033).
Attributes:
| Name | Type | Description |
|---|---|---|
registry | dict[str, SafetyCapability] | Dict of capability-name → SafetyCapability. |
manifest | SafetyManifest | The SafetyManifest that drives settings. |
coverage | SafetyCoverageMap | The SafetyCoverageMap (capabilities + out-of-scope). |
checkpoints | SafetyCoverageMap | Active checkpoint instances awaiting resolution. |
get_capability(name) ¶
Retrieve a capability from the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Capability identifier. | required |
Returns:
| Type | Description |
|---|---|
SafetyCapability | None | The matching |
list_capabilities() ¶
Return all registered capabilities.
Returns:
| Type | Description |
|---|---|
list[SafetyCapability] | List of every built-in, addable, and custom capability in the registry. |
tune(name, value) ¶
Change the current value of a capability and sync to Manifest.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Capability identifier. | required |
value | Any | New current value. Should normally be within | required |
Returns:
| Type | Description |
|---|---|
None | None. Logs a warning if the capability is unknown. |
register_rule(rule) ¶
Register a custom safety rule as a first-class citizen.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rule | CustomSafetyRule | Custom rule definition including name, check function, and default. | required |
Returns:
| Type | Description |
|---|---|
None | None. The rule is mirrored into the registry and coverage map. |
create_checkpoint(trigger='always', timeout=300, on_timeout='escalate', on_reject='fallback', context=None) ¶
Create and track a new human-in-the-loop checkpoint.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
trigger | str | Condition that fires the checkpoint. One of the | 'always' |
timeout | int | Seconds to wait for human review before auto-resolution. | 300 |
on_timeout | str | Action taken when the checkpoint times out. | 'escalate' |
on_reject | str | Action taken when the human reviewer rejects. | 'fallback' |
context | dict[str, Any] | None | Arbitrary dict passed to the reviewer UI/webhook. | None |
Returns:
| Type | Description |
|---|---|
Checkpoint | The created |
resolve_checkpoint(checkpoint_id, resolution) ¶
Resolve a checkpoint by ID (called by reviewer or webhook).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
checkpoint_id | str | UUID of the checkpoint returned by | required |
resolution | CheckpointResolution | Human reviewer's decision and optional edited output. | required |
Returns:
| Type | Description |
|---|---|
None | None. Logs an error if the checkpoint ID is unknown. |
get_surface_map() ¶
Return the complete safety surface as a dict - for UI dashboards.
Returns:
| Type | Description |
|---|---|
dict[str, Any] | Dict with |
dict[str, Any] |
|
show() ¶
Return a human-readable printout of the entire safety surface.
Returns:
| Type | Description |
|---|---|
str | Multi-line string listing every capability's current/default values, |
str | allowed ranges, effects, and the explicit out-of-scope list. |
Checkpoint¶
crp.security.checkpoint.Checkpoint dataclass ¶
Inline human-in-the-loop declaration (SPEC-033 §3).
Attributes:
| Name | Type | Description |
|---|---|---|
checkpoint_id | str | Unique identifier for this checkpoint instance. |
trigger | CheckpointTrigger | Condition that caused the checkpoint to fire. |
timeout | int | Seconds before auto-action is taken. |
on_timeout | CheckpointTimeoutAction | Action if human does not respond in time. |
on_reject | CheckpointRejectAction | Action if human rejects. |
context | dict[str, Any] | Arbitrary context dict for the reviewer UI. |
wait_for_resolution() async ¶
Block until the checkpoint is resolved by a human reviewer.
If timeout expires, auto-resolves per on_timeout.
Returns:
| Type | Description |
|---|---|
CheckpointResolution | A |
Raises:
| Type | Description |
|---|---|
TimeoutError | Internally caught and converted to auto-resolution. |
resolve(resolution) ¶
Called by a human reviewer (or webhook) to resolve this checkpoint.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resolution | CheckpointResolution | The reviewer's decision and optional edited output. | required |
Returns:
| Type | Description |
|---|---|
None | None. Sets the internal resolved event so awaiting tasks unblock. |
decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback') classmethod ¶
Decorator factory: @checkpoint(when="risk >= HIGH").
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
when | str | Trigger condition string. | 'always' |
timeout | int | Seconds to await human review. | 300 |
on_timeout | str | Action if the checkpoint times out. | 'escalate' |
on_reject | str | Action if the reviewer rejects. | 'fallback' |
Returns:
| Type | Description |
|---|---|
Callable | A decorator that wraps the function with checkpoint gating. |
Raises:
| Type | Description |
|---|---|
CRPError | If the reviewer rejects and |