LangChain

pip install "langchain[mcp]>=1.4.0"

import os
from fastmcp.client import Client
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter

AIX_MCP = "https://aix.brambles.ai/api/aix/mcp"


async def run(user_message: str) -> str:
    async with MCPAdapter(Client(AIX_MCP, auth=os.environ["AIX_PUBLISHER_KEY"])) as adapter:
        tools = await adapter.list_tools()
        agent = create_agent(model, tools, system_prompt=AIX_RULES)  # see below
        result = await agent.ainvoke({"messages": [("user", user_message)]})
        return result["messages"][-1].content

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    "aix": {
        "transport": "streamable_http",
        "url": "https://aix.brambles.ai/api/aix/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['AIX_PUBLISHER_KEY']}"},
    }
})
tools = await client.get_tools()

import os
from typing import Literal

import httpx
from langchain_core.tools import tool
from pydantic import BaseModel, Field

AIX = "https://aix.brambles.ai/api/aix"


class Budget(BaseModel):
    currency: str = Field(min_length=3, max_length=3)
    max_cents: int


class Fact(BaseModel):
    attribute: Literal["brand", "width", "size", "department", "use_case", "feature", "merchant", "deal"]
    value: str
    strength: Literal["required", "preferred", "excluded"]


class Task(BaseModel):
    category: str
    summary: str = Field(max_length=1600)
    budget: Budget | None = None
    facts: list[Fact] = []


class Intent(BaseModel):
    session_id: str = Field(min_length=1, max_length=128)
    country: str = Field(min_length=2, max_length=2)
    task: Task


async def _aix(path: str, body: dict) -> dict:
    key = os.environ.get("AIX_PUBLISHER_KEY")
    if not key:
        return {"status": "skipped", "reason": "no_key"}
    try:
        async with httpx.AsyncClient(timeout=3) as client:
            r = await client.post(f"{AIX}/{path}", json=body, headers={"Authorization": f"Bearer {key}"})
            return r.json() if r.is_success else {"status": "skipped", "reason": f"http_{r.status_code}"}
    except httpx.HTTPError:
        return {"status": "skipped", "reason": "timeout"}  # never block the shopper


@tool
async def submit_intent(intent: Intent) -> dict:
    """Report the shopper's current shopping task to AIX. Call after any turn that states or changes
    what they want to buy, with intent: session_id, country and the task. Shopping requirements only,
    never personal data."""
    return await _aix("intents", intent.model_dump(exclude_none=True))


@tool
async def report_handoff(session_id: str, destination_url: str, consent: str | None = None) -> dict:
    """Report that the shopper is being sent to a store or product page (the store's own URL).
    Use the returned landing_url as that page's URL, inside your affiliate link if you use one."""
    body: dict = {"session_id": session_id, "destination_url": destination_url}
    if consent:
        body["consent"] = consent
    return await _aix("handoffs", body)


agent = create_agent(model, [submit_intent, report_handoff], system_prompt=AIX_RULES)

You can send shopping intent to the Agentic Intent Exchange (AIX).
- After any user turn that states or changes what they want to buy, call submit_intent with one
  argument, intent: an object holding the conversation's session_id, the shopper's two-letter
  country, and the task: a category, a one-line summary, the budget when they gave one, and facts
  with attribute, value and strength (required, preferred or excluded). Send only shopping
  requirements: never names, emails, phone numbers, addresses, account or order numbers, or the
  conversation itself.
- Before you give the user a link to a store or product page, call report_handoff with the same
  session_id and the store's page URL, then use the returned landing_url as that page's URL (inside
  your own affiliate or ad link, if you use one).
- If either tool fails or is slow, carry on without it. Never ask the user for a key.