DevOps Pipeline Example

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",
    }

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)

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);