DevOps Pipeline Example

This example builds a DevOps AI agent that manages service deployments, database migrations, firewall rules, and cluster scaling. A custom infrastructure domain profile (created by you using the domain profile framework) can enforce SOC 2 and ISO 27001 compliance by detecting production environment operations, requiring multi-party approval for deployments, and protecting against unverified rollbacks.

What You Will Build

A deployment agent that gates production releases with multi-party approval
Database migration management with DBA sign-off
Firewall rule changes with teach-back verification
Rollback protection that prevents panic-driven emergency changes
Both Python and TypeScript implementations using LangGraph

Configuration

Create an attesta.yaml in your project root:

attesta.yaml

# ─── Domain Profile (optional) ───────────────────────────────────
# domain: my-domain  # Activate a registered domain profile

# ─── Policy ───────────────────────────────────────────────────────
policy:
  minimum_review_seconds:
    low: 0
    medium: 3
    high: 15
    critical: 45

  require_multi_party:
    critical: 2

  fail_mode: deny
  timeout_seconds: 300

# ─── Risk Scoring ─────────────────────────────────────────────────
risk:
  overrides:
    drop_production_database: critical
    disable_firewall: critical
    delete_all_backups: critical
    terminate_production_cluster: critical
    disable_mfa: critical

  amplifiers:
    - pattern: ".*production.*"
      boost: 0.3
    - pattern: ".*deploy.*"
      boost: 0.15
    - pattern: ".*delete.*"
      boost: 0.2
    - pattern: ".*firewall.*"
      boost: 0.15
    - pattern: ".*migrate.*"
      boost: 0.1

# ─── Trust Engine ─────────────────────────────────────────────────
trust:
  influence: 0.25
  ceiling: 0.85
  initial_score: 0.2
  decay_rate: 0.01

# ─── Audit ────────────────────────────────────────────────────────
audit:
  path: ".attesta/infra-audit.jsonl"

A custom infrastructure domain profile might set a production_multiplier of 1.6. A deployment scored at 0.55 (MEDIUM) in staging would then become 0.88 (CRITICAL) in production, triggering multi-party approval. This is intentional — the same operation carries fundamentally different risk in production. See Custom Domains for how to create your own domain profile.

Python Implementation

Install dependencies

pip install attesta[all] langchain-openai langgraph

Define the infrastructure tools

Each function is gated by Attesta. A custom domain profile can be configured to detect deployment operations, database commands, firewall changes, and privilege escalation.

infra_agent.py

from attesta import Attesta
from attesta import AttestaDenied

# Load configuration (optionally with a custom domain profile)
attesta = Attesta.from_config("attesta.yaml")


# ─── Low risk: read-only status checks ────────────────────────────
@attesta.gate()
def get_service_status(service: str, environment: str = "production") -> dict:
    """Check the health and deployment status of a service."""
    return {
        "service": service,
        "environment": environment,
        "status": "healthy",
        "version": "2.0.3",
        "instances": 4,
        "uptime_hours": 720,
        "last_deploy": "2026-02-10T14:30:00Z",
    }


@attesta.gate()
def list_recent_deployments(environment: str, limit: int = 10) -> dict:
    """List recent deployments to an environment."""
    return {
        "environment": environment,
        "deployments": [
            {"service": "api-gateway", "version": "2.0.3",
             "deployed_at": "2026-02-10T14:30:00Z", "status": "active"},
            {"service": "auth-service", "version": "1.8.1",
             "deployed_at": "2026-02-09T09:15:00Z", "status": "active"},
            {"service": "payment-api", "version": "3.2.0",
             "deployed_at": "2026-02-08T16:45:00Z", "status": "active"},
        ],
    }


# ─── Medium risk: staging deployments ─────────────────────────────
@attesta.gate(risk_hints={"environment": "staging"})
def deploy_to_staging(
    service: str, version: str, strategy: str = "rolling"
) -> dict:
    """Deploy a service to the staging environment.

    Uses the specified deployment strategy (rolling, blue-green,
    or canary). Staging deployments require confirmation but not
    multi-party approval.
    """
    return {
        "deployment_id": "DEP-STG-2026-001",
        "service": service,
        "version": version,
        "environment": "staging",
        "strategy": strategy,
        "status": "deployed",
        "instances_updated": 2,
    }


# ─── High/Critical risk: production deployments ───────────────────
@attesta.gate(risk_hints={"production": True})
def deploy_to_production(
    service: str,
    version: str,
    strategy: str = "canary",
    canary_percentage: int = 10,
    change_ticket: str = "",
) -> dict:
    """Deploy a service to the production environment.

    Production deployments require approval from the SRE lead and
    release manager per your domain's escalation rules.
    A valid change ticket from the CAB is expected.
    """
    return {
        "deployment_id": "DEP-PROD-2026-042",
        "service": service,
        "version": version,
        "environment": "production",
        "strategy": strategy,
        "canary_percentage": canary_percentage,
        "change_ticket": change_ticket,
        "status": "canary_deployed",
        "healthy_instances": 1,
        "total_instances": 10,
    }


# ─── High risk: rollback operations ───────────────────────────────
@attesta.gate(risk_hints={"production": True, "emergency": True})
def rollback_deployment(
    service: str,
    target_version: str,
    reason: str,
) -> dict:
    """Rollback a production service to a previous version.

    Rollbacks are emergency operations that bypass the normal
    deployment pipeline. They still require approval to prevent
    panic-driven changes that make incidents worse.
    """
    return {
        "rollback_id": "RBK-2026-007",
        "service": service,
        "from_version": "2.1.0",
        "to_version": target_version,
        "environment": "production",
        "reason": reason,
        "status": "rolled_back",
        "instances_reverted": 10,
    }


# ─── Critical risk: database migrations ───────────────────────────
@attesta.gate(
    risk_hints={"production": True, "affects_database": True}
)
def run_migration(
    migration_file: str,
    database: str,
    environment: str = "production",
) -> dict:
    """Run a database schema migration.

    Schema migrations in production require approval from the DBA
    and backend engineering lead per your domain's escalation
    rules. Migrations are irreversible -- always verify the
    migration file contents before approving.
    """
    return {
        "migration_id": "MIG-2026-015",
        "migration_file": migration_file,
        "database": database,
        "environment": environment,
        "tables_affected": ["users", "orders"],
        "status": "migration_applied",
        "duration_seconds": 12.4,
    }


# ─── High risk: firewall rule changes ─────────────────────────────
@attesta.gate(risk_hints={"security_boundary": True})
def modify_firewall_rule(
    rule_id: str,
    action: str,
    protocol: str,
    port: int,
    cidr: str,
    direction: str = "ingress",
) -> dict:
    """Modify a firewall rule in the production VPC.

    Firewall changes affect the network security boundary. Your
    domain profile can require teach-back verification and
    notification of the security engineer and network admin.
    """
    return {
        "rule_id": rule_id,
        "action": action,
        "protocol": protocol,
        "port": port,
        "cidr": cidr,
        "direction": direction,
        "status": "rule_updated",
    }


# ─── High risk: credential rotation ───────────────────────────────
@attesta.gate(risk_hints={"security_critical": True})
def rotate_credentials(
    service: str,
    credential_type: str,
    environment: str = "production",
) -> dict:
    """Rotate credentials for a production service.

    Credential rotation temporarily creates a window where both
    old and new credentials are valid. Requires verification to
    ensure dependent services are prepared.
    """
    return {
        "service": service,
        "credential_type": credential_type,
        "environment": environment,
        "rotation_id": "ROT-2026-003",
        "status": "rotated",
        "old_credential_expires_at": "2026-02-14T09:00:00Z",
    }

Build the agent loop with LangGraph

This agent uses LangGraph with Attesta inserted as a gate node between the agent and tool execution.

infra_loop.py

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.tools import StructuredTool
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

from attesta import Attesta
from attesta.integrations.langchain import (
    AttestaToolWrapper,
    attesta_node,
)

from infra_agent import (
    get_service_status,
    list_recent_deployments,
    deploy_to_staging,
    deploy_to_production,
    rollback_deployment,
    run_migration,
    modify_firewall_rule,
    rotate_credentials,
)

# Load configuration
attesta = Attesta.from_config("attesta.yaml")
llm = ChatOpenAI(model="gpt-4o")

# Create LangChain tools from the gated functions
tools = [
    StructuredTool.from_function(get_service_status),
    StructuredTool.from_function(list_recent_deployments),
    StructuredTool.from_function(deploy_to_staging),
    StructuredTool.from_function(deploy_to_production),
    StructuredTool.from_function(rollback_deployment),
    StructuredTool.from_function(run_migration),
    StructuredTool.from_function(modify_firewall_rule),
    StructuredTool.from_function(rotate_credentials),
]

# Wrap tools with Attesta -- risk overrides for known-dangerous ops
wrapper = AttestaToolWrapper(
    attesta,
    risk_overrides={
        "deploy_to_production": "high",
        "rollback_deployment": "high",
        "run_migration": "critical",
        "modify_firewall_rule": "high",
    },
)
protected_tools = wrapper.wrap_tools(tools)

# Bind tools to the LLM
llm_with_tools = llm.bind_tools(protected_tools)


# Define graph nodes
async def agent_node(state: MessagesState) -> MessagesState:
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}


def should_continue(state: MessagesState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "gate"
    return "__end__"


# Build the LangGraph
builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("gate", attesta_node(attesta))
builder.add_node("tools", ToolNode(protected_tools))

builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("gate", "tools")
builder.add_edge("tools", "agent")

graph = builder.compile()


async def run_infra_agent(user_message: str) -> str:
    """Run the infrastructure agent with Attesta gating."""
    result = await graph.ainvoke(
        {"messages": [("user", user_message)]}
    )
    for msg in reversed(result["messages"]):
        if hasattr(msg, "content") and msg.content:
            return msg.content
    return "No response"


if __name__ == "__main__":
    # Scenario 1: Status check (LOW)
    print("--- Scenario 1: Service Status (LOW) ---")
    result = asyncio.run(
        run_infra_agent(
            "What is the current status of the api-gateway service?"
        )
    )
    print(result)

    # Scenario 2: Staging deployment (MEDIUM)
    print("\n--- Scenario 2: Staging Deploy (MEDIUM) ---")
    result = asyncio.run(
        run_infra_agent(
            "Deploy api-gateway v2.1.0 to staging using a "
            "rolling update strategy."
        )
    )
    print(result)

    # Scenario 3: Production deployment (CRITICAL)
    print("\n--- Scenario 3: Production Deploy (CRITICAL) ---")
    result = asyncio.run(
        run_infra_agent(
            "Deploy api-gateway v2.1.0 to production with a 10% "
            "canary rollout. Change ticket is CHG-2026-0042."
        )
    )
    print(result)

    # Scenario 4: Database migration (CRITICAL)
    print("\n--- Scenario 4: Database Migration (CRITICAL) ---")
    result = asyncio.run(
        run_infra_agent(
            "Run migration 20260213_add_user_preferences.sql on the "
            "primary database in production."
        )
    )
    print(result)

    # Scenario 5: Emergency rollback (HIGH)
    print("\n--- Scenario 5: Emergency Rollback (HIGH) ---")
    result = asyncio.run(
        run_infra_agent(
            "Rollback api-gateway to v2.0.3 in production. "
            "Reason: v2.1.0 is causing 502 errors on /api/checkout."
        )
    )
    print(result)

    # Scenario 6: Firewall change (HIGH)
    print("\n--- Scenario 6: Firewall Rule Change (HIGH) ---")
    result = asyncio.run(
        run_infra_agent(
            "Open port 443 for ingress from CIDR 10.0.0.0/8 on "
            "firewall rule FW-PROD-001. Protocol TCP."
        )
    )
    print(result)

Run the agent

export OPENAI_API_KEY="sk-..."
python infra_loop.py

TypeScript Implementation

Install dependencies

npm install @kyberon/attesta @langchain/openai @langchain/langgraph @langchain/core zod

Build the infrastructure agent

infra-agent.ts

import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, MessagesAnnotation } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { Attesta } from "@kyberon/attesta";
import { gatedTool, createGateNode } from "@kyberon/attesta/integrations";

// Load configuration (optionally with a custom domain profile)
const attesta = new Attesta();

// ─── Define infrastructure tools ───────────────────────────────────

const getServiceStatus = tool(
  async ({ service, environment }) => {
    return JSON.stringify({
      service,
      environment,
      status: "healthy",
      version: "2.0.3",
      instances: 4,
      uptime_hours: 720,
    });
  },
  {
    name: "get_service_status",
    description: "Check the health and deployment status of a service.",
    schema: z.object({
      service: z.string().describe("Service name"),
      environment: z
        .string()
        .default("production")
        .describe("Target environment"),
    }),
  }
);

const deployToProduction = tool(
  async ({ service, version, strategy, canaryPercentage, changeTicket }) => {
    return JSON.stringify({
      deployment_id: "DEP-PROD-2026-042",
      service,
      version,
      environment: "production",
      strategy,
      canary_percentage: canaryPercentage,
      change_ticket: changeTicket,
      status: "canary_deployed",
    });
  },
  {
    name: "deploy_to_production",
    description:
      "Deploy a service to production. Requires SRE and release manager approval.",
    schema: z.object({
      service: z.string(),
      version: z.string(),
      strategy: z
        .enum(["canary", "rolling", "blue-green"])
        .default("canary"),
      canaryPercentage: z.number().default(10),
      changeTicket: z.string().default(""),
    }),
  }
);

const rollbackDeployment = tool(
  async ({ service, targetVersion, reason }) => {
    return JSON.stringify({
      rollback_id: "RBK-2026-007",
      service,
      from_version: "2.1.0",
      to_version: targetVersion,
      reason,
      status: "rolled_back",
    });
  },
  {
    name: "rollback_deployment",
    description:
      "Rollback a production service to a previous version. Emergency operation.",
    schema: z.object({
      service: z.string(),
      targetVersion: z.string(),
      reason: z.string(),
    }),
  }
);

const runMigration = tool(
  async ({ migrationFile, database, environment }) => {
    return JSON.stringify({
      migration_id: "MIG-2026-015",
      migration_file: migrationFile,
      database,
      environment,
      tables_affected: ["users", "orders"],
      status: "migration_applied",
    });
  },
  {
    name: "run_migration",
    description:
      "Run a database schema migration. Requires DBA and backend lead approval.",
    schema: z.object({
      migrationFile: z.string(),
      database: z.string(),
      environment: z.string().default("production"),
    }),
  }
);

const modifyFirewallRule = tool(
  async ({ ruleId, action, protocol, port, cidr, direction }) => {
    return JSON.stringify({
      rule_id: ruleId,
      action,
      protocol,
      port,
      cidr,
      direction,
      status: "rule_updated",
    });
  },
  {
    name: "modify_firewall_rule",
    description:
      "Modify a firewall rule. Affects network security boundary.",
    schema: z.object({
      ruleId: z.string(),
      action: z.enum(["allow", "deny"]),
      protocol: z.enum(["tcp", "udp", "icmp"]),
      port: z.number(),
      cidr: z.string(),
      direction: z.enum(["ingress", "egress"]).default("ingress"),
    }),
  }
);

// ─── Gate the tools ────────────────────────────────────────────────

const gatedTools = [
  gatedTool(getServiceStatus, {
    agentId: "infra-agent",
    environment: "production",
  }),
  gatedTool(deployToProduction, {
    agentId: "infra-agent",
    environment: "production",
    riskHints: { production: true },
  }),
  gatedTool(rollbackDeployment, {
    agentId: "infra-agent",
    environment: "production",
    riskHints: { production: true, emergency: true },
  }),
  gatedTool(runMigration, {
    agentId: "infra-agent",
    environment: "production",
    riskHints: { production: true, affects_database: true },
  }),
  gatedTool(modifyFirewallRule, {
    agentId: "infra-agent",
    environment: "production",
    riskHints: { security_boundary: true },
  }),
];

// ─── Build the LangGraph ───────────────────────────────────────────

const llm = new ChatOpenAI({ model: "gpt-4o" });
const llmWithTools = llm.bindTools(gatedTools);

async function agentNode(state: typeof MessagesAnnotation.State) {
  const response = await llmWithTools.invoke(state.messages);
  return { messages: [response] };
}

function shouldContinue(state: typeof MessagesAnnotation.State): string {
  const lastMsg = state.messages.at(-1);
  if (
    lastMsg &&
    "tool_calls" in lastMsg &&
    (lastMsg as any).tool_calls?.length > 0
  ) {
    return "gate";
  }
  return "__end__";
}

const gateNode = createGateNode({
  actionName: "infra-operation",
  agentId: "infra-agent",
  environment: "production",
});

const graph = new StateGraph(MessagesAnnotation)
  .addNode("agent", agentNode)
  .addNode("gate", gateNode)
  .addNode("tools", new ToolNode(gatedTools))
  .addConditionalEdges("agent", shouldContinue)
  .addEdge("gate", "tools")
  .addEdge("tools", "agent")
  .compile();

// ─── Run scenarios ─────────────────────────────────────────────────

async function main() {
  console.log("--- Scenario 1: Service Status (LOW) ---");
  const status = await graph.invoke({
    messages: [
      { role: "user", content: "What is the status of api-gateway?" },
    ],
  });
  console.log(status.messages.at(-1)?.content);

  console.log("\n--- Scenario 2: Production Deploy (CRITICAL) ---");
  const deploy = await graph.invoke({
    messages: [
      {
        role: "user",
        content:
          "Deploy api-gateway v2.1.0 to production with 10% canary. " +
          "Change ticket CHG-2026-0042.",
      },
    ],
  });
  console.log(deploy.messages.at(-1)?.content);

  console.log("\n--- Scenario 3: Database Migration (CRITICAL) ---");
  const migration = await graph.invoke({
    messages: [
      {
        role: "user",
        content:
          "Run migration 20260213_add_user_preferences.sql " +
          "on the primary database in production.",
      },
    ],
  });
  console.log(migration.messages.at(-1)?.content);

  console.log("\n--- Scenario 4: Emergency Rollback (HIGH) ---");
  const rollback = await graph.invoke({
    messages: [
      {
        role: "user",
        content:
          "Rollback api-gateway to v2.0.3. " +
          "v2.1.0 is causing 502 errors on /api/checkout.",
      },
    ],
  });
  console.log(rollback.messages.at(-1)?.content);

  console.log("\n--- Scenario 5: Firewall Change (HIGH) ---");
  const firewall = await graph.invoke({
    messages: [
      {
        role: "user",
        content:
          "Allow TCP ingress on port 443 from 10.0.0.0/8 " +
          "on rule FW-PROD-001.",
      },
    ],
  });
  console.log(firewall.messages.at(-1)?.content);
}

main().catch(console.error);

Run the agent

export OPENAI_API_KEY="sk-..."
npx tsx infra-agent.ts

Expected Terminal Output

Scenario 1: Service Status (LOW — auto-approved)

Read-only status check. No deployment or destructive patterns match. Auto-approved silently.

Field	Value
Risk Level	LOW
Action	`get_service_status`
Risk Score	0.10
Domain	custom (SOC 2)
Result	AUTO-APPROVED

Scenario 2: Staging Deployment (MEDIUM — confirm)

The deploy keyword triggers the deployment pattern (contribution: 0.8), but the staging environment hint keeps the risk in MEDIUM range. A simple confirmation is required.

Field	Value
Risk Level	MEDIUM
Action	`deploy_to_staging`
Risk Score	0.42
Domain	custom (SOC 2)
Arguments	`service="api-gateway"`, `version="2.1.0"`, `strategy="rolling"`
Challenge	CONFIRM (min review: 3s)
Prompt	Deploy api-gateway v2.1.0 to staging? [y/N] y
Result	APPROVED (reviewed in 4.2s)

Scenario 3: Production Deployment (CRITICAL — multi-party)

The same deploy function name, but with the production: True hint, triggers the production multiplier (1.6x). The combined score crosses into CRITICAL, requiring dual approval from the SRE lead and release manager.

Field	Value
Risk Level	CRITICAL
Action	`deploy_to_production`
Risk Score	0.88
Domain	custom (SOC 2)
Arguments	`service="api-gateway"`, `version="2.1.0"`, `strategy="canary"`, `canary_percentage=10`, `change_ticket="CHG-2026-0042"`
Escalation	Production deployment
Notified	sre_lead, release_manager
Challenge	MULTI-PARTY (2 of 2 required)

Approver 1 (sre.lead@company.com) — TEACH-BACK:

Deploying api-gateway v2.1.0 to production with a 10% canary rollout under change ticket CHG-2026-0042. This will update 1 of 10 instances initially. Approved (48.2s)

Approver 2 (release.mgr@company.com) — QUIZ:

Q1: Which service is being deployed? api-gateway Q2: What is the change ticket? CHG-2026-0042 Approved (35.4s)

Result: APPROVED (2/2 approvers, total 83.6s)

Scenario 4: Database Migration (CRITICAL — multi-party)

The run_migration function is overridden to CRITICAL risk. A custom domain’s schema migration escalation rule can require DBA and backend lead approval.

Field	Value
Risk Level	CRITICAL
Action	`run_migration`
Risk Score	0.91
Domain	custom (SOC 2)
Arguments	`migration_file="20260213_add_user_preferences.sql"`, `database="primary"`, `environment="production"`
Escalation	Schema migration
Notified	dba, backend_lead
Challenge	MULTI-PARTY (2 of 2 required)

Approver 1 (dba@company.com) — TEACH-BACK:

Running migration 20260213_add_user_preferences.sql on the primary production database. This will alter the users and orders tables to add preference columns. Approved (55.1s)

Approver 2 (backend.lead@company.com) — QUIZ:

Q1: Which database is affected? primary Q2: Is this migration reversible? No (DDL) Approved (41.3s)

Result: APPROVED (2/2 approvers, total 96.4s)

Scenario 5: Emergency Rollback (HIGH — quiz)

Rollbacks are emergency operations. Despite the urgency, Attesta requires a quiz challenge to verify the operator understands the rollback impact. The 15-second minimum review prevents panic-driven approvals.

Field	Value
Risk Level	HIGH
Action	`rollback_deployment`
Risk Score	0.72
Domain	custom (SOC 2)
Arguments	`service="api-gateway"`, `target_version="2.0.3"`, `reason="v2.1.0 causing 502s on /api/checkout"`
Challenge	QUIZ (min review: 15s)

Q1: Which version will the service revert to? 2.0.3 — Correct Q2: What is the reported issue? 502 errors on the checkout endpoint — Correct Q3: How many production instances will be reverted? All 10 — Correct

Result: APPROVED (reviewed in 23.7s)

Rollback protection is deliberate. During incidents, the instinct is to approve everything instantly. Attesta’s minimum review time (15s for HIGH) prevents this while keeping the delay short enough for genuine emergencies. If you need faster rollbacks, lower the high review time — but never remove it entirely.

Scenario 6: Firewall Rule Change (HIGH — teach-back)

Firewall changes trigger the custom domain’s firewall modification escalation rule, requiring teach-back verification. The security engineer and network admin are notified.

Field	Value
Risk Level	HIGH
Action	`modify_firewall_rule`
Risk Score	0.78
Domain	custom (SOC 2)
Arguments	`rule_id="FW-PROD-001"`, `action="allow"`, `protocol="TCP"`, `port=443`, `cidr="10.0.0.0/8"`, `direction="ingress"`
Escalation	Firewall modification
Notified	security_engineer, network_admin
Challenge	TEACH-BACK (min review: 15s)

Explain what this firewall change will do: This will allow inbound TCP traffic on port 443 from the 10.0.0.0/8 CIDR range (internal network) through firewall rule FW-PROD-001. This opens HTTPS access from internal services to the production environment. Key terms matched: allow, TCP, 443, 10.0.0.0/8, ingress

Result: APPROVED (reviewed in 34.8s)

Audit Trail

After running the scenarios, the audit log at .attesta/infra-audit.jsonl contains entries for every operation:

{
  "entry_id": "aud_d3e4f5a6",
  "chain_hash": "e8f2a1b3c5d7...e9f1a3b5c7d9e1",
  "previous_hash": "b4c7d9e1f3a5...c9d1e3f5a7b9c1",
  "action_name": "deploy_to_production",
  "action_description": "Deploy api-gateway v2.1.0 to production (canary 10%)",
  "agent_id": "infra-agent",
  "risk_score": 0.88,
  "risk_level": "critical",
  "challenge_type": "multi_party",
  "challenge_passed": true,
  "approver_ids": [
    "sre.lead@company.com",
    "release.mgr@company.com"
  ],
  "verdict": "approved",
  "review_duration_seconds": 83.6,
  "min_review_met": true,
  "intercepted_at": "2026-02-13T10:30:00.000Z",
  "decided_at": "2026-02-13T10:31:23.600Z",
  "executed_at": "2026-02-13T10:31:23.750Z",
  "session_id": "sess_infra_g7h8i9",
  "environment": "production",
  "metadata": {
    "domain": "my-infra",
    "service": "api-gateway",
    "version": "2.1.0",
    "strategy": "canary",
    "change_ticket": "CHG-2026-0042",
    "soc2_control": "CC8.1"
  }
}

Verify and query the infrastructure audit trail:

# Verify chain integrity
attesta audit verify .attesta/infra-audit.jsonl

# Output:
# ✓ Chain integrity verified (6 entries)
# ✓ All minimum review times met
# ✓ Multi-party approvals validated

# Export all production deployments
attesta audit export --action "deploy_to_production" .attesta/infra-audit.jsonl

# Export all denied actions (useful for incident post-mortems)
attesta audit export --verdict denied .attesta/infra-audit.jsonl

# Check for rubber-stamped rollbacks during incidents
attesta audit rubber-stamps .attesta/infra-audit.jsonl

Risk Level Summary

Function	Risk Level	Challenge	Key Trigger
`get_service_status`	LOW	Auto-approve	Read-only status check
`list_recent_deployments`	LOW	Auto-approve	Read-only history
`deploy_to_staging`	MEDIUM	Confirm (3s)	Deploy pattern + staging environment
`deploy_to_production`	CRITICAL	Multi-party, 2 approvers (45s)	Deploy pattern + production multiplier (1.6x)
`rollback_deployment`	HIGH	Quiz (15s)	Rollback pattern + emergency hint
`run_migration`	CRITICAL	Multi-party, 2 approvers (45s)	Risk override + schema migration escalation
`modify_firewall_rule`	HIGH	Teach-back (15s)	Firewall pattern + security boundary hint
`rotate_credentials`	HIGH	Quiz (15s)	Credential management pattern

A custom infrastructure domain profile can assign lower risk to canary and blue-green deployments (contribution: 0.5) compared to full rollouts (contribution: 0.8). If you primarily use canary deployments, you may find that production deploys land in HIGH rather than CRITICAL. Adjust the production amplifier boost if you want all production deploys to require multi-party approval regardless of strategy.

Compliance Mapping

Framework	Attesta Mechanism
SOC 2 CC6.1 — Logical access security	Access control patterns flag permission and credential changes
SOC 2 CC7.2 — Monitoring system components	Audit trail records all infrastructure operations
SOC 2 CC8.1 — Change management	Multi-party approval for production deployments and migrations
ISO 27001 A.9 — Access control	Credential rotation and permission changes require verification
ISO 27001 A.12 — Operations security	Production multiplier escalates risk for live environment changes
ISO 27001 A.13 — Communications security	Firewall patterns with teach-back challenges

Next Steps

Custom Domain Profiles

Build your own domain-specific risk profiles

Multi-Framework

Same config across LangChain, OpenAI, and Claude

LangChain Integration

Full LangChain and LangGraph reference

Getting Started

Core Concepts

Configuration

Domain Profiles

Examples

Guides

TypeScript

Resources

DevOps Pipeline Example

What You Will Build

Configuration

Python Implementation

TypeScript Implementation

Expected Terminal Output

Scenario 1: Service Status (LOW — auto-approved)

Scenario 2: Staging Deployment (MEDIUM — confirm)

Scenario 3: Production Deployment (CRITICAL — multi-party)

Scenario 4: Database Migration (CRITICAL — multi-party)

Scenario 5: Emergency Rollback (HIGH — quiz)

Scenario 6: Firewall Rule Change (HIGH — teach-back)

Audit Trail

Risk Level Summary

Compliance Mapping

Next Steps

Custom Domain Profiles

Multi-Framework

LangChain Integration

​What You Will Build

​Configuration

​Python Implementation

​TypeScript Implementation

​Expected Terminal Output

​Scenario 1: Service Status (LOW — auto-approved)

​Scenario 2: Staging Deployment (MEDIUM — confirm)

​Scenario 3: Production Deployment (CRITICAL — multi-party)

​Scenario 4: Database Migration (CRITICAL — multi-party)

​Scenario 5: Emergency Rollback (HIGH — quiz)

​Scenario 6: Firewall Rule Change (HIGH — teach-back)

​Audit Trail

​Risk Level Summary

​Compliance Mapping

​Next Steps

Custom Domain Profiles

Multi-Framework

LangChain Integration

What You Will Build

Configuration

Python Implementation

TypeScript Implementation

Expected Terminal Output

Scenario 1: Service Status (LOW — auto-approved)

Scenario 2: Staging Deployment (MEDIUM — confirm)

Scenario 3: Production Deployment (CRITICAL — multi-party)

Scenario 4: Database Migration (CRITICAL — multi-party)

Scenario 5: Emergency Rollback (HIGH — quiz)

Scenario 6: Firewall Rule Change (HIGH — teach-back)

Audit Trail

Risk Level Summary

Compliance Mapping

Next Steps