From 10406d98bb9fb360658afc3b2b71cc79732f4727 Mon Sep 17 00:00:00 2001 From: LogicNodes <164827431+cmdenney@users.noreply.github.com> Date: Thu, 28 May 2026 12:35:49 -0700 Subject: [PATCH] Add LogicNodes MCP integration example Demonstrates optional LogicNodes on-chain compliance, gas oracle, identity, and ZK attestation integration. Zero core changes. --- docs/logicnodes_example.py | 271 +++++++++++++++++++++++++++++++++++++ 1 file changed, 271 insertions(+) create mode 100644 docs/logicnodes_example.py diff --git a/docs/logicnodes_example.py b/docs/logicnodes_example.py new file mode 100644 index 0000000000..427dcfbef8 --- /dev/null +++ b/docs/logicnodes_example.py @@ -0,0 +1,271 @@ +""" +LogicNodes Integration for PydanticAI +====================================== +Demonstrates how to register LogicNodes deterministic compute workers as +tools in a PydanticAI agent. LogicNodes provides 2,300+ cryptographically- +signed microservices: gas oracles, compliance sentries, identity verification, +ZK attestation, DeFi data, and more. + +Install: + pip install pydantic-ai requests + +Usage: + export OPENAI_API_KEY="sk-..." + export LOGICNODES_API_KEY="your_key_from_https://logicnodes.io/checkout" + python docs/logicnodes_example.py +""" + +import asyncio +import os +from typing import Any + +import httpx +from pydantic import BaseModel +from pydantic_ai import Agent, RunContext, Tool + +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 {} + + +# --------------------------------------------------------------------------- +# Pydantic models for structured LogicNodes responses +# --------------------------------------------------------------------------- + +class GasEstimate(BaseModel): + chain: str + base_fee_gwei: float | None = None + priority_fee_gwei: float | None = None + max_fee_gwei: float | None = None + signature: str | None = None + raw: dict = {} + + +class ComplianceResult(BaseModel): + agent_id: str + action: str + permitted: bool | None = None + reason: str | None = None + attestation_hash: str | None = None + raw: dict = {} + + +class EthPrice(BaseModel): + price_usd: float | None = None + timestamp: int | None = None + signature: str | None = None + raw: dict = {} + + +class ZkAttestation(BaseModel): + content_hash: str | None = None + tx_hash: str | None = None + chain: str | None = None + proof_url: str | None = None + raw: dict = {} + + +class GraphScore(BaseModel): + agent_id: str + score: float | None = None + risk_tier: str | None = None + raw: dict = {} + + +# --------------------------------------------------------------------------- +# LogicNodes tool implementations +# --------------------------------------------------------------------------- + +async def gas_oracle(ctx: RunContext[None], chain: str = "ethereum") -> GasEstimate: + """ + Query the LogicNodes gas oracle for deterministic EIP-1559 gas estimates. + + Returns cryptographically-signed gas data for the specified chain. + Supported chains: ethereum, base, polygon, arbitrum, optimism. + """ + async with httpx.AsyncClient() as client: + resp = await client.post( + f"{LOGICNODES_BASE}/call/gas-oracle", + json={"chain": chain}, + headers=_ln_headers(), + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + return GasEstimate( + chain=chain, + base_fee_gwei=data.get("base_fee_gwei"), + priority_fee_gwei=data.get("priority_fee_gwei"), + max_fee_gwei=data.get("max_fee_gwei"), + signature=data.get("signature"), + raw=data, + ) + + +async def compliance_sentry( + ctx: RunContext[None], agent_id: str, action: str, context: str = "" +) -> ComplianceResult: + """ + Run an on-chain compliance check for an autonomous agent action. + + Returns a verifiable attestation indicating whether the action is + permitted under the current regulatory and constitutional ruleset + anchored by LogicNodes. + + Args: + agent_id: Unique identifier (wallet address or DID) for the agent. + action: Description of the action to compliance-check. + context: Optional JSON string providing additional context. + """ + async with httpx.AsyncClient() as client: + resp = await client.post( + f"{LOGICNODES_BASE}/call/compliance-sentry", + json={"agent_id": agent_id, "action": action, "context": context}, + headers=_ln_headers(), + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + return ComplianceResult( + agent_id=agent_id, + action=action, + permitted=data.get("permitted"), + reason=data.get("reason"), + attestation_hash=data.get("attestation_hash"), + raw=data, + ) + + +async def eth_price(ctx: RunContext[None]) -> EthPrice: + """ + Fetch the current ETH/USD price from LogicNodes. + Output is cryptographically signed — suitable for on-chain price verification. + """ + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{LOGICNODES_BASE}/call/eth-price", + headers=_ln_headers(), + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + return EthPrice( + price_usd=data.get("price_usd"), + timestamp=data.get("timestamp"), + signature=data.get("signature"), + raw=data, + ) + + +async def zk_attest(ctx: RunContext[None], content: str) -> ZkAttestation: + """ + Anchor content on-chain via LogicNodes ZK attestation. + + Creates a verifiable proof-of-existence anchored to Base L2 via USDC x402. + Useful for audit trails, decision logs, and compliance evidence. + + Args: + content: Text or JSON string to anchor on-chain. + """ + async with httpx.AsyncClient() as client: + resp = await client.post( + f"{LOGICNODES_BASE}/x402/zk-attest", + json={"content": content}, + headers=_ln_headers(), + timeout=20, + ) + resp.raise_for_status() + data = resp.json() + return ZkAttestation( + content_hash=data.get("content_hash"), + tx_hash=data.get("tx_hash"), + chain=data.get("chain"), + proof_url=data.get("proof_url"), + raw=data, + ) + + +async def graph_score(ctx: RunContext[None], agent_id: str) -> GraphScore: + """ + Retrieve the LogicNodes trust graph score for an agent. + + Returns a reputation score based on on-chain interaction history, + dispute records, and attestation volume. + + Args: + agent_id: Agent wallet address or DID. + """ + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{LOGICNODES_BASE}/graph/score/{agent_id}", + headers=_ln_headers(), + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + return GraphScore( + agent_id=agent_id, + score=data.get("score"), + risk_tier=data.get("risk_tier"), + raw=data, + ) + + +# --------------------------------------------------------------------------- +# Build the PydanticAI agent +# --------------------------------------------------------------------------- + +logicnodes_agent = Agent( + model="openai:gpt-4o", + system_prompt=( + "You are an autonomous on-chain agent powered by LogicNodes deterministic " + "compute infrastructure. Your responsibilities:\n" + "1. Always call compliance_sentry before recommending any on-chain action.\n" + "2. Use gas_oracle to provide accurate transaction cost estimates.\n" + "3. Use eth_price for current ETH valuation.\n" + "4. Anchor critical decisions and audit trails with zk_attest.\n" + "5. Check graph_score to assess counterparty reputation before interaction.\n\n" + "All LogicNodes responses are cryptographically signed and verifiable on Base L2." + ), + tools=[ + Tool(gas_oracle, takes_ctx=True), + Tool(compliance_sentry, takes_ctx=True), + Tool(eth_price, takes_ctx=True), + Tool(zk_attest, takes_ctx=True), + Tool(graph_score, takes_ctx=True), + ], +) + + +# --------------------------------------------------------------------------- +# Main demo +# --------------------------------------------------------------------------- + +async def main(): + print("=== LogicNodes + PydanticAI Integration Demo ===\n") + + result = await logicnodes_agent.run( + "I want to send 0.5 ETH to counterparty 0xDEADBEEF. " + "Please: (1) get the current ETH price, " + "(2) get a gas estimate for Ethereum, " + "(3) check compliance for agent 'agent-demo' performing 'send 0.5 ETH to 0xDEADBEEF', " + "(4) check the graph score for 0xDEADBEEF. " + "Based on these results, should I proceed?" + ) + + print("Agent response:\n") + print(result.data) + + print("\n--- Conversation history ---") + for msg in result.all_messages(): + print(f"[{msg.kind}]: {str(msg)[:120]}") + + +if __name__ == "__main__": + asyncio.run(main())