""" Diogenes Action Provider for Coinbase AgentKit & Base EVM Agents. Enables autonomous agents built on CDP AgentKit to execute Diogenes microservices and settle micropayments natively over Base EVM via Machine Payments Protocol (MPP). """ import json import logging from typing import Any, Dict, Optional import urllib.request import urllib.error logger = logging.getLogger("diogenes.agentkit") DEFAULT_GATEWAY_URL = "https://diogenes-micro-gateway.corinth.workers.dev" BASE_MPP_WALLET = "0x0E97828055Ea1291e57dB32EBEfC6e7b3F7AcACb" class DiogenesActionProvider: """ Coinbase AgentKit Action Provider for Diogenes Edge Microservices. Allows agents on Base to invoke pre-execution guardrails, smart contract audits, and 121 deterministic developer utilities with native MPP settlement or demo mode. """ def __init__( self, gateway_url: str = DEFAULT_GATEWAY_URL, wallet_provider: Optional[Any] = None, demo_mode: bool = False, ): """ Initialize the Diogenes Action Provider. :param gateway_url: Base URL of the Diogenes Gateway (defaults to production edge). :param wallet_provider: Optional Coinbase CDP wallet provider (e.g. CdpWalletProvider). If provided, automatically signs and broadcasts Base EVM micropayments. :param demo_mode: If True, appends ?demo=true for zero-cost trial invocation. """ self.gateway_url = gateway_url.rstrip("/") self.wallet_provider = wallet_provider self.demo_mode = demo_mode def _call_endpoint(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: """ Call a Diogenes endpoint, handling demo mode and Base EVM MPP payment negotiation. """ url = f"{self.gateway_url}{endpoint}" if self.demo_mode and "?demo=true" not in url: url += "?demo=true" data = json.dumps(payload).encode("utf-8") headers = { "Content-Type": "application/json", "User-Agent": "Coinbase-AgentKit-Diogenes/1.0", } req = urllib.request.Request(url, data=data, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as err: if err.code == 402 and not self.demo_mode and self.wallet_provider: # HTTP 402 Payment Required: negotiate Base MPP payment err_body = json.loads(err.read().decode("utf-8")) return self._settle_and_retry(endpoint, payload, err_body, err.headers) elif err.code == 402: # Return informative error if wallet is not configured error_msg = ( "HTTP 402 Payment Required. Set demo_mode=True or provide a " "funded CDP wallet_provider on Base to settle MPP micropayments." ) return {"error": "PAYMENT_REQUIRED", "message": error_msg, "status": 402} else: try: err_json = json.loads(err.read().decode("utf-8")) return {"error": "HTTP_ERROR", "status": err.code, "details": err_json} except Exception: return {"error": "HTTP_ERROR", "status": err.code, "message": str(err)} def _settle_and_retry( self, endpoint: str, payload: Dict[str, Any], challenge_body: Dict[str, Any], headers: Any, ) -> Dict[str, Any]: """ Settle a Base EVM micropayment via the attached wallet provider and retry request. """ recipient = challenge_body.get("recipient") or BASE_MPP_WALLET amount_eth = challenge_body.get("amount_eth", 0.000005) logger.info( f"Settling {amount_eth} ETH on Base to {recipient} via Coinbase AgentKit..." ) try: # Broadcast native ETH transfer on Base using CDP Wallet Provider tx = self.wallet_provider.transfer( amount=amount_eth, asset_id="eth", destination=recipient, ) tx.wait() tx_hash = tx.transaction_hash # Retry request with MPP proof headers retry_url = f"{self.gateway_url}{endpoint}" data = json.dumps(payload).encode("utf-8") retry_headers = { "Content-Type": "application/json", "Payment-Authorization": tx_hash, "X-Payment-TxHash": tx_hash, "Authorization": f"Bearer {tx_hash}", "User-Agent": "Coinbase-AgentKit-Diogenes/1.0", } req = urllib.request.Request( retry_url, data=data, headers=retry_headers, method="POST" ) with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode("utf-8")) except Exception as e: logger.error(f"Failed to settle Base MPP payment: {e}") return {"error": "SETTLEMENT_FAILED", "message": str(e)} # High-Value Agent Actions def audit_smart_contract( self, contract_code: str, language: str = "solidity" ) -> Dict[str, Any]: """ Audit Solidity or Anchor smart contract code before deployment or interaction. Detects honeypots, reentrancy vulnerabilities, hidden owner mint backdoors, and transfer taxes. :param contract_code: Source code of the contract. :param language: 'solidity' or 'rust_anchor'. :return: Audit report with vulnerability scores and recommendations. """ return self._call_endpoint( "/api/v1/agent/contract-audit", {"contract_code": contract_code, "language": language}, ) def verify_agent_guardrail( self, command_or_prompt: str, context: str = "" ) -> Dict[str, Any]: """ Verify safety of an agent prompt, shell command, or SQL query before execution. Prevents prompt injection, credential exfiltration, and destructive file system operations. :param command_or_prompt: The text, prompt, or command to evaluate. :param context: Optional operational context for risk scoring. :return: Verdict dict with 'safe', 'risk_score', and 'detected_threats'. """ return self._call_endpoint( "/api/v1/agent/guardrail-verify", {"input": command_or_prompt, "context": context}, ) def plan_token_budget( self, target_model: str, system_prompt: str, user_prompt: str ) -> Dict[str, Any]: """ Analyze prompt context and calculate optimal token allocation for LLM inference. :param target_model: Model name (e.g. 'claude-3-5-sonnet', 'gemini-2.5-pro', 'gpt-4o'). :param system_prompt: System prompt text. :param user_prompt: User prompt text. :return: Budget breakdown with token counts and headroom analysis. """ return self._call_endpoint( "/api/v1/context/token-budget-planner", { "target_model": target_model, "system_prompt": system_prompt, "user_prompt": user_prompt, }, ) def execute_microservice( self, endpoint: str, payload: Dict[str, Any] ) -> Dict[str, Any]: """ Universally invoke any of the 121 Diogenes edge microservices. :param endpoint: Route path (e.g. '/api/v1/db/nl-to-sql', '/api/v1/crypto/solana-risk'). :param payload: JSON body parameters required by the microservice. :return: Microservice result object. """ if not endpoint.startswith("/"): endpoint = "/" + endpoint return self._call_endpoint(endpoint, payload)