Donner des données produits en direct à un agent IA

"""Two tools an AI model can call: search Amazon and look up one product – live, through everydata.io."""
import json
import os
import re
import time

import requests

API = "https://api.everydata.io"
KEY = os.environ["EVERYDATA_API_KEY"]


def _get(path: str, params: dict) -> dict:
    """One everydata.io request. Errors come back as data, so the model can explain them instead of crashing."""
    retried = False
    for _ in range(4):
        try:
            resp = requests.get(f"{API}{path}", params=params, headers={"x-api-key": KEY}, timeout=60)
        except requests.RequestException as exc:
            return {"error": "network", "detail": str(exc)[:300]}
        if resp.status_code == 429:
            # Too many requests at once. The answer says how long to wait: "... Try again in 12 seconds."
            wait = re.search(r"in (\d+) seconds", resp.text)
            time.sleep(int(wait.group(1)) + 1 if wait else 60)
            continue
        if resp.status_code >= 500 and not retried:
            retried = True  # the page could not be loaded – try once more (5xx answers are not counted)
            time.sleep(5)
            continue
        if resp.status_code >= 400:
            return {"error": resp.status_code, "detail": resp.text[:300]}
        return resp.json()
    return {"error": 429, "detail": "Still rate limited after waiting. Try again in a minute."}


def search_amazon(keyword: str, domainCode: str = "com") -> dict:
    data = _get("/amz/amazon-search-by-keyword-asin",
                {"keyword": keyword, "domainCode": domainCode, "excludeSponsored": "true"})
    if "error" in data:
        return data
    keep = ("asin", "productDescription", "price", "currency", "productRating", "countReview")
    # Only the first five results and only the fields the model needs: fewer tokens, better answers.
    return {"results": [{k: p.get(k) for k in keep} for p in data.get("searchProductDetails", [])[:5]]}


def get_amazon_product(asin: str, domainCode: str = "com") -> dict:
    data = _get("/amz/amazon-lookup-product-by-asin", {"asin": asin, "domainCode": domainCode})
    if "error" in data:
        return data
    keep = ("productTitle", "asin", "price", "retailPrice", "currency",
            "productRating", "countReview", "warehouseAvailability", "features")
    return {k: data.get(k) for k in keep}


FUNCTIONS = {"search_amazon": search_amazon, "get_amazon_product": get_amazon_product}


def run_tool(name: str, args: dict) -> str:
    """Run the tool the model asked for and return the result as a JSON string."""
    fn = FUNCTIONS.get(name)
    if fn is None:
        return json.dumps({"error": f"unknown tool {name}"})
    try:
        return json.dumps(fn(**args), ensure_ascii=False)
    except Exception as exc:  # wrong arguments, unexpected data … – tell the model instead of crashing
        return json.dumps({"error": "tool failed", "detail": f"{type(exc).__name__}: {exc}"[:300]})

# Add to tools.py – one description per tool, written for the model, in plain JSON Schema.
TOOL_SPECS = [
    {
        "name": "search_amazon",
        "description": "Search Amazon right now. Returns up to 5 products with ASIN, title, current price, "
                       "currency and rating. Use it when the user asks what is available or what something costs.",
        "parameters": {
            "type": "object",
            "properties": {
                "keyword": {"type": "string", "description": "What to search for, e.g. 'usb c charger 20w'"},
                "domainCode": {"type": "string", "description": "Amazon marketplace: com, de, co.uk, fr, it, es …"},
            },
            "required": ["keyword"],
        },
    },
    {
        "name": "get_amazon_product",
        "description": "Live details of one Amazon product by ASIN: title, price, list price, availability, "
                       "rating and key features. A price of 0 means the page shows no price right now.",
        "parameters": {
            "type": "object",
            "properties": {
                "asin": {"type": "string", "description": "10-character Amazon product ID, e.g. B07ZPKN6YR"},
                "domainCode": {"type": "string", "description": "Amazon marketplace: com, de, co.uk, fr, it, es …"},
            },
            "required": ["asin"],
        },
    },
]

# OpenAI-style (Chat Completions): the schema goes under "function" -> "parameters".
OPENAI_TOOLS = [{"type": "function", "function": spec} for spec in TOOL_SPECS]

# Anthropic-style (Messages API): the same schema is called "input_schema".
ANTHROPIC_TOOLS = [
    {"name": s["name"], "description": s["description"], "input_schema": s["parameters"]}
    for s in TOOL_SPECS
]

import anthropic

from tools import ANTHROPIC_TOOLS, run_tool

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY
MAX_ROUNDS = 5  # hard limit on tool rounds – every tool call is one everydata.io request
messages = [{"role": "user", "content": "What does a 20W USB-C charger cost on amazon.com right now?"}]

for _ in range(MAX_ROUNDS):
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=16000,
        tools=ANTHROPIC_TOOLS,
        messages=messages,
    )
    if response.stop_reason != "tool_use":
        break
    messages.append({"role": "assistant", "content": response.content})
    messages.append({
        "role": "user",
        "content": [
            {"type": "tool_result", "tool_use_id": block.id, "content": run_tool(block.name, block.input)}
            for block in response.content
            if block.type == "tool_use"
        ],
    })
else:
    raise SystemExit(f"Stopped after {MAX_ROUNDS} tool rounds without a final answer. "
                     "Ask a more specific question or raise MAX_ROUNDS.")

print("".join(block.text for block in response.content if block.type == "text"))

import json
import os

from openai import OpenAI

from tools import OPENAI_TOOLS, run_tool

client = OpenAI()  # reads OPENAI_API_KEY
MODEL = os.environ["OPENAI_MODEL"]  # any chat model that supports function calling
MAX_ROUNDS = 5  # hard limit on tool rounds – every tool call is one everydata.io request
messages = [{"role": "user", "content": "What does a 20W USB-C charger cost on amazon.com right now?"}]

for _ in range(MAX_ROUNDS):
    response = client.chat.completions.create(model=MODEL, messages=messages, tools=OPENAI_TOOLS)
    message = response.choices[0].message
    if not message.tool_calls:
        break
    messages.append(message)
    for call in message.tool_calls:
        try:
            args = json.loads(call.function.arguments)
        except json.JSONDecodeError:
            result = json.dumps({"error": "arguments were not valid JSON"})
        else:
            result = run_tool(call.function.name, args)
        messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
else:
    raise SystemExit(f"Stopped after {MAX_ROUNDS} tool rounds without a final answer. "
                     "Ask a more specific question or raise MAX_ROUNDS.")

print(message.content)