LangChain

Use Cobo Agentic Wallet with LangChain in Python or TypeScript.

Start with the language SDK quickstart first, then move to LangChain once the pact-to-execution flow works without an agent loop.

Before you start

Complete the CLI quickstart first so your runtime already has a paired wallet, API key, and wallet UUID.

5-minute outcome

1. Build LangChain tools from CoboAgentWalletToolkit
2. Submit a pact and wait for owner approval
3. Run an allowed contract-call request
4. Trigger policy denial from a non-compliant contract call
5. Retry using denial guidance
6. Track by request_id and query audit logs

Step 1: Install

Python

pip install "cobo-agentic-wallet[langchain]" langchain-openai

TypeScript

npm install @cobo/agentic-wallet langchain zod

Step 2: Configure environment

export AGENT_WALLET_API_URL=https://api.agenticwallet.cobo.com
export AGENT_WALLET_API_KEY=your-api-key
export OPENAI_API_KEY=your-openai-api-key

Step 3: Wire toolkit and agent

Python

langchain_quickstart.py

import asyncio
import os

from cobo_agentic_wallet import WalletAPIClient
from cobo_agentic_wallet.integrations.langchain import CoboAgentWalletToolkit
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

INCLUDE_TOOLS = [
    "submit_pact", "get_pact", "transfer_tokens",
    "estimate_transfer_fee", "get_transaction_record_by_request_id", "get_audit_logs",
]


async def main() -> None:
    _ = os.environ["OPENAI_API_KEY"]  # consumed implicitly by ChatOpenAI
    wallet_id = os.environ["AGENT_WALLET_WALLET_ID"]
    destination = os.environ.get("CAW_DESTINATION", "0x1111111111111111111111111111111111111111")
    prompt = (
        f"Use wallet {wallet_id}. Submit a pact for a controlled transfer task and wait until "
        f"it is active. Using the newly created pact, transfer 0.001 SETH to {destination} on "
        f"SETH. Next, using the same pact, attempt 0.005 SETH. If denied, follow the denial "
        f"guidance and retry with a compliant amount. Track the result by request_id and "
        f"summarize what happened."
    )

    async with WalletAPIClient(
        base_url=os.environ["AGENT_WALLET_API_URL"],
        api_key=os.environ["AGENT_WALLET_API_KEY"],
    ) as client:
        toolkit = CoboAgentWalletToolkit(client=client, include_tools=INCLUDE_TOOLS)
        try:
            agent = create_agent(
                model=ChatOpenAI(model="gpt-4.1-mini"),
                tools=toolkit.get_tools(),
            )
            result = await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]})
            print(result["messages"][-1].content)
        finally:
            await toolkit.aclose()


asyncio.run(main())

If you want a tighter first integration, start with a smaller tool subset:

toolkit = CoboAgentWalletToolkit(
    client=client,
    include_tools=[
        "submit_pact",
        "get_pact",
        "transfer_tokens",
        "estimate_transfer_fee",
        "get_transaction_record_by_request_id",
        "get_audit_logs",
    ],
)
tools = toolkit.get_tools()

TypeScript

langchain_quickstart.ts

import { randomUUID } from 'node:crypto';
import { tool, createAgent } from 'langchain';
import { z } from 'zod';

import {
  AuditApi,
  Configuration,
  PactsApi,
  TransactionRecordsApi,
  TransactionsApi,
} from '@cobo/agentic-wallet';

// ─── Env ─────────────────────────────────────────────────────────────────────

function requireEnv(name: string): string {
  const v = process.env[name];
  if (!v) throw new Error(`Missing required environment variable: ${name}`);
  return v;
}

const env = {
  basePath: requireEnv('AGENT_WALLET_API_URL'),
  ownerKey: requireEnv('AGENT_WALLET_API_KEY'),
  walletId: requireEnv('AGENT_WALLET_WALLET_ID'),
  openaiApiKey: requireEnv('OPENAI_API_KEY'),
  destination: process.env.CAW_DESTINATION ?? '0x1111111111111111111111111111111111111111',
};

// ─── Owner-scoped API clients ────────────────────────────────────────────────

const ownerConfig = new Configuration({ apiKey: env.ownerKey, basePath: env.basePath });
const pactsApi = new PactsApi(ownerConfig);
const ownerTxApi = new TransactionsApi(ownerConfig);
const recordsApi = new TransactionRecordsApi(ownerConfig);
const auditApi = new AuditApi(ownerConfig);

// ─── Demo pact spec ──────────────────────────────────────────────────────────
// Allow SETH transfers up to 0.002; anything larger is denied by policy.

const CHAIN_ID = 'SETH';
const TOKEN_ID = 'SETH';
const DEMO_PACT_SPEC = {
  policies: [
    {
      name: 'max-tx-limit',
      type: 'transfer',
      rules: {
        effect: 'allow',
        when: {
          chain_in: [CHAIN_ID],
          token_in: [{ chain_id: CHAIN_ID, token_id: TOKEN_ID }],
        },
        deny_if: { amount_gt: '0.002' },
      },
    },
  ],
  completion_conditions: [{ type: 'time_elapsed', threshold: '86400' }],
};

// ─── Pact session store ──────────────────────────────────────────────────────
// Capture (pact_id → api_key) as submit_pact / get_pact responses come back,
// then resolve pact-scoped TransactionsApi clients at transfer time.

const pactApiKeys = new Map<string, string>();

function capturePact(response: unknown): void {
  if (!response || typeof response !== 'object') return;
  const r = response as Record<string, unknown>;
  const pactId = (r.pact_id ?? r.id) as string | undefined;
  const apiKey = r.api_key as string | undefined;
  if (pactId && apiKey) pactApiKeys.set(pactId, apiKey);
}

function txApiForPact(pactId?: string): TransactionsApi {
  if (!pactId) return ownerTxApi;
  const apiKey = pactApiKeys.get(pactId);
  if (!apiKey) throw new Error(`Unknown pact_id ${pactId}. Call submit_pact or get_pact first.`);
  return new TransactionsApi(new Configuration({ apiKey, basePath: env.basePath }));
}

// ─── Denial handling ─────────────────────────────────────────────────────────
// Surface policy denials back to the LLM as `{ error, suggestion }` so it can
// self-correct, instead of aborting the agent loop with an exception.

async function catchPolicyDenial<T>(
  work: () => Promise<T>,
): Promise<T | { error: unknown; suggestion?: string }> {
  try {
    return await work();
  } catch (err) {
    const data = (err as { response?: { data?: { error?: unknown; suggestion?: string } } })
      ?.response?.data;
    return { error: data?.error ?? 'UNKNOWN_ERROR', suggestion: data?.suggestion };
  }
}

// ─── Prompts ─────────────────────────────────────────────────────────────────

const DEMO_USER_PROMPT =
  `Use wallet ${env.walletId}. ` +
  `Submit a pact for a controlled transfer task and wait until it is active. ` +
  `Using the newly created pact, transfer 0.001 ${TOKEN_ID} to ${env.destination} on ${CHAIN_ID}. ` +
  `Next, using the same pact, attempt 0.005 ${TOKEN_ID}. If denied, follow the denial ` +
  `guidance and retry with a compliant amount. ` +
  `Track the result by request_id and summarize what happened.`;

// ─── Tool definitions (LangChain-specific) ───────────────────────────────────

const submitPact = tool(
  async ({ wallet_id, intent }) => {
    const { data } = await pactsApi.submitPact({ wallet_id, intent, spec: DEMO_PACT_SPEC });
    capturePact(data.result);
    return data.result;
  },
  {
    name: 'submit_pact',
    description: 'Submit a pact and return the pact id.',
    schema: z.object({ wallet_id: z.string(), intent: z.string() }),
  },
);

const getPact = tool(
  async ({ pact_id }) => {
    const { data } = await pactsApi.getPact(pact_id);
    capturePact(data.result);
    return data.result;
  },
  {
    name: 'get_pact',
    description: 'Fetch a pact, including its status and api_key once active.',
    schema: z.object({ pact_id: z.string() }),
  },
);

const estimateTransferFee = tool(
  async ({ wallet_uuid, dst_addr, token_id, amount, pact_id }) => {
    const { data } = await txApiForPact(pact_id).estimateTransferFee(wallet_uuid, {
      chain_id: CHAIN_ID,
      dst_addr,
      token_id,
      amount,
    });
    return data.result;
  },
  {
    name: 'estimate_transfer_fee',
    description: 'Estimate fees for a token transfer before submitting it.',
    schema: z.object({
      wallet_uuid: z.string(),
      dst_addr: z.string(),
      token_id: z.string(),
      amount: z.string(),
      pact_id: z.string().optional(),
    }),
  },
);

const transferTokens = tool(
  async ({ wallet_uuid, dst_addr, token_id, amount, pact_id }) =>
    catchPolicyDenial(async () => {
      const { data } = await txApiForPact(pact_id).transferTokens(wallet_uuid, {
        chain_id: CHAIN_ID,
        dst_addr,
        token_id,
        amount,
        request_id: randomUUID(),
      });
      return data.result;
    }),
  {
    name: 'transfer_tokens',
    description:
      'Execute a policy-enforced transfer. A unique request_id is auto-generated ' +
      'and returned; use that value to track or look up the tx later. Pass pact_id ' +
      'to invoke under pact-scoped policy permissions.',
    schema: z.object({
      wallet_uuid: z.string(),
      dst_addr: z.string(),
      token_id: z.string(),
      amount: z.string(),
      pact_id: z.string().optional(),
    }),
  },
);

const getTransactionRecordByRequestId = tool(
  async ({ wallet_uuid, request_id }) => {
    const { data } = await recordsApi.getUserTransactionByRequestId(wallet_uuid, request_id);
    return data.result;
  },
  {
    name: 'get_transaction_record_by_request_id',
    description: 'Look up a transaction record by request id.',
    schema: z.object({ wallet_uuid: z.string(), request_id: z.string() }),
  },
);

const getAuditLogs = tool(
  async ({ wallet_id, limit }) => {
    const { data } = await auditApi.listAuditLogs(
      wallet_id,
      undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined,
      limit ?? 20,
    );
    const items = (data.result as { items?: Array<{ result?: string }> })?.items ?? [];
    return {
      items,
      allowed: items.filter(it => it.result === 'allowed').length,
      denied: items.filter(it => it.result === 'denied').length,
    };
  },
  {
    name: 'get_audit_logs',
    description: 'List recent audit log entries for the wallet.',
    schema: z.object({
      wallet_id: z.string(),
      limit: z.number().int().positive().optional(),
    }),
  },
);

// ─── Boot ────────────────────────────────────────────────────────────────────

const agent = createAgent({
  model: 'openai:gpt-4.1-mini',
  tools: [
    submitPact,
    getPact,
    transferTokens,
    estimateTransferFee,
    getTransactionRecordByRequestId,
    getAuditLogs,
  ],
});

const result = await agent.invoke({
  messages: [{ role: 'user', content: DEMO_USER_PROMPT }],
});
console.log(result.messages.at(-1)?.content);

Add get_audit_logs once you want the model to explain operator history as part of the same loop. Keep the first example narrower so the pact-to-execution path is easier to follow.

Step 4: Confirm denial loop happened

Verify agent output includes:

• a pact submission and activation confirmation
• a successful contract call
• a policy denial payload/text
• a corrected retry
• a transaction lookup by request_id
• audit log summary

Keep the LangChain layer thin

LangChain should wrap the canonical CAW flow, not replace it with a different mental model.

• keep CAW tools limited to the exact runtime role
• keep pact drafting separate from broad wallet management
• prefer get_transaction_record_by_request_id over ad hoc polling patterns
• move deterministic business logic outside the model loop when possible

Go further

In Python, CoboAgentWalletToolkit gives you the widened CAW runtime toolkit directly. In TypeScript, the usual pattern is to wrap @cobo/agentic-wallet calls in LangChain tools and expose only the subset your runtime needs.

• Define custom LangChain tools — in Python, register @tool functions alongside CAW’s adapter; in TypeScript, wrap the CAW SDK in tool(...) definitions with zod schemas.
• Add CLI tools — call caw via subprocess for operations you want to handle as shell commands rather than Python calls.
• Use direct SDK calls when determinism matters — for pact lifecycle management, audit queries, or recovery flows, keep the CAW call outside the agent loop.
• Use role-based presets — a good default split is Pact Drafting, Execution, and Observer. Start narrow and only add more tools when the runtime proves it needs them.

Python SDK

Use WalletAPIClient directly for custom tool functions.