All Posts

How AI Agent Frameworks Connect to UCP: A Developer Guide

A practical guide for connecting LangChain, CrewAI, AutoGen, and other AI agent frameworks to UCP merchant endpoints. Includes code examples for each framework.

July 20, 2026UCPList Team
LangChain UCPCrewAI UCPagent frameworks UCPAI agent commerce integration

The Setup

UCP gives AI agents a standard interface for commerce: discover a merchant's manifest at /.well-known/ucp, read the declared capabilities, and call the endpoints. Simple in principle. The interesting engineering question is how you connect that to the agent frameworks your team is already using.

This guide covers four frameworks: LangChain, CrewAI, OpenAI Agents SDK, and AutoGen. Each one exposes UCP endpoints differently, but the underlying pattern is the same: wrap UCP API calls as callable tools, hand those tools to an agent, let the agent decide when to use them.

LangChain: Tools and LangGraph

LangChain's @tool decorator is the simplest way to expose a UCP endpoint to an agent. Define a Python function, decorate it, and add it to your agent's tool list.

from langchain_core.tools import tool
import httpx

@tool
def ucp_catalog_search(merchant_domain: str, query: str) -> dict:
    """Search a UCP merchant's product catalog by keyword."""
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["catalogEndpoint"], json={"query": query}).json()

@tool
def ucp_checkout(merchant_domain: str, items: list, payment_token: str) -> dict:
    """Complete a UCP checkout. Requires a pre-authorized payment token."""
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["checkoutEndpoint"], json={
        "items": items,
        "paymentToken": payment_token,
    }).json()

For more complex shopping workflows, LangGraph is the right abstraction. Define your workflow as a graph with explicit state transitions:

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class ShoppingState(TypedDict):
    user_request: str
    merchant_domain: str
    catalog_results: List[dict]
    selected_items: List[dict]
    order_confirmation: dict

graph = StateGraph(ShoppingState)

graph.add_node("search_catalog", search_catalog_node)
graph.add_node("select_items", select_items_node)
graph.add_node("confirm_with_user", confirm_with_user_node)
graph.add_node("execute_checkout", execute_checkout_node)

graph.set_entry_point("search_catalog")
graph.add_edge("search_catalog", "select_items")
graph.add_edge("select_items", "confirm_with_user")
graph.add_conditional_edges("confirm_with_user",
    lambda state: "execute_checkout" if state["confirmed"] else END)
graph.add_edge("execute_checkout", END)

LangSmith traces every node execution. When a checkout fails, you can see exactly what payload was sent to which endpoint and what the merchant returned.

CrewAI: Agents with Roles

CrewAI shines when you want agent specialization. A researcher agent searches catalogs. A buyer agent executes checkouts. Each agent only has access to the tools relevant to its role.

from crewai import Agent, Task, Crew, Tool
import httpx

def fetch_catalog(merchant_domain: str, query: str) -> dict:
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["catalogEndpoint"], json={"query": query}).json()

def place_order(merchant_domain: str, items: list, payment_token: str) -> dict:
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["checkoutEndpoint"], json={
        "items": items,
        "paymentToken": payment_token,
    }).json()

catalog_tool = Tool(name="catalog_search", func=fetch_catalog,
    description="Search a UCP merchant's product catalog")
checkout_tool = Tool(name="ucp_checkout", func=place_order,
    description="Execute a UCP checkout order")

researcher = Agent(
    role="Product Researcher",
    goal="Find products matching the user's criteria across UCP merchants",
    tools=[catalog_tool],
    verbose=True
)

buyer = Agent(
    role="Buyer Agent",
    goal="Complete the purchase using the best option from research",
    tools=[checkout_tool],
    verbose=True
)

research_task = Task(description="Find the best matching product at {merchant}", agent=researcher)
purchase_task = Task(description="Place the order for the selected product", agent=buyer)

crew = Crew(agents=[researcher, buyer], tasks=[research_task, purchase_task])
result = crew.kickoff(inputs={"merchant": "allbirds.com"})

The key point: the buyer agent does not have the catalog_tool, and the researcher agent does not have the checkout_tool. Payment token access is scoped to the agent that needs it.

OpenAI Agents SDK: Handoffs

The OpenAI Agents SDK (formerly Swarm) uses handoffs to pass control between agents. A general assistant detects a commerce intent and hands off to a specialized shopping agent.

from agents import Agent, Runner, function_tool
import httpx

@function_tool
def search_catalog(merchant_domain: str, query: str) -> dict:
    """Search a UCP-enabled merchant's product catalog."""
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["catalogEndpoint"], json={"query": query}).json()

@function_tool
def place_order(merchant_domain: str, sku: str, quantity: int, payment_token: str) -> dict:
    """Place a UCP checkout order. Only call after confirming with the user."""
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["checkoutEndpoint"], json={
        "items": [{"sku": sku, "quantity": quantity}],
        "paymentToken": payment_token,
    }).json()

shopping_agent = Agent(
    name="Shopping Agent",
    instructions="""You are a specialized shopping agent with access to UCP merchant endpoints.
    Search catalogs to find products, confirm selection with the user, then place orders.
    Never place an order without explicit user confirmation.""",
    tools=[search_catalog, place_order],
)

general_agent = Agent(
    name="Assistant",
    instructions="Help with general tasks. Hand off shopping requests to the Shopping Agent.",
    handoffs=[shopping_agent],
)

result = await Runner.run(general_agent, "Can you order me some running shoes?")

The handoff pattern keeps payment tokens scoped: only the shopping_agent has place_order access.

AutoGen: Human-in-the-Loop

AutoGen's strength is explicit human approval steps. For commerce workflows where the user needs to confirm before purchase, AutoGen's conversational pattern is more natural than adding a confirmation step inside a tool function.

import autogen
import httpx

def ucp_checkout(merchant_domain: str, items: list, payment_token: str) -> dict:
    manifest = httpx.get(f"https://{merchant_domain}/.well-known/ucp").json()
    return httpx.post(manifest["checkoutEndpoint"], json={
        "items": items,
        "paymentToken": payment_token,
    }).json()

config_list = [{"model": "gpt-4o", "api_key": "..."}]

shopping_agent = autogen.AssistantAgent(
    name="ShoppingAgent",
    llm_config={"config_list": config_list},
    system_message="""Search UCP catalogs and present options to the user.
    Do not call ucp_checkout until the user says 'confirm' or 'proceed'.""",
    function_map={"ucp_checkout": ucp_checkout},
)

# human_input_mode="ALWAYS" means every agent message goes to the human for approval
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="ALWAYS",
    code_execution_config=False,
)

user_proxy.initiate_chat(
    shopping_agent,
    message="Find a standing desk under $500 and let me know what you find."
)

With human_input_mode="ALWAYS", the user sees the agent's catalog results before any checkout is triggered. Set human_input_mode="TERMINATE" if you want the agent to run autonomously and only stop when done.

Practical Notes

Cache the manifest. Every UCP call in these examples fetches the manifest first. In production, cache it with a short TTL (5-10 minutes). The manifest changes rarely and fetching it on every tool call adds latency.

Handle capability gaps. Not every UCP merchant declares every capability. Before calling an endpoint, check that the manifest lists the capability:

def get_capability_endpoint(manifest: dict, capability: str) -> str | None:
    if capability not in manifest.get("capabilities", []):
        return None
    return manifest.get(f"{capability}Endpoint")

Scope payment tokens. Never pass the payment token to catalog search tools. Only the checkout tool needs it. This is true for all four frameworks.

Log tool call results. LangSmith (LangChain), CrewAI's verbose mode, and AutoGen's conversation logs all record tool inputs and outputs. Use them. Debugging a failed checkout without a trace is much harder than with one.

What to Build Next

All four frameworks work with UCP today. The patterns above are sufficient for production commerce agents. The next layer is multi-merchant search: querying UCP catalog endpoints across several merchants simultaneously, merging results, and presenting a unified view to the LLM.

LlamaIndex's query pipelines are built for this pattern. If your use case involves product comparison across merchants rather than transacting with a single known merchant, start there.

Read next