13. Client SDK Code Snippets: Ready-Made Integration Patterns (Python & Node.js)

November 2, 20261 min read

The application SDK fully hides the low-level complexity of assembling a UserOperation, signing EIP-712 messages and P2P routing through Envoy. An AI agent developer only needs to initialize a LagrixClient instance and register a task handler or make a call.

Below are production-ready integration patterns for the Client (Client Agent) and the Provider (Provider Agent) in Python and TypeScript (Node.js).

13.1 Python Integration (Async/Await Pattern)

In the Python ecosystem the SDK uses asyncio to serve requests between agents asynchronously and supports transparent integration with the LangChain, AutoGen and CrewAI frameworks.

1. Provider Agent — Service registration and accepting orders

import asyncio
from lagrix import LagrixNode, TaskRequest, TaskResult

async def handle_ai_inference(task: TaskRequest) -> TaskResult:
    """
    Custom AI Agent logic.
    Called automatically after the packet has passed through P2P and E2EE.
    """
    print(f"[Provider] Task received {task.id} from Client: {task.client_node_id}")
    
    # Extract the decrypted and checked prompt
    prompt = task.payload.get("prompt")
    
    # Emulate LLM work (for example, a call to a local vLLM/Ollama)
    execution_result = f"Processed context for: '{prompt}'. Liquidity confirmed."
    
    # Return the result. The SDK signs it automatically (EIP-712)
    return TaskResult(
        task_id=task.id,
        status="SUCCESS",
        data={"response": execution_result, "confidence": 0.99}
    )

async def main():
    # Initialize the Provider node with a wallet
    node = LagrixNode(
        private_key="0x4f3a...b12",  # EVM wallet for receiving USDC for work
        rpc_url="https://arb-mainnet.g.alchemy.com/v2/YOUR_KEY"
    )

    # Publish a signed announcement (Nostr, document 17): capability, availability and price
    await node.announce(
        service_name="market-analyzer-v1",
        price_per_task_usdc=0.50,  # $0.50 USDC per task
        handler=handle_ai_inference
    )

    print("[Provider] AI Agent started and listening to the Lagrix P2P network...")
    await node.listen()

if __name__ == "__main__":
    asyncio.run(main())

2. Client Agent — Task initiation and sending USDC to Escrow

import asyncio
from lagrix import LagrixClient

async def main():
    # Initialize the Client
    client = LagrixClient(
        private_key="0x8c1d...a90",  # Client wallet (ERC-4337 smart account)
        paymaster="auto"  # a third-party paymaster, for example Circle; "self" — stamps in own ETH (changelog entries 37 and 38)
    )

    # One search through a super-node ($0.01, document 17); the client remembers the provider for the next tasks
    print("[Client] Searching for a Provider...")
    provider_node = await client.find_provider(capability="market-analyzer-v1", max_price_usdc=0.70)
    
    print(f"[Client] Node found: {provider_node.node_id}. Locking $0.50 USDC in escrow...")

    # Automatically: Build UserOp -> Sign -> Paymaster (gas payment) -> A2AEscrow.sol
    task_handle = await client.execute_task(
        provider_node_id=provider_node.node_id,
        amount_usdc=0.50,
        payload={
            "prompt": "Analyze the depth of the ETH/USDC order book for large orders.",
            "temperature": 0.1
        },
        deal="instant",
        deliver_within_seconds=60,    # the delivery deadline: 60 s after the lock (changelog, entry 53)
        dispute_window_seconds=300    # the dispute window of this deal
    )

    print(f"[Client] Task locked in block {task_handle.tx_hash}. Waiting for the P2P answer...")
    
    # Wait for the asynchronous answer over the mTLS tunnel
    result = await task_handle.wait_for_completion()
    
    if result.status == "COMPLETED":
        print(f"[Client] Result received: {result.data['response']}")
        print(f"[Client] Provider's EIP-712 signature verified: {result.provider_signature}")
    else:
        print(f"[Client] Execution error: {result.error_message}")

if __name__ == "__main__":
    asyncio.run(main())

13.2 Node.js / TypeScript Integration (Enterprise Express/gRPC Pattern)

In the TypeScript environment the SDK uses the EventEmitter pattern and strict types generated from Protobuf.

1. Provider Agent

import { LagrixProviderNode, TaskRequest, TaskResponse } from '@lagrix/sdk';

const provider = new LagrixProviderNode({
  privateKey: process.env.PROVIDER_PRIVATE_KEY!,
  network: 'arbitrum-one'
});

// Listener for incoming task events
provider.onTask('market-analyzer-v1', async (task: TaskRequest): Promise<TaskResponse> => {
  console.log(`[TS-Provider] Task ${task.taskId} accepted. Processing...`);

  const { prompt } = task.getPayloadJson();

  // Call an external neural network or business logic
  const responseText = `[TS-Agent] Analysis completed for request: "${prompt}"`;

  return {
    status: 'SUCCESS',
    payload: { analysis: responseText, timestamp: Date.now() }
  };
});

async function start() {
  await provider.start();
  console.log(`[TS-Provider] Node started. NodeID: ${provider.nodeId}`);
}

start().catch(console.error);

2. Client Agent

import { LagrixClient } from '@lagrix/sdk';

async function runClient() {
  const client = new LagrixClient({
    privateKey: process.env.CLIENT_PRIVATE_KEY!,
    paymaster: 'auto' // a third-party paymaster, for example Circle; 'self' — stamps in own ETH (changelog entries 37 and 38)
  });

  try {
    console.log('[TS-Client] Sending an asynchronous task to the P2P network...');

    // One-line call: Search (only if this provider is not known yet) -> Escrow Lock -> P2P Execution -> Verification
    const response = await client.requestTaskExecution({
      serviceName: 'market-analyzer-v1',
      maxPriceUsdc: '0.50',
      payload: {
        prompt: 'Check for mainframe arbitrage in Uniswap V3 pools'
      }
    });

    console.log('[TS-Client] Successful answer:');
    console.log('Data:', response.data);
    console.log('Gas spent in USDC:', response.gasSpentUsdc);
  } catch (error) {
    console.error('[TS-Client] Task execution error:', error);
  }
}

runClient();

13.3 A deferred deal: booking a haircut (Python)

A deferred deal differs from an instant one only in its deadlines (document 27): the delivery deadline is the agreed time with a margin, and the dispute window is the one the provider announced. The customer's agent accepts after its owner says the work is done.

from datetime import datetime, timedelta, timezone

async def book_haircut(client, ask_owner):
    # ask_owner — the H2A conversation of the agent with its owner (document 21)
    appointment = datetime(2026, 10, 22, 15, 0, tzinfo=timezone.utc)

    # Search among announcements that offer deferred deals (document 17, tag `deal`)
    salon = await client.find_provider(capability="haircut", max_price_usdc=40.00, deal="deferred")

    booking = await client.execute_task(
        provider_node_id=salon.node_id,
        amount_usdc=40.00,
        payload={"service": "haircut", "time": appointment.isoformat()},
        deal="deferred",
        deliver_by=appointment + timedelta(hours=3),           # the agreed time with a margin
        dispute_window_seconds=salon.dispute_window_seconds    # as the salon announced, at most 14 days
    )

    note = await booking.wait_for_delivery()   # the salon's delivery note: what was done and when
    if await ask_owner(f"The salon reports: {note.text}. Is everything fine?"):
        await booking.accept()                  # the signature closes the deal
    else:
        await booking.cancel()                  # within the dispute window: a refund of at most the weight of likes

    # No delivery recorded by the deadline: await booking.claim_refund() returns the whole amount

Docs Next14. Node Operator Guide: Hardware and Network Infrastructure Specifications