> ## Documentation Index
> Fetch the complete documentation index at: https://cobo.com/products/agentic-wallet/manual/llms.txt
> Use this file to discover all available pages before exploring further.
# 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](/products/agentic-wallet/manual/developer/cli) 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
```bash theme={null}
pip install "cobo-agentic-wallet[langchain]" langchain-openai
```
```bash theme={null}
npm install @cobo/agentic-wallet langchain zod
```
## Step 2: Configure environment
```bash theme={null}
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 title="langchain_quickstart.py" theme={null}
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:
```python theme={null}
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 title="langchain_quickstart.ts" theme={null}
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();
function capturePact(response: unknown): void {
if (!response || typeof response !== 'object') return;
const r = response as Record;
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(
work: () => Promise,
): Promise {
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.
Use WalletAPIClient directly for custom tool functions.