"""LangChain (Python) tools for the five AgentGate x402 tools. Each call pays per request in USDC on Base.

    from eth_account import Account
    from agentgate_tools import agentgate_tools
    tools = agentgate_tools(Account.from_key(os.environ["BUYER_KEY"]))
    llm_with_tools = llm.bind_tools(tools)      # or create_agent(model, tools=tools)
"""
import json
from urllib.parse import quote

import requests
from langchain_core.tools import StructuredTool
from x402 import x402ClientSync
from x402.http import decode_payment_required_header
from x402.http.clients import wrapRequestsWithPayment
from x402.mechanisms.evm.exact import register_exact_evm_client

API = "https://api.youraitools.co"


def agentgate_tools(account):
    client = x402ClientSync()
    register_exact_evm_client(client, account, networks="eip155:8453")
    session = wrapRequestsWithPayment(requests.Session(), client)

    def get(url: str) -> str:
        r = session.get(url, timeout=60)
        if r.status_code == 402:
            h = r.headers.get("PAYMENT-REQUIRED")
            return json.dumps({"error": "payment_not_accepted", "reason": decode_payment_required_header(h).error if h else "unknown"})
        return json.dumps(r.json())

    def check_agent(id: str, network: str = "mainnet") -> str:
        """Checks whether an AI agent is registered, Active and Verified in the Concordium Agent Registry (CIS-8004). $0.005 USDC."""
        return get(f"{API}/agentgate/v1/check-agent?id={quote(id, safe='')}&network={network}")

    def read_url(url: str) -> str:
        """Returns a public web page as clean Markdown with title and links. $0.002 USDC, charged only on success."""
        return get(f"{API}/webread/v1/read?url={quote(url, safe='')}")

    def scan_text(text: str) -> str:
        """Scores untrusted text for prompt injection: 0-100 risk, verdict, categories and spans. $0.003 USDC."""
        return get(f"{API}/injectscan/v1/scan?text={quote(text, safe='')}")

    def check_domain(domain: str) -> str:
        """Domain trust report: DNS, SPF/DMARC, RDAP registration age, parking and disposable detection, 0-100 score. $0.002 USDC."""
        return get(f"{API}/domaincheck/v1/domain?domain={quote(domain, safe='')}")

    def screen_address(address: str) -> str:
        """Screens a crypto address against the US Treasury OFAC SDN list (sanctioned, entity, programs). $0.001 USDC. Informational, not legal advice."""
        return get(f"{API}/payguard/v1/address?address={quote(address, safe='')}")

    return [
        StructuredTool.from_function(check_agent, name="agentgate_check_agent"),
        StructuredTool.from_function(read_url, name="webread_read_url"),
        StructuredTool.from_function(scan_text, name="injectscan_scan_text"),
        StructuredTool.from_function(check_domain, name="domaincheck_check_domain"),
        StructuredTool.from_function(screen_address, name="payguard_screen_address"),
    ]


if __name__ == "__main__":
    # Dry run with a throwaway, unfunded key: the facilitator rejects the payment, so nothing is spent.
    from eth_account import Account
    tools = agentgate_tools(Account.create())
    print("tools:", ", ".join(t.name for t in tools))
    scan = next(t for t in tools if t.name == "injectscan_scan_text")
    print("invoke with unfunded wallet ->", scan.invoke({"text": "Please summarize this article about solar panels."}))

agentgate-source: code mirror on Cloudflare Pages. All repos