Multi-Framework Example

# ─── Domain Profile ───────────────────────────────────────────────
# domain: my-domain  # Optional: activate a registered domain profile

# ─── Policy ───────────────────────────────────────────────────────
policy:
  minimum_review_seconds:
    low: 0
    medium: 3
    high: 10
    critical: 30

  require_multi_party:
    critical: 2

  fail_mode: deny
  timeout_seconds: 300

# ─── Risk Scoring ─────────────────────────────────────────────────
risk:
  overrides:
    deploy_to_production: critical
    delete_database: critical
    modify_firewall: critical

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

# ─── Trust Engine ─────────────────────────────────────────────────
trust:
  influence: 0.3
  ceiling: 0.9
  initial_score: 0.3
  decay_rate: 0.01

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

from attesta import Attesta
from attesta import AttestaDenied

# Single Attesta instance shared across all frameworks
attesta = Attesta.from_config("attesta.yaml")


# ─── Low risk: read-only operations ──────────────────────────────
@attesta.gate()
def get_service_status(service: str) -> dict:
    """Check the health status of a service. Read-only."""
    return {
        "service": service,
        "status": "healthy",
        "version": "2.0.3",
        "uptime": "99.97%",
    }


@attesta.gate()
def list_services(environment: str = "production") -> dict:
    """List all services in an environment. Read-only."""
    return {
        "environment": environment,
        "services": [
            {"name": "api-gateway", "version": "2.0.3", "status": "healthy"},
            {"name": "auth-service", "version": "1.8.1", "status": "healthy"},
            {"name": "worker", "version": "3.2.0", "status": "healthy"},
        ],
    }


# ─── Medium risk: state-changing but reversible ──────────────────
@attesta.gate(risk_hints={"affects_config": True})
def update_config(
    service: str, key: str, value: str, environment: str = "staging"
) -> dict:
    """Update a configuration value for a service."""
    return {
        "service": service,
        "key": key,
        "value": value,
        "environment": environment,
        "status": "config_updated",
    }


# ─── High risk: production database query ────────────────────────
@attesta.gate(risk_hints={"production": True, "database": True})
def execute_query(query: str, database: str = "primary") -> dict:
    """Execute a SQL query against a production database.

    Read queries are lower risk; write queries are flagged by the
    risk scorer based on SQL verb detection (INSERT, UPDATE, DELETE).
    """
    return {
        "query": query,
        "database": database,
        "rows_affected": 0,
        "status": "query_executed",
    }


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

    This is a critical action requiring multi-party approval from
    the SRE lead and release manager.
    """
    return {
        "deployment_id": "DEP-2026-042",
        "service": service,
        "version": version,
        "environment": "production",
        "strategy": strategy,
        "status": "deployed",
    }


# ─── Critical risk: destructive operations ───────────────────────
@attesta.gate(risk_hints={"destructive": True, "production": True})
def delete_database(database: str, confirm_name: str) -> dict:
    """Delete a production database. Irreversible.

    The confirm_name parameter must match the database name exactly
    as a safety check.
    """
    if confirm_name != database:
        raise ValueError("Confirmation name does not match database name")
    return {
        "database": database,
        "status": "deleted",
    }

import { Attesta, createActionContext, Verdict } from "@kyberon/attesta";
import { gatedTool } from "@kyberon/attesta/integrations";

// ─── Shared Attesta instance ──────────────────────────────────────
const attesta = new Attesta();

// ─── Shared tool functions ────────────────────────────────────────
async function getServiceStatus(service: string) {
  return { service, status: "healthy", version: "2.0.3" };
}

async function deployToProduction(service: string, version: string) {
  return { service, version, environment: "production", status: "deployed" };
}

// ─── LangChain integration ────────────────────────────────────────
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const lcGetStatus = gatedTool(
  tool(
    async ({ service }) => JSON.stringify(await getServiceStatus(service)),
    {
      name: "get_service_status",
      description: "Check service health",
      schema: z.object({ service: z.string() }),
    }
  ),
  { agentId: "devops-agent", environment: "production" }
);

const lcDeploy = gatedTool(
  tool(
    async ({ service, version }) =>
      JSON.stringify(await deployToProduction(service, version)),
    {
      name: "deploy_to_production",
      description: "Deploy to production",
      schema: z.object({ service: z.string(), version: z.string() }),
    }
  ),
  {
    agentId: "devops-agent",
    environment: "production",
    riskHints: { production: true },
  }
);

// ─── OpenAI integration ───────────────────────────────────────────
// Uses the same attesta instance via createActionContext
async function gateForOpenAI(
  toolName: string,
  args: Record<string, unknown>
): Promise<boolean> {
  const ctx = createActionContext({
    functionName: toolName,
    args: [],
    kwargs: args,
    hints: {},
    environment: "production",
    metadata: { source: "openai-agents" },
  });

  const result = await attesta.evaluate(ctx);
  return result.verdict === Verdict.APPROVED;
}

// ─── Anthropic integration ────────────────────────────────────────
// Uses the same attesta instance via createActionContext
async function gateForClaude(
  toolName: string,
  args: Record<string, unknown>
): Promise<boolean> {
  const ctx = createActionContext({
    functionName: toolName,
    args: [],
    kwargs: args,
    hints: {},
    environment: "production",
    metadata: { source: "anthropic-claude" },
  });

  const result = await attesta.evaluate(ctx);
  return result.verdict === Verdict.APPROVED;
}

// ─── All three frameworks use the same audit trail ────────────────
async function main() {
  // LangChain agent calls get_service_status
  console.log("--- LangChain: get_service_status ---");
  const statusResult = await lcGetStatus.invoke({ service: "api-gateway" });
  console.log(statusResult);

  // OpenAI agent calls deploy_to_production
  console.log("\n--- OpenAI: deploy_to_production ---");
  const openaiApproved = await gateForOpenAI("deploy_to_production", {
    service: "api-gateway",
    version: "2.1.0",
  });
  if (openaiApproved) {
    console.log(await deployToProduction("api-gateway", "2.1.0"));
  } else {
    console.log("Deployment denied by Attesta");
  }

  // Claude agent calls get_service_status
  console.log("\n--- Claude: get_service_status ---");
  const claudeApproved = await gateForClaude("get_service_status", {
    service: "auth-service",
  });
  if (claudeApproved) {
    console.log(await getServiceStatus("auth-service"));
  }
}

main().catch(console.error);

import asyncio
from shared_tools import attesta

# Import the three agents
from langchain_agent import executor as langchain_executor
from openai_agent import run_openai_agent
from claude_agent import run_claude_agent


async def main():
    print("=" * 60)
    print("MULTI-FRAMEWORK DEMO")
    print("All three frameworks share the same attesta.yaml")
    print("=" * 60)

    # 1. LangChain: status check (LOW -- auto-approved)
    print("\n--- LangChain Agent: Service Status (LOW) ---")
    lc_result = langchain_executor.invoke({
        "input": "What is the status of api-gateway?"
    })
    print(f"Result: {lc_result['output']}")

    # 2. OpenAI: database query (HIGH -- quiz)
    print("\n--- OpenAI Agent: Database Query (HIGH) ---")
    oai_result = await run_openai_agent(
        "Run SELECT count(*) FROM orders WHERE status='pending'"
    )
    print(f"Result: {oai_result}")

    # 3. Claude: production deployment (CRITICAL -- multi-party)
    print("\n--- Claude Agent: Production Deploy (CRITICAL) ---")
    claude_result = await run_claude_agent(
        "Deploy api-gateway v2.1.0 to production with canary strategy"
    )
    print(f"Result: {claude_result}")

    # 4. LangChain: config update (MEDIUM -- confirm)
    print("\n--- LangChain Agent: Config Update (MEDIUM) ---")
    lc_result2 = langchain_executor.invoke({
        "input": "Update the log_level to debug for api-gateway in staging"
    })
    print(f"Result: {lc_result2['output']}")

    print("\n" + "=" * 60)
    print("All actions recorded in a single audit trail")
    print("=" * 60)


asyncio.run(main())