Control

The admission gate: decide whether a consequential action may run, and record the decision either way.

tulip.control is the domain-neutral surface. The implementations live under tulip.security for historical reasons — that is where the layer grew up — and are re-exported here, which is the import path to use.

For the concepts, start with The control layer and Writing a policy that holds.

Admitting an action

admit() evaluates the policy, records the decision on the audit trail, and runs the action only if it was allowed. A held or denied action raises AdmissionError carrying the ApprovalDecision that explains why.

admit async

admit(action: Action, perform: Callable[[], Awaitable[T]], *, policy: ControlPolicy, finding: Evidence | None = None, verdict: VerificationResult | None = None, trail: AuditTrail | None = None) -> T

Run perform only if action clears the trust chain; else reject.

The mandatory gate that turns the composable chain into an enforced one:

1. :func:~tulip.security.policy.approve weighs the action against the evidence (finding), the verification (verdict), and the policy.
2. The decision is recorded to trail (if given) — admitted or not — so no side effect is un-audited.
3. On ALLOW, perform is awaited and its result returned. On require_human or deny, :class:AdmissionError is raised with the decision attached.

Parameters:

Name	Type	Description	Default
action	Action	The proposed side-effecting action.	required
perform	Callable[[], Awaitable[T]]	A zero-arg async callable that performs the side effect.	required
policy	ControlPolicy	The governing :class:~tulip.security.policy.ControlPolicy.	required
finding	Evidence | None	The evidence the action responds to.	None
verdict	VerificationResult | None	The :func:~tulip.security.verify.verify result.	None
trail	AuditTrail | None	An :class:~tulip.security.audit.AuditTrail to record the decision on.	None

Returns:

Type	Description
T	Whatever perform returns.

Raises:

Type	Description
AdmissionError	if the action is not admitted (require_human or deny).

Source code in .sdk/src/tulip/security/admit.py

async def admit(
    action: Action,
    perform: Callable[[], Awaitable[T]],
    *,
    policy: ControlPolicy,
    finding: Evidence | None = None,
    verdict: VerificationResult | None = None,
    trail: AuditTrail | None = None,
) -> T:
    """Run ``perform`` only if ``action`` clears the trust chain; else reject.

    The mandatory gate that turns the composable chain into an enforced one:

    1. :func:`~tulip.security.policy.approve` weighs the action against the evidence
       (``finding``), the verification (``verdict``), and the ``policy``.
    2. The decision is recorded to ``trail`` (if given) — admitted or not — so no
       side effect is un-audited.
    3. On ALLOW, ``perform`` is awaited and its result returned. On require_human or
       deny, :class:`AdmissionError` is raised with the decision attached.

    Args:
        action: The proposed side-effecting action.
        perform: A zero-arg async callable that performs the side effect.
        policy: The governing :class:`~tulip.security.policy.ControlPolicy`.
        finding: The evidence the action responds to.
        verdict: The :func:`~tulip.security.verify.verify` result.
        trail: An :class:`~tulip.security.audit.AuditTrail` to record the decision on.

    Returns:
        Whatever ``perform`` returns.

    Raises:
        AdmissionError: if the action is not admitted (require_human or deny).
    """
    decision = approve(action, policy=policy, finding=finding, verdict=verdict)
    if trail is not None:
        trail.record(
            "action-admission",
            {
                "action": action.name,
                "asset": action.asset,
                "outcome": decision.outcome,
                "reason": decision.reason,
            },
        )
    if not decision.allowed:
        raise AdmissionError(decision)
    return await perform()

AdmissionError

AdmissionError(decision: ApprovalDecision)

Bases: Exception

A side-effecting action failed admission — it did not clear the trust chain.

Carries the :class:~tulip.security.policy.ApprovalDecision so the caller can route a require_human hold to an approver or surface a deny reason.

Source code in .sdk/src/tulip/security/admit.py

def __init__(self, decision: ApprovalDecision) -> None:
    self.decision = decision
    super().__init__(
        f"action {decision.action.name!r} not admitted ({decision.outcome}): {decision.reason}"
    )

Deciding

approve() is the pure decision function — no I/O, no side effects. It takes an action and a policy and returns the outcome. Rules combine by taking the strongest result, so deny beats require_human beats allow.

approve

approve(action: Action, *, policy: ControlPolicy, finding: Evidence | None = None, verdict: VerificationResult | None = None, advisor: ControlAdvisor | None = None) -> ApprovalDecision

Decide whether action may proceed: allow / require_human / deny.

Weighs every rule and returns the strongest triggered outcome (deny > require_human > allow), recording each check that fired so the decision is auditable.

Parameters:

Name	Type	Description	Default
action	Action	The proposed action.	required
policy	ControlPolicy	The governing :class:ControlPolicy.	required
finding	Evidence | None	The evidence the action responds to (optional).	None
verdict	VerificationResult | None	The :func:~tulip.security.verify.verify result (optional, but auto-allow needs one that clears the policy bar).	None
advisor	ControlAdvisor | None	An optional trained control model. It may only raise the decision toward caution — see :func:_combine. Omitting it, or passing one that fails, yields exactly the decision policy alone would have made.	None

Returns:

Name	Type	Description
An	ApprovalDecision	class:ApprovalDecision.

Source code in .sdk/src/tulip/security/policy.py

def approve(
    action: Action,
    *,
    policy: ControlPolicy,
    finding: Evidence | None = None,
    verdict: VerificationResult | None = None,
    advisor: ControlAdvisor | None = None,
) -> ApprovalDecision:
    """Decide whether ``action`` may proceed: allow / require_human / deny.

    Weighs every rule and returns the **strongest** triggered outcome (deny >
    require_human > allow), recording each check that fired so the decision is
    auditable.

    Args:
        action: The proposed action.
        policy: The governing :class:`ControlPolicy`.
        finding: The evidence the action responds to (optional).
        verdict: The :func:`~tulip.security.verify.verify` result (optional, but
            auto-allow needs one that clears the policy bar).
        advisor: An optional trained control model. It may only raise the
            decision toward caution — see :func:`_combine`. Omitting it, or
            passing one that fails, yields exactly the decision policy alone
            would have made.

    Returns:
        An :class:`ApprovalDecision`.
    """
    triggered: list[tuple[str, str]] = []
    labels = action.labels()

    denied = labels & policy.deny_for
    if denied:
        triggered.append((ApprovalOutcome.DENY, f"labels {sorted(denied)} are denied by policy"))

    unsandboxed = labels & policy.require_sandbox_for
    if unsandboxed and SANDBOXED_TAG not in labels:
        triggered.append(
            (
                ApprovalOutcome.DENY,
                f"labels {sorted(unsandboxed)} require sandboxed execution "
                f"(the action carries no {SANDBOXED_TAG!r} tag)",
            )
        )

    if finding is not None and not severity_at_least(finding.severity, policy.min_severity):
        triggered.append(
            (
                ApprovalOutcome.DENY,
                f"finding severity {finding.severity.value} is below the policy "
                f"minimum {policy.min_severity.value}",
            )
        )

    if policy.require_verification_score > 0:
        if verdict is None:
            triggered.append((ApprovalOutcome.REQUIRE_HUMAN, "no verification provided"))
        elif not verdict.survives:
            triggered.append((ApprovalOutcome.DENY, "the finding did not survive verification"))
        elif verdict.confidence < policy.require_verification_score:
            triggered.append(
                (
                    ApprovalOutcome.REQUIRE_HUMAN,
                    f"verification confidence {verdict.confidence:.2f} is below the bar "
                    f"{policy.require_verification_score:.2f}",
                )
            )

    if action.blast_radius > policy.max_blast_radius:
        triggered.append(
            (
                ApprovalOutcome.REQUIRE_HUMAN,
                f"blast radius {action.blast_radius} exceeds the maximum {policy.max_blast_radius}",
            )
        )

    needs_human = labels & policy.require_human_for
    if needs_human:
        triggered.append(
            (ApprovalOutcome.REQUIRE_HUMAN, f"labels {sorted(needs_human)} require human approval")
        )

    if not triggered:
        policy_outcome = ApprovalOutcome.ALLOW
        # Empty, not seeded with a passing note: ``checks`` records what *fired*,
        # and a clean allow fires nothing. That is the existing contract and the
        # combiner must not quietly change it.
        checks: list[str] = []
    else:
        policy_outcome = max((o for o, _ in triggered), key=lambda o: _ORDER[o])
        checks = [why for _, why in triggered]

    outcome, model_outcome, advisory = _combine(action, policy_outcome, advisor)
    if advisory is not None:
        checks = [*checks, advisory]

    return ApprovalDecision(
        outcome=outcome,
        reason="; ".join(checks) if checks else "all policy checks passed",
        action=action,
        checks=checks,
        policy_outcome=policy_outcome,
        model_outcome=model_outcome,
    )

ControlPolicy dataclass

ControlPolicy(require_verification_score: float = 0.8, max_blast_radius: int = 1, require_human_for: frozenset[str] = (lambda: frozenset({'production'}))(), deny_for: frozenset[str] = frozenset(), min_severity: Severity = Severity.LOW, require_sandbox_for: frozenset[str] = frozenset())

The CISO knobs. Defaults are conservative — auto-allow only the safe path.

• require_verification_score: minimum :class:VerificationResult confidence to auto-allow; below it (or with no verdict) a human is required.
• max_blast_radius: most assets an action may affect to auto-allow.
• require_human_for: action labels (environment / kind / tag) that always need a human (default: anything in production).
• deny_for: labels that are hard-denied outright.
• min_severity: don't act on findings below this band.
• require_sandbox_for: labels whose actions must execute in a sandbox — an action matching one of these is denied unless it carries the :data:SANDBOXED_TAG tag. Enforced at the agent loop's tool seam by :class:~tulip.tools.sandbox.SandboxEnforcerHook.

ApprovalDecision dataclass

ApprovalDecision(outcome: str, reason: str, action: Action, checks: list[str] = list(), policy_outcome: str = ApprovalOutcome.ALLOW, model_outcome: str | None = None)

The outcome of weighing an action against evidence, verification, and policy.

escalated_by_model property

escalated_by_model: bool

Whether a control model made this decision stricter than policy alone.

ApprovalOutcome

Outcome labels (kept simple/stable as plain strings).

Describing an action

A policy matches on what an action is — its environment, kind, blast radius, and tags — never on the name of the tool performing it.

Action dataclass

Action(name: str, asset: str = '', blast_radius: int = 1, environment: str = 'unknown', kind: str = '', tags: frozenset[str] = frozenset())

A proposed response action, with the attributes policy reasons over.

labels

labels() -> set[str]

The environment / kind / tags as one label set for policy matching.

Source code in .sdk/src/tulip/security/policy.py

def labels(self) -> set[str]:
    """The environment / kind / tags as one label set for policy matching."""
    return {self.environment, self.kind, *self.tags} - {""}

Deriving action labels

Turn a tool call into an Action using declarative rules, so the labels a policy matches on are not hand-written per call site.

resolve_action

resolve_action(spec: ActionSpec | None, name: str, kwargs: Mapping[str, Any]) -> Action

Resolve an :class:ActionSpec (or None) into a concrete :class:Action.

Source code in .sdk/src/tulip/control/action.py

def resolve_action(spec: ActionSpec | None, name: str, kwargs: Mapping[str, Any]) -> Action:
    """Resolve an :class:`ActionSpec` (or ``None``) into a concrete :class:`Action`."""
    if spec is None:
        return default_action(name, kwargs)
    if isinstance(spec, Action):
        return spec
    return spec(name, kwargs)

default_action

default_action(name: str, kwargs: Mapping[str, Any], *, environment: str = 'unknown', kind: str = '', blast_radius: int = 1, tags: frozenset[str] | None = None) -> Action

A conservative :class:Action for name when none was supplied.

Fail-safe by construction: environment="unknown" plus the stock :class:~tulip.security.policy.ControlPolicy (which requires a verification score) lands an un-verified call on require_human rather than auto-allowing it.

tags defaults to the action's own name, so a policy can always gate one specific tool by naming it — the one thing that worked before labels were derived at all.

Source code in .sdk/src/tulip/control/action.py

def default_action(
    name: str,
    kwargs: Mapping[str, Any],
    *,
    environment: str = "unknown",
    kind: str = "",
    blast_radius: int = 1,
    tags: frozenset[str] | None = None,
) -> Action:
    """A conservative :class:`Action` for ``name`` when none was supplied.

    Fail-safe by construction: ``environment="unknown"`` plus the stock
    :class:`~tulip.security.policy.ControlPolicy` (which requires a verification
    score) lands an un-verified call on ``require_human`` rather than
    auto-allowing it.

    ``tags`` defaults to the action's own name, so a policy can always gate one
    specific tool by naming it — the one thing that worked before labels were
    derived at all.
    """
    return Action(
        name=name,
        asset=asset_from_args(kwargs),
        blast_radius=blast_radius,
        environment=environment,
        kind=kind,
        tags=tags if tags is not None else frozenset({name}),
    )

action_from_labels

action_from_labels(name: str, kwargs: Mapping[str, Any], *, labels: Mapping[str, Any] | None = None, environment: str | None = None, blast_radius: int = 1) -> Action

Build an :class:Action from a tool's declared labels.

labels is what a tool definition declares about the actions it performs — environment, kind, blast_radius, tags. Anything absent falls back: the caller's environment (the agent's, or the deployment's), then "unknown".

The tool's own name is always among the tags, so naming a tool in require_human_for keeps working regardless of what it declares.

labels["derive"] may carry argument-derived rules (see :func:derive_labels) — the only part of this that reads kwargs for labelling. Derived tags join the declared ones, a derived set_kind / set_environment wins over the declared value (it describes this call), and a derived blast radius only ever raises the declared one. With no derive key the result is exactly what it was before.

Source code in .sdk/src/tulip/control/action.py

def action_from_labels(
    name: str,
    kwargs: Mapping[str, Any],
    *,
    labels: Mapping[str, Any] | None = None,
    environment: str | None = None,
    blast_radius: int = 1,
) -> Action:
    """Build an :class:`Action` from a tool's declared labels.

    ``labels`` is what a tool definition declares about the actions it performs
    — ``environment``, ``kind``, ``blast_radius``, ``tags``. Anything absent
    falls back: the caller's ``environment`` (the agent's, or the deployment's),
    then ``"unknown"``.

    The tool's own name is always among the tags, so naming a tool in
    ``require_human_for`` keeps working regardless of what it declares.

    ``labels["derive"]`` may carry argument-derived rules (see
    :func:`derive_labels`) — the only part of this that reads ``kwargs`` for
    labelling. Derived tags join the declared ones, a derived ``set_kind`` /
    ``set_environment`` wins over the declared value (it describes *this* call),
    and a derived blast radius only ever raises the declared one. With no
    ``derive`` key the result is exactly what it was before.
    """
    declared = dict(labels or {})
    declared_env = str(declared.get("environment") or "").strip()
    declared_kind = str(declared.get("kind") or "").strip()
    declared_tags = declared.get("tags") or []
    if isinstance(declared_tags, str):
        declared_tags = [declared_tags]

    radius = declared.get("blast_radius")
    try:
        resolved_radius = int(radius) if radius is not None else blast_radius
    except (TypeError, ValueError):
        resolved_radius = blast_radius

    derived = derive_labels(declared.get("derive"), kwargs)
    if derived.blast_radius is not None:
        resolved_radius = max(resolved_radius, derived.blast_radius)

    tags = {name, *(str(t) for t in declared_tags if t), *derived.tags}
    if derived.undetermined:
        tags.add(UNDETERMINED_TAG)

    return default_action(
        name,
        kwargs,
        environment=(
            derived.environment or declared_env or (environment or "").strip() or "unknown"
        ),
        kind=derived.kind or declared_kind,
        blast_radius=resolved_radius,
        tags=frozenset(tags),
    )

derive_labels

derive_labels(rules: Any, kwargs: Mapping[str, Any]) -> DerivedLabels

Evaluate a tool's derive rules against one call's arguments.

Declarative and total: comparisons only, never eval, never a callable, so nothing a tool receives can execute during labelling. Rules apply in order and all matching rules apply. Anything that cannot be evaluated — a missing argument, an argument of the wrong type, a malformed rule — is skipped and records :data:UNDETERMINED_TAG, so "we could not tell" reaches the policy as a fact rather than as silence.

Source code in .sdk/src/tulip/control/action.py

def derive_labels(rules: Any, kwargs: Mapping[str, Any]) -> DerivedLabels:
    """Evaluate a tool's ``derive`` rules against one call's arguments.

    Declarative and total: comparisons only, never ``eval``, never a callable,
    so nothing a tool receives can execute during labelling. Rules apply in
    order and all matching rules apply. Anything that cannot be evaluated — a
    missing argument, an argument of the wrong type, a malformed rule — is
    skipped *and* records :data:`UNDETERMINED_TAG`, so "we could not tell"
    reaches the policy as a fact rather than as silence.
    """
    derived = DerivedLabels()
    if rules is None:
        return derived
    if not isinstance(rules, list | tuple):
        derived.mark_undetermined()
        return derived
    for rule in rules:
        _apply_rule(rule, kwargs, derived)
    return derived

DerivedLabels

DerivedLabels()

Accumulator for what a derive list adds to an action.

Source code in .sdk/src/tulip/control/action.py

def __init__(self) -> None:
    self.tags: set[str] = set()
    self.kind: str = ""
    self.environment: str = ""
    self.blast_radius: int | None = None
    self.undetermined: bool = False

raise_radius

raise_radius(value: int) -> None

Deriving may raise the blast radius; it may never lower it.

Source code in .sdk/src/tulip/control/action.py

def raise_radius(self, value: int) -> None:
    """Deriving may raise the blast radius; it may never lower it."""
    self.blast_radius = value if self.blast_radius is None else max(self.blast_radius, value)

asset_from_args

asset_from_args(kwargs: Mapping[str, Any]) -> str

Best-effort asset label from a tool call's arguments.

Source code in .sdk/src/tulip/control/action.py

def asset_from_args(kwargs: Mapping[str, Any]) -> str:
    """Best-effort asset label from a tool call's arguments."""
    for key in _ASSET_KEYS:
        value = kwargs.get(key)
        if value:
            return str(value)
    return ""

The record

A hash-chained log of every decision. Each record commits to the previous hash, so editing any record breaks verify().

Tamper-evident, not tamper-proof

This is a keyless SHA-256 chain held in memory. It detects edits when checked against a head hash you retain out-of-band; it does not prevent them, sign them, or anchor the log. Persist the JSONL and pin the head hash externally before relying on it as compliance evidence.

AuditTrail

AuditTrail(*, clock: Callable[[], str] | None = None)

An append-only, hash-chained log of agent actions.

Append with :meth:record (or :meth:record_event for a Tulip event); check integrity with :meth:verify; ship with :meth:export_jsonl. Pass clock to make timestamps deterministic in tests.

Source code in .sdk/src/tulip/security/audit.py

def __init__(self, *, clock: Callable[[], str] | None = None) -> None:
    self._records: list[AuditRecord] = []
    self._clock = clock or _utc_now_iso

head property

head: str

Hash of the latest record, or the genesis anchor when empty.

record

record(event_type: str, payload: Mapping[str, Any] | None = None) -> AuditRecord

Append a record committing to the current chain head.

Source code in .sdk/src/tulip/security/audit.py

def record(self, event_type: str, payload: Mapping[str, Any] | None = None) -> AuditRecord:
    """Append a record committing to the current chain head."""
    seq = len(self._records)
    prev = self.head
    ts = self._clock()
    body = dict(payload or {})
    rec = AuditRecord(
        seq=seq,
        ts=ts,
        event_type=event_type,
        payload=body,
        prev_hash=prev,
        hash=_entry_hash(seq, ts, event_type, body, prev),
    )
    self._records.append(rec)
    return rec

record_event

record_event(event: Any) -> AuditRecord

Append a record for a Tulip event (duck-typed; safe scalar fields).

Source code in .sdk/src/tulip/security/audit.py

def record_event(self, event: Any) -> AuditRecord:
    """Append a record for a Tulip event (duck-typed; safe scalar fields)."""
    payload: dict[str, Any] = {}
    for key in ("name", "tool", "final_message", "reason", "content", "asset"):
        val = getattr(event, key, None)
        if isinstance(val, str | int | float | bool):
            payload[key] = val
    return self.record(type(event).__name__, payload)

records

records() -> list[AuditRecord]

A copy of the records, in order.

Source code in .sdk/src/tulip/security/audit.py

def records(self) -> list[AuditRecord]:
    """A copy of the records, in order."""
    return list(self._records)

verify

verify() -> bool

Whether the chain is intact — no edit, deletion, or reorder.

Source code in .sdk/src/tulip/security/audit.py

def verify(self) -> bool:
    """Whether the chain is intact — no edit, deletion, or reorder."""
    prev = _GENESIS
    for i, rec in enumerate(self._records):
        if rec.seq != i or rec.prev_hash != prev:
            return False
        if _entry_hash(rec.seq, rec.ts, rec.event_type, rec.payload, rec.prev_hash) != rec.hash:
            return False
        prev = rec.hash
    return True

export_jsonl

export_jsonl() -> str

The chain as newline-delimited JSON — one record per line, SIEM-ready.

Source code in .sdk/src/tulip/security/audit.py

def export_jsonl(self) -> str:
    """The chain as newline-delimited JSON — one record per line, SIEM-ready."""
    return "\n".join(json.dumps(asdict(rec), default=str) for rec in self._records)

from_records classmethod

from_records(records: Iterable[AuditRecord]) -> AuditTrail

Rebuild a trail from records (e.g. to :meth:verify an exported chain).

Source code in .sdk/src/tulip/security/audit.py

@classmethod
def from_records(cls, records: Iterable[AuditRecord]) -> AuditTrail:
    """Rebuild a trail from records (e.g. to :meth:`verify` an exported chain)."""
    trail = cls()
    trail._records = list(records)
    return trail

AuditRecord dataclass

AuditRecord(seq: int, ts: str, event_type: str, payload: dict[str, Any], prev_hash: str, hash: str)

One link in the audit chain. hash commits to prev_hash.

AuditHook

AuditHook(trail: AuditTrail, *, priority: int = HookPriority.OBSERVABILITY_DEFAULT)

Bases: HookProvider

Records the agent's lifecycle into a tamper-evident :class:AuditTrail.

Source code in .sdk/src/tulip/security/secure.py

def __init__(
    self,
    trail: AuditTrail,
    *,
    priority: int = HookPriority.OBSERVABILITY_DEFAULT,
) -> None:
    self._trail = trail
    self._priority = priority

name property

name: str

Hook provider name for identification.

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in .sdk/src/tulip/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

Governed agents

An Agent pre-wired with grounding, guardrails, and an audit trail.

governed_agent

governed_agent(model: Any = None, tools: list[Any] | None = None, *, system_prompt: str | None = None, profile: GovernanceProfile | None = None, audit_trail: AuditTrail | None = None, hooks: list[Any] | None = None, **kwargs: Any) -> GovernedAgent

Build a secure-by-default agent: grounded, guarded, and audited.

Parameters:

Name	Type	Description	Default
model	Any	Model string or instance (as :class:tulip.Agent).	None
tools	list[Any] | None	Tools available to the agent.	None
system_prompt	str | None	System prompt.	None
profile	GovernanceProfile | None	Which controls to enable (default: all on).	None
audit_trail	AuditTrail | None	Reuse an existing trail; one is created if omitted.	None
hooks	list[Any] | None	Extra hooks to add alongside the security hooks.	None
**kwargs	Any	Passed through to :class:tulip.Agent.	{}

Returns:

Name	Type	Description
A	GovernedAgent	class:GovernedAgent wrapping the configured agent and its audit trail.

Source code in .sdk/src/tulip/security/secure.py

def governed_agent(
    model: Any = None,
    tools: list[Any] | None = None,
    *,
    system_prompt: str | None = None,
    profile: GovernanceProfile | None = None,
    audit_trail: AuditTrail | None = None,
    hooks: list[Any] | None = None,
    **kwargs: Any,
) -> GovernedAgent:
    """Build a secure-by-default agent: grounded, guarded, and audited.

    Args:
        model: Model string or instance (as :class:`tulip.Agent`).
        tools: Tools available to the agent.
        system_prompt: System prompt.
        profile: Which controls to enable (default: all on).
        audit_trail: Reuse an existing trail; one is created if omitted.
        hooks: Extra hooks to add alongside the security hooks.
        **kwargs: Passed through to :class:`tulip.Agent`.

    Returns:
        A :class:`GovernedAgent` wrapping the configured agent and its audit trail.
    """
    profile = profile or GovernanceProfile()
    # NB: an empty AuditTrail is falsy (len 0), so check identity, not truthiness.
    trail = audit_trail if audit_trail is not None else AuditTrail()
    hook_list: list[Any] = list(hooks or [])
    if profile.guardrails:
        hook_list.append(GuardrailsHook())
    if profile.audit:
        hook_list.append(AuditHook(trail))
    agent = Agent(
        model=model,
        tools=tools,
        system_prompt=system_prompt,
        grounding=profile.grounding,
        hooks=hook_list,
        **kwargs,
    )
    return GovernedAgent(agent=agent, audit_trail=trail, profile=profile)

GovernedAgent dataclass

GovernedAgent(agent: Agent, audit_trail: AuditTrail, profile: GovernanceProfile)

A secure-by-default :class:tulip.Agent plus its audit trail.

run / run_sync pass through to the wrapped agent; audit_trail is the tamper-evident record of everything it did.

arun async

arun(prompt: str, **kwargs: Any) -> Any

Async, thread-free twin of run_sync — delegates to the wrapped agent's arun so a governed agent runs where threads aren't available (e.g. the browser / Pyodide).

Source code in .sdk/src/tulip/security/secure.py

async def arun(self, prompt: str, **kwargs: Any) -> Any:
    """Async, thread-free twin of ``run_sync`` — delegates to the wrapped
    agent's ``arun`` so a governed agent runs where threads aren't
    available (e.g. the browser / Pyodide)."""
    return await self.agent.arun(prompt, **kwargs)

GovernanceProfile dataclass

GovernanceProfile(grounding: bool = True, guardrails: bool = True, audit: bool = True)

Which secure-by-default controls a :func:governed_agent turns on.

All on by default — that is what makes the agent secure out of the box.

Verification

Evidence quality and adversarial refutation, feeding the require_verification_score and min_severity rules on a policy.

verify async

verify(finding: FindingLike, *, skeptics: Sequence[Skeptic] | None = None, threshold: float = 0.6) -> VerificationResult

Independently challenge a finding; return whether it survives.

Runs each skeptic (default: a single :class:EvidenceQualitySkeptic), collects their refutations, and re-grades confidence as the grounding score minus the refutation penalties — where non-fatal penalties are capped (:data:_MAX_NONFATAL_PENALTY) so volume of caveats alone can't refute a well-grounded finding; a single fatal refutation zeroes it outright. A finding survives only if nothing fatal was raised and confidence clears threshold.

Parameters:

Name	Type	Description	Default
finding	FindingLike	A :class:~tulip.security.findings.Evidence or finding-shaped mapping (framework-agnostic).	required
skeptics	Sequence[Skeptic] | None	The challenge panel; defaults to the deterministic skeptic. Plug semantic/LLM skeptics here.	None
threshold	float	Minimum confidence to survive (default 0.6).	0.6

Returns:

Name	Type	Description
A	VerificationResult	class:VerificationResult.

Source code in .sdk/src/tulip/security/verify.py

async def verify(
    finding: FindingLike,
    *,
    skeptics: Sequence[Skeptic] | None = None,
    threshold: float = 0.6,
) -> VerificationResult:
    """Independently challenge a finding; return whether it survives.

    Runs each skeptic (default: a single :class:`EvidenceQualitySkeptic`),
    collects their refutations, and re-grades confidence as the grounding score
    minus the refutation penalties — where non-fatal penalties are **capped**
    (:data:`_MAX_NONFATAL_PENALTY`) so volume of caveats alone can't refute a
    well-grounded finding; a single ``fatal`` refutation zeroes it outright. A
    finding survives only if nothing fatal was raised and confidence clears
    ``threshold``.

    Args:
        finding: A :class:`~tulip.security.findings.Evidence` or finding-shaped
            mapping (framework-agnostic).
        skeptics: The challenge panel; defaults to the deterministic skeptic.
            Plug semantic/LLM skeptics here.
        threshold: Minimum confidence to survive (default 0.6).

    Returns:
        A :class:`VerificationResult`.
    """
    panel: list[Skeptic] = list(skeptics) if skeptics is not None else [EvidenceQualitySkeptic()]
    refutations: list[Refutation] = []
    for skeptic in panel:
        refutations.extend(await skeptic.challenge(finding))

    base = _coerce(finding).gsar_score
    fatal = any(r.weight == "fatal" for r in refutations)
    nonfatal = sum(_PENALTY.get(r.weight, 0.2) for r in refutations if r.weight != "fatal")
    confidence = 0.0 if fatal else max(0.0, min(1.0, base - min(nonfatal, _MAX_NONFATAL_PENALTY)))

    survives = not fatal and confidence >= threshold
    notes = (
        "Survives independent challenge."
        if survives
        else "Refuted by independent challenge — do not act on this finding as-is."
    )
    return VerificationResult(
        survives=survives,
        confidence=confidence,
        evidence_quality=confidence,
        refutations=refutations,
        alternatives=[],
        notes=notes,
    )

VerificationResult dataclass

VerificationResult(survives: bool, confidence: float, evidence_quality: float, refutations: list[Refutation] = list(), alternatives: list[str] = list(), notes: str = '')

The outcome of verifying a finding.

survives is False if any refutation is fatal or confidence falls below the threshold. alternatives is populated by semantic skeptics (the deterministic one leaves it empty).

Evidence

Bases: BaseModel

A grounded security finding.

The gsar_score and evidence_refs fields are required: a Evidence always knows how strongly it is grounded and what it is grounded in. Build findings via :func:tulip.security.ground_finding rather than constructing them directly — that is the path that enforces the grounding threshold.

Severity

Bases: StrEnum

Ordered severity band. StrEnum so it serialises as the bare string.

Not directly comparable with < (string ordering would be wrong); use :func:severity_at_least or :data:SEVERITY_ORDER for ranking.