Amazon: Shopping Assistant

pip install maximem-synap maximem-synap-openai-agents openai-agents

uv add maximem-synap maximem-synap-openai-agents openai-agents
# pip-compatible (existing venv): uv pip install maximem-synap maximem-synap-openai-agents openai-agents

npm install @maximem/synap-js-sdk @maximem/synap-vercel-adk ai @ai-sdk/openai zod

# .env
SYNAP_API_KEY=...
OPENAI_API_KEY=...

# .env.local
SYNAP_API_KEY=...
OPENAI_API_KEY=...

SESSIONS: dict[str, str] = {}

def conv_for(session_id: str) -> str:
    return SESSIONS.setdefault(session_id, str(uuid.uuid4()))

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

function convFor(sessionId: string): string {
  if (!SESSIONS.has(sessionId)) SESSIONS.set(sessionId, crypto.randomUUID());
  return SESSIONS.get(sessionId)!;
}

from agents import function_tool

@function_tool
async def search_catalog(query: str, filters: dict = None, limit: int = 10) -> list[dict]:
    """Search the catalog. Filters: {category, price_max, brand, dietary, material}."""
    return await catalog_api.search(query, filters or {}, limit)

@function_tool
async def get_product_details(asin: str) -> dict:
    """Return full product detail: title, price, specs, reviews_summary, in_stock."""
    return await catalog_api.detail(asin)

@function_tool
async def get_order_history(buyer_id: str, limit: int = 10) -> list[dict]:
    """Return the buyer's recent orders, newest first."""
    return await orders_api.history(buyer_id, limit)

@function_tool
async def add_to_cart(buyer_id: str, asin: str, qty: int = 1) -> dict:
    return await cart_api.add(buyer_id, asin, qty)

@function_tool
async def track_order(order_id: str) -> dict:
    return await orders_api.track(order_id)

@function_tool
async def request_return(order_id: str, reason: str, items: list[str]) -> dict:
    return await returns_api.create(order_id, reason, items)

import { tool } from "ai";
import { z } from "zod";

const businessTools = {
  search_catalog: tool({
    description: "Search the catalog with optional filters.",
    parameters: z.object({
      query: z.string(),
      filters: z.object({
        category: z.string().optional(),
        priceMax: z.number().optional(),
        brand: z.string().optional(),
        dietary: z.string().optional(),
        material: z.string().optional(),
      }).optional(),
      limit: z.number().default(10),
    }),
    execute: async ({ query, filters, limit }) =>
      catalogApi.search(query, filters ?? {}, limit),
  }),
  get_product_details: tool({
    description: "Return full product detail.",
    parameters: z.object({ asin: z.string() }),
    execute: async ({ asin }) => catalogApi.detail(asin),
  }),
  get_order_history: tool({
    description: "Return the buyer's recent orders, newest first.",
    parameters: z.object({ buyerId: z.string(), limit: z.number().default(10) }),
    execute: async ({ buyerId, limit }) => ordersApi.history(buyerId, limit),
  }),
  add_to_cart: tool({
    description: "Add an item to the buyer's cart.",
    parameters: z.object({ buyerId: z.string(), asin: z.string(), qty: z.number().default(1) }),
    execute: async ({ buyerId, asin, qty }) => cartApi.add(buyerId, asin, qty),
  }),
  track_order: tool({
    description: "Return tracking info for an order.",
    parameters: z.object({ orderId: z.string() }),
    execute: async ({ orderId }) => ordersApi.track(orderId),
  }),
  request_return: tool({
    description: "Open a return for one or more items in an order.",
    parameters: z.object({
      orderId: z.string(),
      reason: z.string(),
      items: z.array(z.string()),
    }),
    execute: async ({ orderId, reason, items }) =>
      returnsApi.create(orderId, reason, items),
  }),
};

import os, uuid, asyncio
from agents import Agent, FunctionTool, Runner
from maximem_synap import MaximemSynapSDK
from synap_openai_agents import create_search_tool, create_store_tool

sdk = MaximemSynapSDK()
await sdk.initialize()

async def handle_message(buyer_id: str, session_id: str, text: str) -> str:
    conv_id = conv_for(session_id)

    synap_search = create_search_tool(sdk=sdk, user_id=buyer_id, customer_id="amazon")
    synap_store  = create_store_tool(sdk=sdk, user_id=buyer_id, customer_id="amazon")

    agent = Agent(
        name="amazon_shopping",
        instructions=SYSTEM,
        tools=[
            FunctionTool(synap_search, name_override="synap_search"),
            FunctionTool(synap_store,  name_override="synap_store"),
            search_catalog,
            get_product_details,
            get_order_history,
            add_to_cart,
            track_order,
            request_return,
        ],
    )

    result = await Runner.run(agent, input=text)
    reply = result.final_output

    asyncio.create_task(sdk.memories.create(
        document=f"Buyer: {text}\nAssistant: {reply}",
        document_type="ai-chat-conversation",
        user_id=buyer_id,
        customer_id="amazon",
        metadata={"conversation_id": conv_id, "channel": "web-chat"},
    ))
    return reply

import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";
import { createSynap } from "@maximem/synap-vercel-adk";

const synap = await createSynap({ apiKey: process.env.SYNAP_API_KEY! });

export async function handleMessage(
  buyerId: string,
  sessionId: string,
  text: string,
): Promise<string> {
  const conversationId = convFor(sessionId);

  const model = synap.wrap(openai("gpt-4o"), {
    userId: buyerId,
    customerId: "amazon",
    conversationId,
  });

  const { text: reply } = await generateText({
    model,
    system: SYSTEM,
    prompt: text,
    tools: businessTools,
  });

  return reply;
}

docs = await sdk.memories.list(user_id="buyer_42", customer_id="amazon", limit=20)
for d in docs:
    print(d.document_type, ":", d.content[:120])

const docs = await synap.sdk.memories.list({ userId: "buyer_42", customerId: "amazon", limit: 20 });
for (const d of docs) console.log(d.documentType, ":", d.content.slice(0, 120));