AI Agent Integration

import anthropic

client = anthropic.Anthropic()

# Register LimitGuard as an MCP server
# Claude will call check_entity_trust automatically when it needs to verify an entity
response = client.beta.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    mcp_servers=[
        {
            "type": "url",
            "url": "https://api.limitguard.ai/mcp",
            "name": "limitguard",
            "authorization_token": "lg_live_xxxxxxxxxxxxxxxxxxxx",
        }
    ],
    messages=[
        {
            "role": "user",
            "content": (
                "I need to pay invoice #INV-2024-447 from Acme Corp BV "
                "(KVK: 12345678, NL). Should I proceed?"
            ),
        }
    ],
    betas=["mcp-client-2025-04-04"],
)

print(response.content)
# Claude calls check_entity_trust internally, interprets the score,
# and returns a recommendation with evidence.

import httpx

class LimitGuardA2AClient:
    """A2A client for LimitGuard trust verification."""

    AGENT_CARD_URL = "https://api.limitguard.ai/.well-known/agent.json"
    BASE_URL = "https://api.limitguard.ai"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self._card = None

    def discover(self) -> dict:
        """Fetch and cache the agent card."""
        if self._card is None:
            self._card = httpx.get(self.AGENT_CARD_URL).json()
        return self._card

    def get_skill(self, skill_id: str) -> dict | None:
        """Look up a skill by ID from the agent card."""
        card = self.discover()
        return next(
            (s for s in card["skills"] if s["id"] == skill_id),
            None,
        )

    def invoke(self, skill_id: str, payload: dict) -> dict:
        """Invoke a LimitGuard skill via A2A."""
        skill = self.get_skill(skill_id)
        if skill is None:
            raise ValueError(f"Skill '{skill_id}' not found in agent card")

        # Map skill IDs to LimitGuard endpoints
        endpoint_map = {
            "entity_trust_check": "/v1/entity/check",
            "risk_score": "/v1/risk/score",
            "wallet_screening": "/v1/wallet/screen",
        }

        endpoint = endpoint_map.get(skill_id)
        if endpoint is None:
            raise ValueError(f"No endpoint mapped for skill '{skill_id}'")

        response = httpx.post(
            f"{self.BASE_URL}{endpoint}",
            headers={"X-API-Key": self.api_key, "Content-Type": "application/json"},
            json=payload,
            timeout=10.0,
        )
        response.raise_for_status()
        return response.json()


# Usage in an orchestrator agent
client = LimitGuardA2AClient(api_key="lg_live_xxxxxxxxxxxxxxxxxxxx")

# Discover what LimitGuard can do
card = client.discover()
print(f"Available skills: {[s['id'] for s in card['skills']]}")
# Available skills: ['entity_trust_check', 'risk_score', 'wallet_screening']

# Invoke entity trust check
result = client.invoke(
    "entity_trust_check",
    {"entity_name": "Acme Corp BV", "country": "NL", "kvk_number": "12345678"},
)
print(f"Trust score: {result['trust_score']} — {result['recommendation']}")

import base64
import json
import secrets
import time
import httpx
from eth_account import Account
from eth_account.messages import encode_typed_data

USDC_BASE = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
CHAIN_ID = "eip155:8453"

def build_x402_payment(
    amount_usdc: int,    # in 6-decimal units (50000 = $0.05)
    sender_key: str,     # agent wallet private key
    recipient: str,      # facilitator address from 402 response
) -> str:
    sender = Account.from_key(sender_key).address
    chain_num = int(CHAIN_ID.split(":")[1])  # 8453
    nonce = "0x" + secrets.token_hex(32)
    now = int(time.time())

    domain = {
        "name": "USD Coin",
        "version": "2",
        "chainId": chain_num,
        "verifyingContract": USDC_BASE,
    }
    types = {
        "EIP712Domain": [
            {"name": "name", "type": "string"},
            {"name": "version", "type": "string"},
            {"name": "chainId", "type": "uint256"},
            {"name": "verifyingContract", "type": "address"},
        ],
        "TransferWithAuthorization": [
            {"name": "from", "type": "address"},
            {"name": "to", "type": "address"},
            {"name": "value", "type": "uint256"},
            {"name": "validAfter", "type": "uint256"},
            {"name": "validBefore", "type": "uint256"},
            {"name": "nonce", "type": "bytes32"},
        ],
    }
    message = {
        "from": sender,
        "to": recipient,
        "value": amount_usdc,
        "validAfter": now - 10,
        "validBefore": now + 300,
        "nonce": bytes.fromhex(nonce[2:]),
    }

    signable = encode_typed_data(domain, types, "TransferWithAuthorization", message)
    signed = Account.sign_message(signable, private_key=sender_key)

    payload = {
        "chainId": CHAIN_ID,
        "amount": str(amount_usdc),
        "sender": sender,
        "recipient": recipient,
        "nonce": nonce,
        "signature": signed.signature.hex(),
        "validAfter": now - 10,
        "validBefore": now + 300,
    }
    return base64.b64encode(json.dumps(payload).encode()).decode()


def verify_entity_x402(
    entity_name: str,
    country: str,
    sender_key: str,
    quality: str = "fresh",  # cached | fresh | enhanced
) -> dict:
    tier_prices = {"cached": 10_000, "fresh": 50_000, "enhanced": 150_000}
    amount = tier_prices[quality]

    # Step 1: Probe to get facilitator address
    probe = httpx.post(
        "https://api.limitguard.ai/v1/entity/check",
        headers={"X-Response-Quality": quality, "Content-Type": "application/json"},
        json={"entity_name": entity_name, "country": country},
    )
    assert probe.status_code == 402, f"Expected 402, got {probe.status_code}"
    facilitator = probe.json()["accepts"][0]["recipient"]

    # Step 2: Build payment and call
    x_payment = build_x402_payment(amount, sender_key, facilitator)

    response = httpx.post(
        "https://api.limitguard.ai/v1/entity/check",
        headers={
            "X-PAYMENT": x_payment,
            "X-Response-Quality": quality,
            "Content-Type": "application/json",
        },
        json={"entity_name": entity_name, "country": country},
        timeout=15.0,
    )
    response.raise_for_status()
    return response.json()


# Usage
result = verify_entity_x402(
    entity_name="Acme Corp BV",
    country="NL",
    sender_key="0xYOUR_AGENT_WALLET_PRIVATE_KEY",
    quality="fresh",
)
print(f"Trust score: {result['trust_score']} — {result['recommendation']}")

import base64
import json
import os
import secrets
import time
from dataclasses import dataclass
from enum import Enum

import httpx
from eth_account import Account
from eth_account.messages import encode_typed_data


class TrustDecision(Enum):
    PROCEED = "proceed"
    REVIEW = "review"
    ENHANCED_DUE_DILIGENCE = "enhanced_due_diligence"
    BLOCK = "block"


@dataclass
class TrustResult:
    score: int
    level: str
    decision: TrustDecision
    cluster: str
    sources_checked: list[str]
    processing_ms: int
    paid_usdc: float


class TrustGatedAgent:
    """
    An autonomous payment agent that verifies counterparty trust
    before executing transactions — no human in the loop.
    """

    DISCOVERY_URL = "https://api.limitguard.ai/.well-known/x402.json"
    BASE_URL = "https://api.limitguard.ai"
    USDC_BASE = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
    CHAIN_ID = "eip155:8453"

    def __init__(self, wallet_private_key: str):
        self.wallet = Account.from_key(wallet_private_key)
        self._discovery_cache: dict | None = None

    # ------------------------------------------------------------------ #
    # Discovery                                                            #
    # ------------------------------------------------------------------ #

    def discover(self) -> dict:
        """Fetch and cache the x402 service listing."""
        if self._discovery_cache is None:
            response = httpx.get(self.DISCOVERY_URL, timeout=5.0)
            response.raise_for_status()
            self._discovery_cache = response.json()
        return self._discovery_cache

    def get_price(self, endpoint: str, quality: str = "fresh") -> int:
        """Return price in USDC 6-decimal units for an endpoint + quality tier."""
        listing = self.discover()
        entry = next(
            (e for e in listing["endpoints"] if e["path"] == endpoint), None
        )
        if entry is None:
            raise ValueError(f"Endpoint {endpoint} not in discovery listing")
        tiers = entry.get("quality_tiers", {})
        price_str = tiers.get(quality) or entry.get("price_usdc", "0.05")
        return int(float(price_str) * 1_000_000)

    # ------------------------------------------------------------------ #
    # Payment                                                              #
    # ------------------------------------------------------------------ #

    def _get_facilitator(self, endpoint: str, payload: dict) -> str:
        """Probe an endpoint to get the current facilitator address."""
        probe = httpx.post(
            f"{self.BASE_URL}{endpoint}",
            headers={"Content-Type": "application/json"},
            json=payload,
            timeout=10.0,
        )
        if probe.status_code != 402:
            raise RuntimeError(f"Expected 402 probe, got {probe.status_code}")
        accepts = probe.json()["accepts"]
        # Prefer Base chain
        for option in accepts:
            if option["chainId"] == self.CHAIN_ID:
                return option["recipient"]
        return accepts[0]["recipient"]

    def _build_payment(self, amount: int, recipient: str) -> str:
        """Build an EIP-3009 x402 payment header."""
        chain_num = int(self.CHAIN_ID.split(":")[1])
        nonce = "0x" + secrets.token_hex(32)
        now = int(time.time())

        domain = {
            "name": "USD Coin",
            "version": "2",
            "chainId": chain_num,
            "verifyingContract": self.USDC_BASE,
        }
        types = {
            "EIP712Domain": [
                {"name": "name", "type": "string"},
                {"name": "version", "type": "string"},
                {"name": "chainId", "type": "uint256"},
                {"name": "verifyingContract", "type": "address"},
            ],
            "TransferWithAuthorization": [
                {"name": "from", "type": "address"},
                {"name": "to", "type": "address"},
                {"name": "value", "type": "uint256"},
                {"name": "validAfter", "type": "uint256"},
                {"name": "validBefore", "type": "uint256"},
                {"name": "nonce", "type": "bytes32"},
            ],
        }
        message = {
            "from": self.wallet.address,
            "to": recipient,
            "value": amount,
            "validAfter": now - 10,
            "validBefore": now + 300,
            "nonce": bytes.fromhex(nonce[2:]),
        }

        signable = encode_typed_data(
            domain, types, "TransferWithAuthorization", message
        )
        signed = Account.sign_message(signable, private_key=self.wallet.key)

        payload = {
            "chainId": self.CHAIN_ID,
            "amount": str(amount),
            "sender": self.wallet.address,
            "recipient": recipient,
            "nonce": nonce,
            "signature": signed.signature.hex(),
            "validAfter": now - 10,
            "validBefore": now + 300,
        }
        return base64.b64encode(json.dumps(payload).encode()).decode()

    def _call(self, endpoint: str, payload: dict, quality: str) -> dict:
        """Make a paid API call."""
        amount = self.get_price(endpoint, quality)
        facilitator = self._get_facilitator(endpoint, payload)
        x_payment = self._build_payment(amount, facilitator)

        response = httpx.post(
            f"{self.BASE_URL}{endpoint}",
            headers={
                "X-PAYMENT": x_payment,
                "X-Response-Quality": quality,
                "Content-Type": "application/json",
            },
            json=payload,
            timeout=15.0,
        )
        response.raise_for_status()
        return response.json()

    # ------------------------------------------------------------------ #
    # Trust verification logic                                             #
    # ------------------------------------------------------------------ #

    def _score_to_decision(self, score: int) -> TrustDecision:
        if score >= 80:
            return TrustDecision.PROCEED
        elif score >= 60:
            return TrustDecision.REVIEW
        elif score >= 40:
            return TrustDecision.ENHANCED_DUE_DILIGENCE
        else:
            return TrustDecision.BLOCK

    def verify(self, entity_name: str, country: str, **kwargs) -> TrustResult:
        """
        Verify an entity using a two-pass strategy:
        1. Cheap risk score pre-screen
        2. Full entity check only if pre-screen passes
        """
        payload = {"entity_name": entity_name, "country": country, **kwargs}
        total_paid = 0.0

        # Pass 1: Fast pre-screen ($0.01)
        risk_data = self._call("/v1/risk/score", payload, quality="cached")
        total_paid += 0.01
        pre_score = risk_data.get("risk_score", 50)

        if pre_score > 80:
            # High risk — block immediately without paying for full check
            return TrustResult(
                score=100 - pre_score,
                level="high_risk",
                decision=TrustDecision.BLOCK,
                cluster="unknown",
                sources_checked=risk_data.get("sources_checked", []),
                processing_ms=risk_data.get("processing_time_ms", 0),
                paid_usdc=total_paid,
            )

        # Pass 2: Full entity check ($0.05 fresh)
        trust_data = self._call("/v1/entity/check", payload, quality="fresh")
        total_paid += 0.05

        trust_score = trust_data["trust_score"]
        decision = self._score_to_decision(trust_score)

        return TrustResult(
            score=trust_score,
            level=trust_data["trust_level"],
            decision=decision,
            cluster=trust_data.get("cluster", "unknown"),
            sources_checked=trust_data.get("sources_checked", []),
            processing_ms=trust_data.get("processing_time_ms", 0),
            paid_usdc=total_paid,
        )

    def should_pay_invoice(
        self,
        invoice_id: str,
        supplier_name: str,
        country: str,
        amount_eur: float,
        **entity_kwargs,
    ) -> tuple[bool, TrustResult]:
        """
        Top-level decision: should this invoice be paid autonomously?
        Returns (approved, trust_result).
        """
        result = self.verify(supplier_name, country, **entity_kwargs)

        approved = result.decision in (TrustDecision.PROCEED, TrustDecision.REVIEW)

        print(
            f"Invoice {invoice_id} | Supplier: {supplier_name} | "
            f"Amount: €{amount_eur:.2f} | "
            f"Trust: {result.score}/100 ({result.decision.value}) | "
            f"Cost: ${result.paid_usdc:.3f} USDC"
        )

        return approved, result


# ------------------------------------------------------------------ #
# Usage                                                                #
# ------------------------------------------------------------------ #

agent = TrustGatedAgent(wallet_private_key=os.environ["AGENT_WALLET_KEY"])

approved, trust = agent.should_pay_invoice(
    invoice_id="INV-2024-447",
    supplier_name="Acme Corp BV",
    country="NL",
    amount_eur=500.00,
    kvk_number="12345678",
    domain="acme-corp.nl",
)

if approved:
    print(f"Proceeding with payment. Cluster: {trust.cluster}")
    # execute_payment(...)
else:
    print(f"Payment blocked. Score: {trust.score}. Alerting operator.")
    # alert_operator(...)

import httpx

def load_limitguard_context(full: bool = False) -> str:
    """Load LimitGuard API context for injection into an LLM prompt."""
    url = "https://api.limitguard.ai/llms-full.txt" if full else "https://api.limitguard.ai/llms.txt"
    response = httpx.get(url, timeout=5.0)
    response.raise_for_status()
    return response.text


# Inject into a system prompt for a coding assistant
context = load_limitguard_context(full=True)

system_prompt = f"""You are a payment automation assistant.
You have access to the LimitGuard Trust Intelligence API.

{context}

Before authorizing any payment, call the entity check endpoint.
Block all transactions where recommendation is 'block'.
"""

import httpx

def verify_entity_sandbox(entity_name: str, country: str, api_key: str) -> dict:
    """Sandbox mode — deterministic mock data, no payment, no quota."""
    response = httpx.post(
        "https://api.limitguard.ai/v1/entity/check",
        headers={
            "X-API-Key": api_key,
            "X-LimitGuard-Mode": "sandbox",
            "Content-Type": "application/json",
        },
        json={"entity_name": entity_name, "country": country},
        timeout=5.0,
    )
    response.raise_for_status()
    return response.json()

# Sandbox response is deterministic — useful for unit tests
result = verify_entity_sandbox("Any Company", "NL", api_key="lg_live_xxxxxxxxxxxxxxxxxxxx")
# Always returns: trust_score: 75, recommendation: "review"