From ae0cc4ed470980d6d06d054f072b6d0806d3ee7b Mon Sep 17 00:00:00 2001 From: LogicNodes <164827431+cmdenney@users.noreply.github.com> Date: Thu, 28 May 2026 12:36:05 -0700 Subject: [PATCH] Add LogicNodes MCP integration example Demonstrates optional LogicNodes on-chain compliance, gas oracle, identity, and ZK attestation integration. Zero core changes. --- python/examples/logicnodes_integration.py | 327 ++++++++++++++++++++++ 1 file changed, 327 insertions(+) create mode 100644 python/examples/logicnodes_integration.py diff --git a/python/examples/logicnodes_integration.py b/python/examples/logicnodes_integration.py new file mode 100644 index 000000000..2554ddf54 --- /dev/null +++ b/python/examples/logicnodes_integration.py @@ -0,0 +1,327 @@ +""" +LogicNodes On-Chain Compliance Integration for Coinbase AgentKit +================================================================= +Demonstrates how to integrate LogicNodes deterministic compute workers into +a Coinbase AgentKit agent for on-chain compliance, gas estimation, identity +verification, and ZK attestation. + +LogicNodes provides 2,300+ cryptographically-signed microservices callable +via REST or MCP — zero subscriptions, pay-per-call via USDC x402 on Base. + +Install: + pip install coinbase-agentkit requests + +Usage: + export CDP_API_KEY_NAME="..." + export CDP_API_KEY_PRIVATE_KEY="..." + export OPENAI_API_KEY="sk-..." + export LOGICNODES_API_KEY="your_key_from_https://logicnodes.io/checkout" + python python/examples/logicnodes_integration.py +""" + +import asyncio +import os +from typing import Any + +import requests +from coinbase_agentkit import ( + AgentKit, + AgentKitConfig, + CdpWalletProvider, + CdpWalletProviderConfig, +) +from coinbase_agentkit_langchain import get_langchain_tools +from langchain_openai import ChatOpenAI +from langgraph.prebuilt import create_react_agent + +LOGICNODES_API_KEY = os.environ.get("LOGICNODES_API_KEY", "") +LOGICNODES_BASE = "https://logicnodes.io" + + +def _ln_headers() -> dict: + """Return authorization headers for LogicNodes API.""" + if LOGICNODES_API_KEY: + return {"Authorization": f"Bearer {LOGICNODES_API_KEY}"} + return {} + + +def _call_worker(worker_name: str, payload: dict | None = None) -> dict: + """Generic LogicNodes worker call via REST.""" + if payload: + resp = requests.post( + f"{LOGICNODES_BASE}/call/{worker_name}", + json=payload, + headers=_ln_headers(), + timeout=15, + ) + else: + resp = requests.get( + f"{LOGICNODES_BASE}/call/{worker_name}", + headers=_ln_headers(), + timeout=15, + ) + resp.raise_for_status() + return resp.json() + + +# --------------------------------------------------------------------------- +# LogicNodes tool functions (compatible with LangChain/AgentKit tool format) +# --------------------------------------------------------------------------- + +from langchain.tools import StructuredTool +from pydantic import BaseModel, Field + + +class GasOracleInput(BaseModel): + chain: str = Field( + default="base", + description="Chain name: base, ethereum, polygon, arbitrum", + ) + + +class ComplianceSentryInput(BaseModel): + agent_id: str = Field(description="Agent wallet address or DID.") + action: str = Field(description="Description of the action to check.") + context: str = Field(default="", description="Optional JSON context string.") + + +class EthPriceInput(BaseModel): + pass + + +class ZkAttestInput(BaseModel): + content: str = Field(description="Text or JSON content to anchor on-chain.") + + +class GraphScoreInput(BaseModel): + agent_id: str = Field(description="Agent wallet address or DID.") + + +def gas_oracle(chain: str = "base") -> str: + """ + Query the LogicNodes gas oracle for deterministic EIP-1559 gas estimates. + Returns a cryptographically-signed payload suitable for on-chain verification. + """ + try: + data = _call_worker("gas-oracle", {"chain": chain}) + return ( + f"Gas oracle ({chain}): " + f"base_fee={data.get('base_fee_gwei', 'N/A')} gwei, " + f"priority_fee={data.get('priority_fee_gwei', 'N/A')} gwei, " + f"max_fee={data.get('max_fee_gwei', 'N/A')} gwei. " + f"Signature: {data.get('signature', 'N/A')[:16]}..." + ) + except Exception as e: + return f"Gas oracle error: {e}" + + +def compliance_sentry(agent_id: str, action: str, context: str = "") -> str: + """ + Run an on-chain compliance check for an autonomous agent action via LogicNodes. + Returns a verifiable attestation of whether the action is permitted. + """ + try: + data = _call_worker( + "compliance-sentry", + {"agent_id": agent_id, "action": action, "context": context}, + ) + permitted = data.get("permitted", "unknown") + reason = data.get("reason", "") + attestation = data.get("attestation_hash", "")[:16] + return ( + f"Compliance check for '{action}' by {agent_id}: " + f"permitted={permitted}. Reason: {reason}. " + f"Attestation: {attestation}..." + ) + except Exception as e: + return f"Compliance sentry error: {e}" + + +def eth_price_logicnodes() -> str: + """ + Fetch the current ETH/USD price from LogicNodes (cryptographically signed). + Suitable for on-chain price verification and autonomous agent decisions. + """ + try: + resp = requests.get( + f"{LOGICNODES_BASE}/call/eth-price", + headers=_ln_headers(), + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + return ( + f"ETH price: ${data.get('price_usd', 'N/A')} USD " + f"(timestamp: {data.get('timestamp', 'N/A')}, " + f"signature: {data.get('signature', 'N/A')[:16]}...)" + ) + except Exception as e: + return f"ETH price error: {e}" + + +def zk_attest_content(content: str) -> str: + """ + Anchor content on-chain via LogicNodes ZK attestation (Base L2, USDC x402). + Returns a verifiable proof-of-existence for audit trails and compliance. + """ + try: + resp = requests.post( + f"{LOGICNODES_BASE}/x402/zk-attest", + json={"content": content}, + headers=_ln_headers(), + timeout=20, + ) + resp.raise_for_status() + data = resp.json() + return ( + f"ZK attestation created: " + f"content_hash={data.get('content_hash', 'N/A')[:24]}..., " + f"tx_hash={data.get('tx_hash', 'N/A')[:24]}..., " + f"proof_url={data.get('proof_url', 'N/A')}" + ) + except Exception as e: + return f"ZK attestation error: {e}" + + +def graph_score_lookup(agent_id: str) -> str: + """ + Retrieve the LogicNodes trust graph score for an agent. + Returns reputation score and risk tier based on on-chain history. + """ + try: + resp = requests.get( + f"{LOGICNODES_BASE}/graph/score/{agent_id}", + headers=_ln_headers(), + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + return ( + f"Trust graph score for {agent_id}: " + f"score={data.get('score', 'N/A')}, " + f"risk_tier={data.get('risk_tier', 'N/A')}" + ) + except Exception as e: + return f"Graph score error: {e}" + + +# Build LangChain StructuredTools +LOGICNODES_LANGCHAIN_TOOLS = [ + StructuredTool.from_function( + func=gas_oracle, + name="logicnodes_gas_oracle", + description=( + "Query LogicNodes gas oracle for deterministic EIP-1559 gas estimates " + "on Base, Ethereum, Polygon, or Arbitrum. Output is cryptographically signed." + ), + args_schema=GasOracleInput, + ), + StructuredTool.from_function( + func=compliance_sentry, + name="logicnodes_compliance_sentry", + description=( + "Run on-chain compliance check for an agent action via LogicNodes. " + "Returns verifiable attestation of whether action is permitted." + ), + args_schema=ComplianceSentryInput, + ), + StructuredTool.from_function( + func=eth_price_logicnodes, + name="logicnodes_eth_price", + description=( + "Fetch the current ETH/USD price from LogicNodes (cryptographically signed). " + "Use for on-chain price decisions." + ), + args_schema=EthPriceInput, + ), + StructuredTool.from_function( + func=zk_attest_content, + name="logicnodes_zk_attest", + description=( + "Anchor content on-chain via LogicNodes ZK attestation on Base L2. " + "Returns verifiable proof-of-existence for audit trails." + ), + args_schema=ZkAttestInput, + ), + StructuredTool.from_function( + func=graph_score_lookup, + name="logicnodes_graph_score", + description=( + "Get the LogicNodes trust graph score for an agent's wallet address or DID. " + "Returns reputation score and risk tier." + ), + args_schema=GraphScoreInput, + ), +] + + +# --------------------------------------------------------------------------- +# Main demo: AgentKit + LogicNodes +# --------------------------------------------------------------------------- + +def build_agentkit_agent(): + """Build a Coinbase AgentKit agent augmented with LogicNodes tools.""" + # Initialize wallet provider + wallet_provider = CdpWalletProvider( + CdpWalletProviderConfig( + api_key_name=os.environ.get("CDP_API_KEY_NAME", ""), + api_key_private_key=os.environ.get("CDP_API_KEY_PRIVATE_KEY", ""), + ) + ) + + # Initialize AgentKit + agentkit = AgentKit( + AgentKitConfig( + wallet_provider=wallet_provider, + ) + ) + + # Combine AgentKit tools with LogicNodes tools + agentkit_tools = get_langchain_tools(agentkit) + all_tools = agentkit_tools + LOGICNODES_LANGCHAIN_TOOLS + + # Build LangGraph ReAct agent + llm = ChatOpenAI(model="gpt-4o") + agent = create_react_agent( + llm, + all_tools, + state_modifier=( + "You are an autonomous on-chain agent powered by Coinbase AgentKit " + "and LogicNodes deterministic compute. Before executing any transaction:\n" + "1. Call logicnodes_compliance_sentry to verify the action is permitted.\n" + "2. Call logicnodes_gas_oracle to get accurate gas estimates on Base.\n" + "3. Call logicnodes_eth_price for current ETH value.\n" + "4. After execution, call logicnodes_zk_attest to anchor the decision.\n" + "All LogicNodes outputs are cryptographically signed and verifiable on-chain." + ), + ) + return agent + + +def main(): + print("=== LogicNodes + Coinbase AgentKit Integration Demo ===\n") + + agent = build_agentkit_agent() + + events = agent.stream( + { + "messages": [ + ( + "user", + "Check the current ETH price and gas estimate on Base. " + "Then verify compliance for agent 'agentkit-demo' performing " + "'transfer 0.01 ETH to 0xRecipient'. " + "Should I proceed with the transaction?", + ) + ] + }, + stream_mode="values", + ) + + for event in events: + if "messages" in event: + event["messages"][-1].pretty_print() + + +if __name__ == "__main__": + main()