Appearance

intutic_clawde ​

Classes ​

ClawdeBlockedError ​

python
class ClawdeBlockedError(
    verdict: str,
    code: str,
    status: int,
    message: str,
    rule_id: Optional[str] = None,
    retry_after_seconds: Optional[int] = None,
)

Bases: ClawdeVerdictError

The proxy refused the request: a policy block, a reask, a loop run held for review, a spend cap, a DLP block, or a tool call the model made that the proxy withheld. Never retried — the same request would be refused again. verdict is kill, reask or hold, code the proxy's refusal code, rule_id the rule that decided when the proxy names one (every refusal it answers with a 200 does), and the message its reason.

ClawdeClient ​

python
class ClawdeClient(
    api_key: str,
    base_url: Optional[str] = None,
    control_plane_url: Optional[str] = None,
    provider: Optional[str] = None,
    auto_context: bool = True,
    timeout: float = 30.0,
    retries: int = 2,
)

The data-plane client: chat() sends an OpenAI-format request through the Intutic proxy, which governs it and forwards it to the provider, and raises ClawdeBlockedError when the proxy refuses it. check_budget() and circuit_breaker() add a workspace budget check, and on() registers callbacks for refusal verdicts. For the operator APIs, use ControlPlaneClient.

base_url is the proxy's address: INTUTIC_BASE_URL, then http://localhost:4000. control_plane_url is used by check_budget() and session registration: INTUTIC_CONTROL_PLANE_URL, then Intutic's hosted control plane. auto_context=False turns off context resolution and session registration. timeout is how long one attempt may take, in seconds, and retries how many times a transport failure, a timeout or a 5xx answer is retried.

Attributes ​

NameTypeDefaultDescription
listenersDict[str, List[Callable[[Dict[str, Any]], None]]]—

Methods ​

ClawdeClient.chat() ​
python
def chat(model: str, messages: List[Dict[str, str]], **kwargs: Any) -> Dict[str, Any]

Send a chat request through the proxy's /v1/chat/completions route.

Returns the completion with verdict set to "allow" when the proxy let the request through, and upstream set to {"attempts", "fallback_from"} when the proxy's retry layer made more than one upstream call for it. A governance refusal, including one the proxy answers with a 200 and names in x-intutic-refusal (or, on a stream, in its : intutic-refusal line), fires the matching event and raises ClawdeBlockedError, unretried. Transport failures, timeouts and 5xx answers are retried, except a 5xx the proxy already retried upstream; anything else raises ClawdeConnectionError.

ClawdeClient.check_budget() ​
python
def check_budget(model: str, estimated_tokens: int) -> Dict[str, Any]
ClawdeClient.circuit_breaker() ​
python
def circuit_breaker(
    tool_name: str,
    max_cost_usd: Optional[float] = None,
    fail_open: bool = False,
    require_budget: bool = False,
) -> Callable[[Callable[[], Any]], Any]

require_budget=True runs check_budget() first and refuses to run the function when the workspace has no budget left. max_cost_usd is the deprecated spelling of the same switch: the amount is not compared with anything, because nothing reports a call's cost before it is made.

ClawdeClient.emit() ​
python
def emit(event: str, payload: Dict[str, Any]) -> None
ClawdeClient.off() ​
python
def off(event: str, callback: Callable[[Dict[str, Any]], None]) -> None
ClawdeClient.on() ​
python
def on(event: str, callback: Callable[[Dict[str, Any]], None]) -> None
ClawdeClient.resolve_context() ​
python
def resolve_context() -> Dict[str, Any]

ClawdeConnectionError ​

python
class ClawdeConnectionError(ClawdeError)

Raised when the proxy is unreachable, times out, or answers with an error that is not a governance refusal.

ClawdeError ​

python
class ClawdeError(Exception)

Base exception for intutic-clawde SDK.

ClawdeVerdictError ​

python
class ClawdeVerdictError(verdict: str, message: str)

Bases: ClawdeError

A governance verdict stopped the call: a proxy refusal (ClawdeBlockedError) or the circuit breaker's budget check.

ControlPlaneClient ​

python
class ControlPlaneClient(api_key: str, base_url: Optional[str] = None)

The control-plane management client: the operator APIs the intutic CLI exposes, as methods with the same names as the TypeScript SDK's in snake_case. Each method's docstring names the route it calls.

api_key is an Intutic virtual key (vk_...) or a login JWT. base_url is the control plane's address: INTUTIC_CONTROL_PLANE_URL, then Intutic's hosted control plane. A call that fails raises ClawdeConnectionError with the method, path, status and the control plane's answer.

Methods ​

ControlPlaneClient.add_policy_source() ​
python
def add_policy_source(source: PolicySourceInput) -> Dict[str, Any]

POST /api/v1/connectors

ControlPlaneClient.adjudicate_finding() ​
python
def adjudicate_finding(
    finding_id: str,
    outcome: str,
    note: Optional[str] = None,
) -> FindingAdjudication

POST /api/v1/findings/:findingId/adjudicate -- outcome is "TRUE_POSITIVE" or "FALSE_POSITIVE"; the ruling is recorded under the key's member.

ControlPlaneClient.approve_decision() ​
python
def approve_decision(hold_id: str, reason: Optional[str] = None) -> DecisionReviewResult

POST /api/v1/decisions/:holdId/review (OWNER, ADMIN, EM) -- approve a held call.

ControlPlaneClient.approve_guardrail_shadow() ​
python
def approve_guardrail_shadow(guardrail_id: str) -> GuardrailTransitionResult

POST .../guardrails/:guardrailId/approve-shadow -- into SHADOW: distributed as warn, measuring, enforcing nothing.

ControlPlaneClient.approve_loop_run() ​
python
def approve_loop_run(
    loop_run_id: str,
    note: Optional[str] = None,
) -> LoopRunActionResult

POST /api/v1/loops/:loopRunId/review (OWNER, ADMIN, EM) -- approve a run held for review; it is ACTIVE again.

ControlPlaneClient.approve_mcp_server() ​
python
def approve_mcp_server(server_id: str) -> McpServerDecision

POST /api/v1/mcp/servers/:serverId/status -- approve a server (OWNER, ADMIN).

ControlPlaneClient.assign_org_gateway() ​
python
def assign_org_gateway(org_id: str, gateway_id: Optional[str]) -> Dict[str, Any]

PATCH /api/v1/orgs/:orgId/gateway -- pass None to clear the org default.

ControlPlaneClient.assign_workspace_gateway() ​
python
def assign_workspace_gateway(gateway_id: Optional[str]) -> Dict[str, Any]

PATCH /api/v1/workspace/gateway -- pass None to clear the override.

ControlPlaneClient.attenuate_key() ​
python
def attenuate_key(
    parent_key_id: str,
    requested_caps: List[str],
    ttl_seconds: Optional[int] = None,
) -> AttenuationResult

POST /api/v1/attenuate -- a child of one of your keys with narrower scopes: each of requested_caps is an API key scope (gateway:use, traces:read, ...) the parent grants (ttl_seconds 60 to 86400). The child key is in the response once and never stored.

ControlPlaneClient.block_mcp_server() ​
python
def block_mcp_server(server_id: str) -> McpServerDecision

POST /api/v1/mcp/servers/:serverId/status -- block a server (OWNER, ADMIN).

ControlPlaneClient.check_domain_verification() ​
python
def check_domain_verification(verification_id: str) -> Dict[str, Any]

GET /api/v1/domain-verification/:id -- re-checks DNS for the TXT record. Safe to call repeatedly; performs a fresh lookup every call.

ControlPlaneClient.collect_evidence() ​
python
def collect_evidence(
    period_start: Optional[str] = None,
    period_end: Optional[str] = None,
) -> EvidenceCollectResult

POST /api/v1/compliance/soc2-collect (OWNER, ADMIN) -- run a fresh evidence collection and seal it. The period bounds are ISO 8601.

ControlPlaneClient.complete_loop_run() ​
python
def complete_loop_run(
    loop_run_id: str,
    outcome: Optional[str] = None,
) -> LoopRunActionResult

POST /api/v1/loops/:loopRunId/complete -- outcome ("SUCCEEDED" or "FAILED") records how the run ended.

ControlPlaneClient.create_guardrail() ​
python
def create_guardrail(guardrail: AuthoredGuardrailInput) -> AuthoredGuardrailWriteResult

POST /api/v1/policy-guardrails/guardrails (OWNER, ADMIN) -- an authored guardrail, created PROPOSED. An IR a check refuses is a 400 naming the check.

ControlPlaneClient.create_notification_rule() ​
python
def create_notification_rule(rule: NotificationRuleInput) -> NotificationRule

POST /api/v1/notifications/rules. A webhook rule comes back with its signingSecret, which is not shown again.

ControlPlaneClient.create_org() ​
python
def create_org(
    org_name: str,
    domain: str,
    verification_id: str,
    region: Optional[str] = None,
) -> Dict[str, Any]

POST /api/v1/orgs -- creates a real org from an already-authenticated caller. Requires a 'verified', unconsumed domain verification (see start_domain_verification); the verification is consumed atomically inside the org-insert transaction, so it backs exactly this one org. region picks the managed gateway cell's placement ('us', 'eu', ...); omitted, the deployment's home region applies. Server-validated.

ControlPlaneClient.create_org_sop() ​
python
def create_org_sop(sop: SopInput) -> Dict[str, Any]

POST /api/v1/workspace/org-sops -- a mandatory org-wide floor (OWNER, ADMIN of a workspace in the org); title, markdown_content and an optional risk_tier.

ControlPlaneClient.create_siem_destination() ​
python
def create_siem_destination(destination: SiemDestinationInput) -> SiemDestination

POST /api/v1/siem/destinations (OWNER, ADMIN). The control plane refuses a destination that points at an internal address. A webhook destination comes back with its signingSecret, which is not shown again.

ControlPlaneClient.create_sop() ​
python
def create_sop(sop: SopInput) -> SopCreateResult

POST /api/v1/sops -- one workspace SOP; markdown_content without front matter.

ControlPlaneClient.create_team() ​
python
def create_team(org_id: str, name: str) -> Dict[str, Any]

POST /api/v1/orgs/:orgId/teams

ControlPlaneClient.create_workspace() ​
python
def create_workspace(team_id: str, name: str) -> Dict[str, Any]

POST /api/v1/teams/:teamId/workspaces

ControlPlaneClient.delete_guardrail() ​
python
def delete_guardrail(guardrail_id: str) -> GuardrailTransitionResult

DELETE /api/v1/policy-guardrails/guardrails/:guardrailId (OWNER, ADMIN) -- retires it; its history is kept.

ControlPlaneClient.delete_notification_rule() ​
python
def delete_notification_rule(rule_id: str) -> None

DELETE /api/v1/notifications/rules/:ruleId

ControlPlaneClient.delete_org_sop() ​
python
def delete_org_sop(org_sop_id: str) -> None

DELETE /api/v1/workspace/org-sops/:orgSopId -- answers 404 both for no such SOP and for a member without admin access.

ControlPlaneClient.delete_siem_destination() ​
python
def delete_siem_destination(destination_id: str) -> None

DELETE /api/v1/siem/destinations/:destinationId (OWNER, ADMIN) -- deactivates it; update_siem_destination(id, {"isActive": True}) turns it back on.

ControlPlaneClient.disable_mcp_tool() ​
python
def disable_mcp_tool(server_id: str, tool: str) -> McpServerDecision

POST /api/v1/mcp/servers/:serverId/tools -- switch one tool off (OWNER, ADMIN).

ControlPlaneClient.download_evidence() ​
python
def download_evidence(run_id: str) -> bytes

GET /api/v1/compliance/soc2-export/:runId (OWNER, ADMIN) -- a stored archive, as the file. Check it with verify_evidence_archive.

ControlPlaneClient.download_framework_coverage() ​
python
def download_framework_coverage(framework_id: str, format: str) -> bytes

GET /api/v1/compliance/frameworks/:frameworkId/coverage?format= -- the report as a file: format is "json", "md", "csv" or "pdf". Unsigned: the signed copies are the ones sealed in the evidence archive.

ControlPlaneClient.download_inventory_harnesses_csv() ​
python
def download_inventory_harnesses_csv(
    status: Optional[str] = None,
    harness: Optional[str] = None,
    device: Optional[str] = None,
    search: Optional[str] = None,
) -> bytes

GET /api/v1/inventory/harnesses?format=csv -- the CSV the dashboard downloads.

ControlPlaneClient.download_inventory_mcp_servers_csv() ​
python
def download_inventory_mcp_servers_csv(
    status: Optional[str] = None,
    harness: Optional[str] = None,
    device: Optional[str] = None,
    search: Optional[str] = None,
) -> bytes

GET /api/v1/inventory/mcp-servers?format=csv

ControlPlaneClient.enable_mcp_tool() ​
python
def enable_mcp_tool(server_id: str, tool: str) -> McpServerDecision

POST /api/v1/mcp/servers/:serverId/tools -- switch one tool on (OWNER, ADMIN).

ControlPlaneClient.extract_policy_document() ​
python
def extract_policy_document(doc_id: str, llm: bool = True) -> Dict[str, Any]

POST /api/v1/policy-guardrails/documents/:docId/extract -- propose guardrails from the document's clauses. llm=False runs only the front-matter lift, with no model call.

ControlPlaneClient.get_attenuation_chain() ​
python
def get_attenuation_chain(chain_id: str) -> List[Dict[str, Any]]

GET /api/v1/attenuate/chain/:chainId (OWNER, ADMIN) -- the delegation lineage.

ControlPlaneClient.get_branch_usage() ​
python
def get_branch_usage(period: str = 'monthly') -> BranchUsageResponse

GET /api/v1/usage/branches

ControlPlaneClient.get_commit_usage() ​
python
def get_commit_usage(period: str = 'monthly') -> CommitUsageResponse

GET /api/v1/usage/commits

ControlPlaneClient.get_finding_stats() ​
python
def get_finding_stats() -> FindingStats

GET /api/v1/findings/stats -- false-positive rate per detector, over adjudicated findings.

ControlPlaneClient.get_framework_coverage() ​
python
def get_framework_coverage(framework_id: str) -> FrameworkCoverage

GET /api/v1/compliance/frameworks/:frameworkId/coverage -- eu_ai_act, iso_42001, nist_ai_rmf or mitre_atlas.

ControlPlaneClient.get_gate_liveness() ​
python
def get_gate_liveness() -> GateLiveness

GET /api/v1/governance/gate-liveness (OWNER, ADMIN, EM) -- whether each installed harness's gate is reporting.

ControlPlaneClient.get_gateway_config() ​
python
def get_gateway_config(gateway_id: str) -> Dict[str, Any]

GET /api/v1/gateways/:id/config -- the flags set on the gateway (requireVk, requireProvisionedKey) and configVersion, as the gateway pulls them. A flag never set is absent: the gateway runs its deployment's own value.

ControlPlaneClient.get_gateway_status() ​
python
def get_gateway_status(gateway_id: str) -> GatewayStatus

GET /api/v1/gateways/:id/status

ControlPlaneClient.get_github_webhook() ​
python
def get_github_webhook() -> GitHubWebhookInfo

GET /api/v1/integrations/github/webhook (OWNER, ADMIN) -- the payload URL; never the secret.

ControlPlaneClient.get_guardrail() ​
python
def get_guardrail(guardrail_id: str) -> Dict[str, Any]

GET /api/v1/policy-guardrails/guardrails/:guardrailId

ControlPlaneClient.get_guardrail_coverage() ​
python
def get_guardrail_coverage(token: str) -> Dict[str, Any]

GET /api/v1/policy-guardrails/coverage?token= -- the passages that mention a tool or action, and the guardrails on them.

ControlPlaneClient.get_guardrail_impact() ​
python
def get_guardrail_impact(
    doc_id: Optional[str] = None,
    passage_id: Optional[str] = None,
) -> Dict[str, Any]

GET /api/v1/policy-guardrails/impact -- what a change to a document or a passage reaches. Name one of them.

ControlPlaneClient.get_guardrail_readiness() ​
python
def get_guardrail_readiness(guardrail_id: str) -> Dict[str, Any]

GET /api/v1/policy-guardrails/guardrails/:guardrailId/readiness -- whether a SHADOW guardrail's evidence clears promotion.

ControlPlaneClient.get_incident() ​
python
def get_incident(incident_id: str) -> Incident

GET /api/v1/incidents/:incidentId

ControlPlaneClient.get_integrity_chain() ​
python
def get_integrity_chain() -> IntegrityChain

GET /api/v1/integrity/chain -- the root chain walk; intact: False lists the breaks (the control plane answers a break with a 409, which is still the walk).

ControlPlaneClient.get_integrity_config_chain() ​
python
def get_integrity_config_chain() -> ConfigChain

GET /api/v1/integrity/config-chain -- the harness config snapshot chain walk.

ControlPlaneClient.get_integrity_root() ​
python
def get_integrity_root(root_id: str) -> IntegrityRootDetail

GET /api/v1/integrity/roots/:rootId -- the root, its signature and its leaves.

ControlPlaneClient.get_inventory_summary() ​
python
def get_inventory_summary() -> InventorySummary

GET /api/v1/inventory/summary

ControlPlaneClient.get_loop_run() ​
python
def get_loop_run(loop_run_id: str) -> Dict[str, Any]

GET /api/v1/loops/:loopRunId

ControlPlaneClient.get_member_usage() ​
python
def get_member_usage(period: str = 'monthly') -> MemberUsageResponse

GET /api/v1/usage/members -- period is "daily" or "monthly".

ControlPlaneClient.get_mirror_adoption_report() ​
python
def get_mirror_adoption_report(candidate_model: str) -> Dict[str, Any]

GET /api/v1/routing/mirror-adoption-report -- how a mirror-tested candidate model compared. Reported only; changes nothing.

ControlPlaneClient.get_policy_document() ​
python
def get_policy_document(doc_id: str) -> Dict[str, Any]

GET /api/v1/policy-guardrails/documents/:docId -- passages, clauses and extraction runs.

ControlPlaneClient.get_pull_request_usage() ​
python
def get_pull_request_usage(period: str = 'monthly') -> PullRequestUsageResponse

GET /api/v1/usage/pull-requests

ControlPlaneClient.get_response_echo_report() ​
python
def get_response_echo_report(
    since: Optional[str] = None,
    until: Optional[str] = None,
) -> ResponseEchoReport

GET /api/v1/findings/response-echo/report -- the bounds are ISO 8601.

ControlPlaneClient.get_rule_candidate_source() ​
python
def get_rule_candidate_source(candidate_id: str) -> RuleCandidateSource

GET /api/v1/rule-candidates/:candidateId/source -- the source of record a candidate's bundle must be compiled from. A source that does not hash to the sourceSha256 served with it is refused (ClawdeConnectionError): something between the control plane and this process changed it.

ControlPlaneClient.get_siem_destination() ​
python
def get_siem_destination(destination_id: str) -> SiemDestination

GET /api/v1/siem/destinations/:destinationId

ControlPlaneClient.get_signing_keys() ​
python
def get_signing_keys() -> SigningJwks

GET /.well-known/intutic-trace-signing.json -- the published signing keys, fetched without credentials. verify_evidence_archive takes them.

ControlPlaneClient.get_sop() ​
python
def get_sop(sop_id: str) -> Dict[str, Any]

GET /api/v1/sops/:sopId

ControlPlaneClient.get_team_usage() ​
python
def get_team_usage(period: str = 'monthly') -> TeamUsageResponse

GET /api/v1/usage/teams -- per SCIM group.

ControlPlaneClient.get_trace() ​
python
def get_trace(trace_id: str) -> Dict[str, Any]

GET /api/v1/traces/:traceId

ControlPlaneClient.get_workspace_settings() ​
python
def get_workspace_settings() -> Dict[str, Any]

GET /api/v1/workspace/settings -- every setting, resolved with its default.

ControlPlaneClient.kill_loop_run() ​
python
def kill_loop_run(loop_run_id: str) -> LoopRunActionResult

POST /api/v1/loops/:loopRunId/kill

ControlPlaneClient.list_findings() ​
python
def list_findings(
    unadjudicated: bool = False,
    detector_id: Optional[str] = None,
    limit: Optional[int] = None,
) -> List[Dict[str, Any]]

GET /api/v1/findings -- detector findings, newest first; unadjudicated=True keeps only the ones nobody has ruled on.

ControlPlaneClient.list_gateways() ​
python
def list_gateways() -> List[Dict[str, Any]]

GET /api/v1/gateways

ControlPlaneClient.list_guardrail_conflicts() ​
python
def list_guardrail_conflicts() -> List[Dict[str, Any]]

GET /api/v1/policy-guardrails/conflicts -- live guardrails and front-matter rules that contradict each other.

ControlPlaneClient.list_guardrail_duplicates() ​
python
def list_guardrail_duplicates(min_jaccard: Optional[float] = None) -> Dict[str, Any]

GET /api/v1/policy-guardrails/duplicates -- overlapping passages and rules cited more than once. min_jaccard is 0 to 1.

ControlPlaneClient.list_guardrails() ​
python
def list_guardrails(
    status: Optional[str] = None,
    target: Optional[str] = None,
    provenance: Optional[str] = None,
    doc_id: Optional[str] = None,
    limit: Optional[int] = None,
) -> List[Dict[str, Any]]

GET /api/v1/policy-guardrails/guardrails

ControlPlaneClient.list_incidents() ​
python
def list_incidents(
    status: Optional[str] = None,
    severity: Optional[str] = None,
    type: Optional[str] = None,
    page: Optional[int] = None,
    limit: Optional[int] = None,
) -> IncidentList

GET /api/v1/incidents (OWNER, ADMIN, EM) -- ranked by review priority. status is OPEN, RESOLVED or AUTO_RESOLVED; severity CRITICAL, HIGH, MEDIUM or LOW; type an anomaly category such as SCOPE_VIOLATION, or WASM_RULE_REFUSED or SYSTEM_ANOMALY, and the control plane refuses one it does not know. limit is 1 to 100 (default 20).

ControlPlaneClient.list_integrity_roots() ​
python
def list_integrity_roots(loop_run_id: Optional[str] = None) -> IntegrityRootList

GET /api/v1/integrity/roots -- the sealed trace roots.

ControlPlaneClient.list_inventory_devices() ​
python
def list_inventory_devices() -> List[Dict[str, Any]]

GET /api/v1/inventory/devices -- one row per machine.

ControlPlaneClient.list_inventory_disconnects() ​
python
def list_inventory_disconnects(limit: Optional[int] = None) -> List[DeviceDisconnect]

GET /api/v1/inventory/disconnects -- machines that ran intutic disconnect, newest first (default 100, at most 500).

ControlPlaneClient.list_inventory_harnesses() ​
python
def list_inventory_harnesses(
    status: Optional[str] = None,
    harness: Optional[str] = None,
    device: Optional[str] = None,
    search: Optional[str] = None,
) -> List[Dict[str, Any]]

GET /api/v1/inventory/harnesses -- device is a device id; search matches the harness name and the machine's hostname.

ControlPlaneClient.list_inventory_mcp_servers() ​
python
def list_inventory_mcp_servers(
    status: Optional[str] = None,
    harness: Optional[str] = None,
    device: Optional[str] = None,
    search: Optional[str] = None,
) -> List[Dict[str, Any]]

GET /api/v1/inventory/mcp-servers

ControlPlaneClient.list_inventory_skills() ​
python
def list_inventory_skills(
    device: Optional[str] = None,
    search: Optional[str] = None,
) -> List[Dict[str, Any]]

GET /api/v1/inventory/skills

ControlPlaneClient.list_loop_runs() ​
python
def list_loop_runs() -> List[Dict[str, Any]]

GET /api/v1/loops

ControlPlaneClient.list_mcp_servers() ​
python
def list_mcp_servers() -> McpServerList

GET /api/v1/mcp/servers -- every server the workspace's MCP proxies have reported, and the registry-wide policy.

ControlPlaneClient.list_notification_rules() ​
python
def list_notification_rules() -> List[NotificationRule]

GET /api/v1/notifications/rules

ControlPlaneClient.list_org_sops() ​
python
def list_org_sops() -> List[Dict[str, Any]]

GET /api/v1/workspace/org-sops -- the org-wide SOP floors of the key's org.

ControlPlaneClient.list_policy_documents() ​
python
def list_policy_documents() -> List[Dict[str, Any]]

GET /api/v1/policy-guardrails/documents

ControlPlaneClient.list_policy_sources() ​
python
def list_policy_sources() -> List[Dict[str, Any]]

GET /api/v1/connectors -- the workspace's policy sources (Notion, Confluence, GitHub, Google Drive).

ControlPlaneClient.list_provider_credentials() ​
python
def list_provider_credentials() -> List[Dict[str, Any]]

GET /api/v1/workspace/provider-credentials

ControlPlaneClient.list_provider_models() ​
python
def list_provider_models(provider: str) -> Dict[str, Any]

GET /api/v1/workspace/provider-credentials/:provider/models -- the models the stored key can reach, as the provider listed them when the key was last checked: {"provider", "models", "checkedAt"}, with models and checkedAt None until a check has recorded a list.

ControlPlaneClient.list_siem_destinations() ​
python
def list_siem_destinations() -> SiemDestinationList

GET /api/v1/siem/destinations -- the destinations and the event sources they can receive.

ControlPlaneClient.list_siem_sources() ​
python
def list_siem_sources() -> SiemSources

The event sources a destination can receive; defaults are the ones an empty sourceTables means.

ControlPlaneClient.list_sops() ​
python
def list_sops(limit: Optional[int] = None) -> List[Dict[str, Any]]

GET /api/v1/sops

ControlPlaneClient.list_team_workspaces() ​
python
def list_team_workspaces(team_id: str) -> List[Dict[str, Any]]

GET /api/v1/teams/:teamId/workspaces

ControlPlaneClient.list_teams() ​
python
def list_teams(org_id: str) -> List[Dict[str, Any]]

GET /api/v1/orgs/:orgId/teams

ControlPlaneClient.list_traces() ​
python
def list_traces(
    limit: Optional[int] = None,
    offset: Optional[int] = None,
    since: Optional[str] = None,
    enforcement: Optional[str] = None,
    model: Optional[str] = None,
) -> TraceList

GET /api/v1/traces -- newest first. limit is 1 to 100 (default 20); since is ISO 8601 or a relative duration such as "24h".

ControlPlaneClient.predict_cost() ​
python
def predict_cost(params: PredictCostParams) -> CostPrediction

POST /api/v1/predict-cost -- estimate a call's cost before it runs. Size the input with inputTokenCount or inputText.

ControlPlaneClient.promote_guardrail() ​
python
def promote_guardrail(
    guardrail_id: str,
    acknowledge_no_traffic: bool = False,
) -> GuardrailTransitionResult

POST .../guardrails/:guardrailId/promote -- into ENFORCING. Refused with a 409 until the shadow evidence clears the thresholds; a guardrail that never fired needs acknowledge_no_traffic.

ControlPlaneClient.recompute_integrity_root() ​
python
def recompute_integrity_root(root_id: str) -> IntegrityRecompute

POST /api/v1/integrity/roots/:rootId/recompute -- re-derive the root from the stored traces.

ControlPlaneClient.reconfirm_guardrail() ​
python
def reconfirm_guardrail(guardrail_id: str) -> GuardrailTransitionResult

POST .../guardrails/:guardrailId/reconfirm -- re-confirm a stale citation against a live passage.

ControlPlaneClient.refresh_pull_request_usage() ​
python
def refresh_pull_request_usage() -> PullRequestRefreshResult

POST /api/v1/usage/pull-requests/refresh -- look the branches up on GitHub now (OWNER, ADMIN, EM).

ControlPlaneClient.register_gateway() ​
python
def register_gateway(name: str, deployment_target: str) -> Dict[str, Any]

POST /api/v1/gateways

ControlPlaneClient.reject_decision() ​
python
def reject_decision(hold_id: str, reason: Optional[str] = None) -> DecisionReviewResult

POST /api/v1/decisions/:holdId/review (OWNER, ADMIN, EM) -- reject a held call.

ControlPlaneClient.reject_guardrail() ​
python
def reject_guardrail(guardrail_id: str, reason: str) -> GuardrailTransitionResult

POST .../guardrails/:guardrailId/reject -- recorded with the reason.

ControlPlaneClient.reject_loop_run() ​
python
def reject_loop_run(loop_run_id: str, note: Optional[str] = None) -> LoopRunActionResult

POST /api/v1/loops/:loopRunId/review (OWNER, ADMIN, EM) -- reject a run held for review; it is KILLED.

ControlPlaneClient.replay_guardrail() ​
python
def replay_guardrail(guardrail_id: str) -> GuardrailReplay

POST .../guardrails/:guardrailId/replay -- what the guardrail would have done on captured traffic.

ControlPlaneClient.replay_wasm_rule() ​
python
def replay_wasm_rule(
    rule_id: str,
    limit: Optional[int] = None,
    since: Optional[str] = None,
) -> WasmReplayResult

POST /api/v1/wasm-rules/:ruleId/replay -- what a rule would have done on sampled traffic, without enforcing.

ControlPlaneClient.reset_mcp_server() ​
python
def reset_mcp_server(server_id: str) -> McpServerDecision

POST /api/v1/mcp/servers/:serverId/status -- return a server to the approval queue (OWNER, ADMIN).

ControlPlaneClient.resolve_gateway() ​
python
def resolve_gateway() -> Dict[str, Any]

GET /api/v1/workspace/gateway-resolution

ControlPlaneClient.retire_guardrail() ​
python
def retire_guardrail(guardrail_id: str) -> GuardrailTransitionResult

POST .../guardrails/:guardrailId/retire -- gone from every rule endpoint on the next poll.

ControlPlaneClient.revoke_gateway() ​
python
def revoke_gateway(gateway_id: str, reason: Optional[str] = None) -> None

DELETE /api/v1/gateways/:id

ControlPlaneClient.rotate_gateway_token() ​
python
def rotate_gateway_token(gateway_id: str) -> Dict[str, Any]

POST /api/v1/gateways/:id/rotate

ControlPlaneClient.rotate_github_webhook_secret() ​
python
def rotate_github_webhook_secret() -> GitHubWebhookSecret

POST /api/v1/integrations/github/webhook/secret (OWNER, ADMIN) -- make the webhook, or replace its secret. Deliveries signed with an earlier secret are refused from then on.

ControlPlaneClient.rotate_notification_rule_secret() ​
python
def rotate_notification_rule_secret(rule_id: str) -> NotificationSecretRotation

POST /api/v1/notifications/rules/:ruleId/signing-secret -- deliveries are signed with the new secret from now on.

ControlPlaneClient.rotate_siem_destination_secret() ​
python
def rotate_siem_destination_secret(destination_id: str) -> SiemSecretRotation

POST /api/v1/siem/destinations/:destinationId/signing-secret (OWNER, ADMIN)

ControlPlaneClient.search_policy_passages() ​
python
def search_policy_passages(text: str) -> Dict[str, Any]

GET /api/v1/policy-guardrails/search?q= -- full-text search over live passages, best match first.

ControlPlaneClient.set_gateway_config() ​
python
def set_gateway_config(
    gateway_id: str,
    require_vk: Optional[bool] = None,
    require_provisioned_key: Optional[bool] = None,
) -> Dict[str, Any]

PATCH /api/v1/gateways/:id/config

ControlPlaneClient.set_provider_credential() ​
python
def set_provider_credential(provider: str, fields: Dict[str, str]) -> Dict[str, Any]

PUT /api/v1/workspace/provider-credentials/:provider

ControlPlaneClient.signup_org() ​
python
def signup_org(
    email: str,
    password: str,
    name: str,
    org_name: str,
    *,
    accept_terms: bool,
) -> Dict[str, Any]

POST /api/v1/auth/signup/org -- unauthenticated, creates the calling user.

accept_terms=True is required: the caller is 18 or older and agrees to the Terms and the Privacy Policy. Without it the call fails with 400 TERMS_NOT_ACCEPTED.

Closed by default in production (INTUTIC_PUBLIC_ORG_SIGNUP, off unless a deployment has built its own anonymous domain-verification story): creating a real org auto-provisions a real managed gateway cell, so org creation now requires DNS domain-ownership proof, and an anonymous caller has no session to own a verification attempt against. Prefer start_domain_verification + check_domain_verification + create_org with an already-authenticated api_key -- the same flow the CLI's intutic org create and the dashboard's "Create Organization" modal use.

ControlPlaneClient.start_domain_verification() ​
python
def start_domain_verification(domain: str) -> Dict[str, Any]

POST /api/v1/domain-verification/start -- mints a DNS TXT-record verification token for domain. Publish a TXT record at txtRecordName with value txtRecordValue, then poll check_domain_verification until status is 'verified'.

ControlPlaneClient.start_loop_run() ​
python
def start_loop_run(
    name: str,
    budget_limit_usd: Optional[Union[float, str]] = None,
    sops: Optional[List[str]] = None,
    auto_judge: Optional[bool] = None,
) -> Dict[str, Any]

POST /api/v1/loops/start -- register a loop run. Send its id with the run's requests (INTUTIC_LOOP_RUN_ID, as intutic loop exec sets it) so the proxy files them under it.

ControlPlaneClient.sync_policy_source() ​
python
def sync_policy_source(connector_id: str) -> PolicySourceSyncResult

POST /api/v1/connectors/:connectorId/sync

ControlPlaneClient.unset_provider_credential() ​
python
def unset_provider_credential(provider: str) -> None

DELETE /api/v1/workspace/provider-credentials/:provider

ControlPlaneClient.update_guardrail() ​
python
def update_guardrail(
    guardrail_id: str,
    changes: AuthoredGuardrailInput,
) -> AuthoredGuardrailWriteResult

PUT /api/v1/policy-guardrails/guardrails/:guardrailId (OWNER, ADMIN). A changed IR creates the next version (forked), PROPOSED with no evidence; a name or description edit keeps the evidence.

ControlPlaneClient.update_notification_rule() ​
python
def update_notification_rule(
    rule_id: str,
    changes: NotificationRuleInput,
) -> NotificationRule

PUT /api/v1/notifications/rules/:ruleId -- only the fields given change. A switch to the webhook channel comes back with the new signingSecret.

ControlPlaneClient.update_siem_destination() ​
python
def update_siem_destination(
    destination_id: str,
    changes: SiemDestinationInput,
) -> SiemDestination

PUT /api/v1/siem/destinations/:destinationId (OWNER, ADMIN) -- only the fields given change.

ControlPlaneClient.update_workspace_settings() ​
python
def update_workspace_settings(settings: Dict[str, Any]) -> Dict[str, Any]

PUT /api/v1/workspace/settings -- the route intutic settings set calls.

Only the keys given change; the rest are kept (featureFlags merges one level deeper, so one flag does not switch the others off). The control plane is the authority on which keys exist and what each accepts: an unknown key or a bad value is refused with a 400 naming it, a setting the workspace's plan does not include (the group policy below Biz Org) with a 403 "Upgrade required", and a member below OWNER or ADMIN with a 403. Each refusal raises ClawdeConnectionError carrying the server's answer.

ControlPlaneClient.upload_rule_candidate_bundle() ​
python
def upload_rule_candidate_bundle(
    candidate_id: str,
    wasm: bytes,
    source_sha256: str,
    file_name: str = 'rule.wasm',
) -> BundleUploadResult

POST /api/v1/rule-candidates/:candidateId/bundle (OWNER, ADMIN) -- upload a compiled rule with the hash of the source it was compiled from. The control plane runs its gates; accepted says whether the rule went into shadow, and gates why.

ControlPlaneClient.whoami() ​
python
def whoami() -> Dict[str, Any]

GET /api/v1/auth/me

Gate ​

python
class Gate(cfg: GateConfig, client: Optional[GateClient] = None)

The pre-execution gate. guard() evaluates one tool call against the local tiers (SSO group policy, policy snapshot, MCP servers, image integrity) and, with a GateClient, the control-plane tiers (SOP rules, hook gate), and raises IntuticGateRefusal to refuse it. Without a client only the local tiers run.

Methods ​

Gate.guard() ​
python
def guard(tool_name: str, tool_input: dict) -> None

Raise IntuticGateRefusal if this call must not run. Return to allow.

Under a chaos run's plan, the call is reported to the run's events file whatever the tiers decide, and a call they allowed can still be refused by the fault that is due (CHAOS_FAULT, or IntuticGateHold for a held one). A plan that is set but unreadable refuses every call: an experiment run without the faults it asked for reports a resilience it never measured.

Gate.policy() ​
python
def policy() -> dict

Load the image allowlist.

A missing or malformed allowlist is NOT treated as "allow everything". An empty policy with require_digest still refuses every unpinned image, which is the safe direction; a policy we cannot read at all raises, and the caller turns that into a block.

Gate.snapshot() ​
python
def snapshot() -> snapshot.Snapshot
Gate.sop_rules() ​
python
def sop_rules() -> list

Active SOP rules, fetched once per process.

Deliberately not refreshed on a timer the way policy.ts does: a rule set that changes mid-run makes a verdict depend on when the fetch landed. Construct a new Gate (or clear _sop_rules) to refresh.

GateClient ​

python
class GateClient(
    base_url: Optional[str] = None,
    api_key: str = '',
    workspace_id: str = '',
    session_id: str = '',
    harness: str = 'langgraph',
    fail_closed: bool = True,
    timeout: float = 2.0,
    transport: Optional[Transport] = None,
    get_transport: Optional[GetTransport] = None,
)

Control-plane client for the pre-execution gate.

base_url is the control plane's address: INTUTIC_CONTROL_PLANE_URL, then Intutic's hosted control plane. Without an api_key the client is local_only. harness defaults to "langgraph" and is sent as harnessType on every gate call and event, so incidents attribute to the right adapter. Override it when wrapping a different framework. fail_closed blocks a call when the hook gate gives no verdict (the control plane cannot be reached, answers without one, or answers GOVERNANCE_UNAVAILABLE); False lets such a call go ahead. timeout is how long one control-plane request may take, in seconds. transport and get_transport replace the HTTP POST and GET calls, for tests.

Attributes ​

NameTypeDefaultDescription
local_onlybool—True when there is no API key. Such a client cannot authenticate, so it sends nothing: the control-plane tiers (SOP register, hook gate) are skipped, events and holds are not reported, and the local tiers decide.

Methods ​

GateClient.approved_bypasses() ​
python
def approved_bypasses() -> Optional[list]

The workspace's approved bypasses (GET /api/v1/decisions/approved-bypasses), or None when they cannot be read. Never raises: a hold that cannot be checked for an approval stays held.

GateClient.emit() ​
python
def emit(
    event: str,
    tool_name: str,
    reason: str = '',
    tool_input: Optional[dict] = None,
    incident_id: Optional[str] = None,
    file_path: Optional[str] = None,
    gate_decision_id: Optional[str] = None,
) -> bool

Fire-and-forget telemetry. Never raises.

gate_decision_id is the hook gate's decisionId for the call this event reports: the control plane then files the decision's incident once and does not export it to SIEM a second time.

A tool_blocked event creates a governance_incidents row whose description embeds our reason verbatim:

[Hook Gate] Tool "shell" blocked by local SOP enforcement. Reason: <reason> Harness: langgraph

That string is what appears on the incidents dashboard.

GateClient.from_env() ​
python
@classmethod
def from_env(
    session_id: Optional[str] = None,
    harness: str = 'langgraph',
    fail_closed: bool = True,
) -> 'GateClient'

Build a client from the environment, falling back to CLI credentials.

Reading ~/.intutic/credentials.json means the gate works with whatever workspace intutic login already bound, instead of duplicating the key. With no key in either place the client is local_only.

GateClient.hook_gate() ​
python
def hook_gate(tool_name: str, tool_input: dict) -> GateResponse

Synchronous allow/deny for one tool call.

Worth being precise about: the endpoint matches ~19 DLP regexes against the serialised toolInput, plus SOP rules. It is defence in depth on top of the local tiers, not a replacement for them.

When there is no verdict (the control plane cannot be reached, answers non-2xx or without a boolean allowed, or answers GOVERNANCE_UNAVAILABLE because its own checks could not run), the verdict is decided by fail_closed: True (the default) reports BLOCK, because a client with no verdict has nothing to defer to; False reports allow. A 400 INVALID_REQUEST is a BLOCK whatever fail_closed says. A shadow-mode workspace's {allowed: true, degraded: true} is an allow.

A local_only client makes no call and answers allow with code LOCAL_ONLY, whatever fail_closed says: with no key there is no control-plane tier to fail, only local ones, which have already run.

GateClient.judge_finalize() ​
python
def judge_finalize(
    full_content: str,
    monitored_model: str,
    personal_sops: Optional[list] = None,
) -> dict

Grade generated text against the workspace's active SOP.

POST /api/v1/judge/finalize. Note that plain /api/v1/judge does NOT exist — it is a JWT-bypass prefix in PUBLIC_ROUTES, not a handler, and returns 404.

The endpoint interpolates the active SOP's markdown verbatim into the judge's system prompt, which is what makes it usable as a general "does this text contradict this document" check. The SOP must be VALIDATED and active; only one is ever loaded (newest by updatedAt).

Returns {verdict, triggered, personalTriggered, correctionSummary, independent}. verdict is 'PASS' | 'TRIGGERED' | 'REVIEW' from the control plane, or 'UNAVAILABLE' — set by the control plane on its own failures (it returns HTTP 503, which raises in _post and lands in the except below) and by this method on transport errors. Branch on verdict, and treat only 'PASS' as clean:

  • 'TRIGGERED': a violation was found.
  • 'REVIEW': the response was held for human review. The typed judge was unsure, or did not answer in time (the judge GPU was busy). No org verdict exists yet: triggered is False because nothing was decided, not because the response is clean. A fail-closed caller must treat REVIEW as not clean. The control plane records the interim verdict and the reviewer's ruling on its side.
  • 'UNAVAILABLE' (or any other value): NO grading happened.

This cannot block. lib/monitorModel.ts is explicit: no LLM verdict blocks a request; the KILL sites in the proxy are deterministic. The caller must branch on the verdict itself.

History worth keeping: the endpoint used to fail OPEN — every internal failure (LiteLLM down, 30s timeout, unparsable verdict) returned HTTP 200 {triggered: false}, indistinguishable from a clean pass, and this client's except-branch only caught transport errors. The control plane now 503s with verdict:'UNAVAILABLE' on every failure mode, which routes them all through the except below — so a fail-closed caller finally covers what it always claimed to.

GateClient.record_hold() ​
python
def record_hold(hold: Dict[str, Any]) -> bool

Record one hold in the review queue (POST /api/v1/decisions), the way the hook gates' records reach it. Never raises; False when the control plane did not accept it, so the refusal can say there is nothing to approve yet.

GateConfig ​

python
@dataclass
class GateConfig(
    repo_root: str = '.',
    workspace_id: str = '',
    allowlist_path: str = '.intutic/image-allowlist.json',
    enforce: bool = True,
    use_hook_gate: bool = True,
    use_sop_rules: bool = True,
)

How a Gate evaluates calls. Every field has a default.

Attributes ​

NameTypeDefaultDescription
allowlist_pathstr'.intutic/image-allowlist.json'The image allowlist, relative to repo_root unless absolute.
enforceboolTrueFalse turns Gate.guard() into a no-op.
repo_rootstr'.'Where relative manifest and allowlist paths resolve from.
use_hook_gateboolTrueCall the control plane's hook gate (needs a client).
use_sop_rulesboolTrueFetch and apply the workspace's SOP rules (needs a client).
workspace_idstr''Compared with the workspace the policy snapshot was compiled for, and scopes the SOP-rule fetch.

Methods ​

GateConfig.allowlist_abs() ​
python
def allowlist_abs() -> str

GateResponse ​

python
@dataclass
class GateResponse(
    allowed: bool,
    reason: str = '',
    incident_id: Optional[str] = None,
    reached: bool = True,
    code: Optional[str] = None,
    decision_id: Optional[str] = None,
)

Attributes ​

NameTypeDefaultDescription
allowedbool—
codeOptional[str]None
decision_idOptional[str]None
incident_idOptional[str]None
reachedboolTrue
reasonstr''

GatewayStatus ​

python
class GatewayStatus(TypedDict)

GET /api/v1/gateways/:id/status; the twin of the TypeScript SDK's GatewayStatus.

status, reportedAt, appliedConfigVersion and desiredConfigVersion are always sent. The other keys are None when the gateway is unreachable, and absent when its last heartbeat did not report them.

Attributes ​

NameTypeDefaultDescription
activeWorkspacesOptional[int]—
appliedConfigVersionOptional[int]—
desiredConfigVersionint—
lastErrorOptional[str]—
litellmReachableOptional[bool]—
proxyVersionOptional[str]—
reportedAtOptional[str]—
statusLiteral['online', 'degraded', 'unreachable', 'pending']—
uptimeSecondsOptional[int]—

IntuticGateHold ​

python
class IntuticGateHold(reason: str, hold_id: Optional[str])

Bases: IntuticGateRefusal

Raised when a hold rule stopped the call to ask a person first (code HELD). A subclass, so a caller that stops on every refusal stops on this too; one that tells the user about holds catches it first.

hold_id names the hold in the review queue, and is None when the hold could not be recorded (no control plane to record it in), in which case there is nothing to approve yet. The message starts [Intutic Governance] HELD:, as the hook gates print a hold, and says who can approve it and when a retry passes. See hold.py.

IntuticGateRefusal ​

python
class IntuticGateRefusal(reason: str, code: str, incident_id: Optional[str] = None)

Bases: ClawdeError

Raised when a tool call must not run.

The python-raise contract: the exception's message begins [Intutic Governance] BLOCKED: — the same family the Open WebUI filter raises — so harnesses and log scrapers that already recognise that prefix recognise this refusal too. The structured fields (reason, code, one of GATE_REFUSAL_CODES, and incident_id) carry the machine-readable version.

ProxyRefusal ​

python
class ProxyRefusal(_ProxyRefusalBase)

Attributes ​

NameTypeDefaultDescription
rule_idstr—

Functions ​

guard() ​

python
def guard(
    func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    gate: Optional[Gate] = None,
) -> Callable

Decorator: evaluate the gate before the tool body runs.

Usage:

python
@guard
def shell(command: str) -> str: ...

@guard(name="shell")           # explicit tool name
def run_command(command: str) -> str: ...

On deny the wrapper raises IntuticGateRefusal and the body never runs; the deny has already been reported via hook-events by the gate. gate overrides the process-wide instance registered with install(); without either, calling the wrapped tool raises RuntimeError rather than silently running unguarded.

guard_tools() ​

python
def guard_tools(tools: Iterable[Any], *, gate: Optional[Gate] = None) -> list

Wrap a list of tools (LangChain/LangGraph/smolagents tool objects, or plain callables) so each is gated before execution.

Duck-typed on purpose — no langchain/smolagents import, no hard dependency:

  • an object with a callable .func (StructuredTool, @tool-created Tool) gets its .func wrapped in place;
  • otherwise an object with ._run (LangChain BaseTool subclass) gets ._run wrapped in place (set via object.setattr, so pydantic models that restrict attribute assignment still work);
  • otherwise an object with .forward (smolagents Tool subclass — Tool.__call__ dispatches to self.forward(...)) gets .forward wrapped in place, the same way. Added when the smolagents adapter (adapters/smolagents.py) was built: without this branch, a smolagents Tool instance is still callable (it has __call__), so it fell into the plain-callable branch below and got REPLACED by a bare wrapper function — silently dropping the .name/.inputs/ schema attributes ToolCallingAgent needs to expose it to the model. The .forward branch keeps the original Tool object (and its schema) intact, only swapping its execution method, matching the ._run branch's contract exactly;
  • a plain callable is returned wrapped by @guard — which is also how CrewAI and AutoGen tools are guarded, directly with the decorator.

Already-guarded entries pass through untouched, so calling this twice does not double-gate. Returns the same objects (mutated in place) in order.

intutic_headers() ​

python
def intutic_headers(
    session_id: Optional[str] = None,
    workspace_id: Optional[str] = None,
    harness: str = 'langgraph',
) -> Dict[str, str]

Default headers for a ChatOpenAI/OpenAI client pointed at the Intutic proxy.

x-session-id matters: the proxy defaults an unset one to the literal "unknown", which merges every run into a single dashboard session. x-intutic-harness attributes traces to this adapter — the proxy honours it for trace attribution (unknown headers are ignored by older proxies, so it is always safe to send).

Usage:

python
ChatOpenAI(base_url=..., default_headers=intutic_headers(session_id=run_id))

stream_refusal() ​

python
def stream_refusal(text: str) -> Optional[ProxyRefusal]

The refusal named anywhere in a stream's text, or None when there is none. Pass the whole body, or each line as it arrives.

verify_evidence_archive() ​

python
def verify_evidence_archive(
    archive: Dict[str, Any],
    jwks: Optional[SigningJwks],
) -> EvidenceVerification

Checks an evidence archive the way it was sealed: the manifest's hashes, then the signature over the archive hash against the published key its keyId names (jwks, from ControlPlaneClient.get_signing_keys() or a saved copy; None when it could not be had).

verified is True only for an archive whose hashes match and whose signature a published key accepts. An unsigned archive, or one whose key is not published, is not verified even when its hashes match: nothing then shows who produced it.

verify_integrity_root() ​

python
def verify_integrity_root(
    root: Mapping[str, Any],
    jwks: Optional[SigningJwks],
) -> IntegrityRootSignatureState

Checks a sealed root's signature against the published key the root itself names (jwks, from ControlPlaneClient.get_signing_keys() or a saved copy; None when it could not be had). Pass the root of ControlPlaneClient.get_integrity_root().

Only "invalid" says the root was changed: a published key rejected it. "unverifiable" (the key is not published, or the root names an algorithm or preimage version this build cannot check) and "keys_unavailable" are no verdict, and "unsigned" is a deployment that had no signing key. The key is chosen by signing_key_id and never by trying others: a root that verifies under some other key is not the same claim.

Constants ​

PROXY_REFUSALS ​

python
PROXY_REFUSALS: Dict[str, Tuple[int, str]] = ...

refusal code -> (HTTP status, verdict). The same table as PROXY_REFUSALS in the TypeScript SDK.

REFUSAL_HEADER ​

python
REFUSAL_HEADER = 'x-intutic-refusal'

The headers the proxy sets on a 200 that is a refusal in the shape of an answer: the response gate withheld a tool call the model made, output DLP withheld the body, or the cost-prediction gate answered instead of the model. The first names the refusal's code, the second the rule that decided.

REFUSAL_RULE_HEADER ​

python
REFUSAL_RULE_HEADER = 'x-intutic-refusal-rule'

STREAM_REFUSAL_MARKER ​

python
STREAM_REFUSAL_MARKER = ': intutic-refusal '

The prefix of the SSE comment line that names a refusal on a stream: : intutic-refusal {"code", "rule", "message"}. The response headers went out before the refusal happened, so a stream says it in band; every SSE parser skips comment lines, so clients that do not look for it are unaffected.

UPSTREAM_ATTEMPTS_HEADER ​

python
UPSTREAM_ATTEMPTS_HEADER = 'x-intutic-upstream-attempts'

Set by the proxy when its retry layer made more than one upstream call.

UPSTREAM_FALLBACK_HEADER ​

python
UPSTREAM_FALLBACK_HEADER = 'x-intutic-upstream-fallback-from'

Set by the proxy when a fallback answered: the model whose retries ran out.

The circuit breaker for AI agents

© 2026 Intutic Community. All rights reserved.