Amazon — Shopping Assistant

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

# .env
SYNAP_API_KEY=...
SYNAP_SERVER_URL=<maximem-server>
OPENAI_API_KEY=...

SESSIONS: dict[str, str] = {}

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

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 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

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])