Using Mixpeek in the IAB Tech Lab buyer agent: classification, brand safety, contextual search

# setup_iab_namespace.py  (Python 3.10+, stdlib only)
import csv, io, json, os, sys, time, urllib.request

API = "https://api.mixpeek.com"
TSV = ("https://raw.githubusercontent.com/InteractiveAdvertisingBureau/Taxonomies/main/"
       "Content%20Taxonomies/Content%20Taxonomy%203.1.tsv")

def call(method, path, body=None, ns=None):
    headers = {"Authorization": f"Bearer {os.environ['MIXPEEK_API_KEY']}", "Content-Type": "application/json"}
    if ns:
        headers["X-Namespace"] = ns
    req = urllib.request.Request(API + path, method=method, headers=headers,
                                 data=json.dumps(body).encode() if body is not None else None)
    return json.load(urllib.request.urlopen(req, timeout=300))

rows = list(csv.reader(io.StringIO(urllib.request.urlopen(TSV).read().decode()), delimiter="\t"))
start = next(i + 1 for i, r in enumerate(rows) if r and r[0].strip() == "Unique ID")
cats = []
for r in rows[start:]:
    if r and r[0].strip():
        path = [t.strip() for t in (r + [""] * 7)[3:7] if t.strip()]
        cats.append({"id": r[0].strip(), "name": r[2].strip(), "tier": len(path), "path": path})

ns = call("POST", "/v1/namespaces", {"namespace_name": "iab-content-taxonomy-3-1",
          "feature_extractors": [{"feature_extractor_name": "text_extractor", "version": "v1"}]})["namespace_id"]
call("PATCH", f"/v1/namespaces/{ns}", {"spend_budget": {"daily_usd": 1, "monthly_usd": 2}})

bucket = call("POST", "/v1/buckets", {"bucket_name": "iab_categories", "bucket_schema": {"properties": {
    "category_text": {"type": "text", "required": True},
    "iab_category_id": {"type": "string", "required": True},
    "iab_category_name": {"type": "string", "required": True},
    "iab_tier": {"type": "integer", "required": True},
    "iab_path": {"type": "array", "items": {"type": "string"}, "required": False}}}}, ns)["bucket_id"]

objs = [{"blobs": [{"property": "category_text", "type": "text",
                    "data": f"{c['name']}. Category path: {' > '.join(c['path'])}."}],
         "iab_category_id": c["id"], "iab_category_name": c["name"],
         "iab_tier": c["tier"], "iab_path": c["path"]} for c in cats]
for i in range(0, len(objs), 100):
    call("POST", f"/v1/buckets/{bucket}/objects/batch", {"objects": objs[i:i + 100]}, ns)

col = call("POST", "/v1/collections", {"collection_name": "iab_categories_text",
    "source": {"type": "bucket", "bucket_ids": [bucket]},
    "feature_extractor": {"feature_extractor_name": "text_extractor", "version": "v1",
        "input_mappings": {"text": "category_text"},
        "field_passthrough": [{"source_path": k} for k in
                              ("iab_category_id", "iab_category_name", "iab_tier", "iab_path")]}}, ns)["collection_id"]
task = call("POST", f"/v1/collections/{col}/trigger", {}, ns)["task_id"]

ret = call("POST", "/v1/retrievers", {"retriever_name": "iab_text_search", "collection_identifiers": [col],
    "input_schema": {"query": {"type": "text", "required": True}},
    "stages": [{"stage_name": "semantic_search", "stage_type": "filter", "config": {
        "stage_id": "feature_search", "parameters": {
            "searches": [{"feature_uri": "mixpeek://text_extractor@v1/multilingual_e5_large_instruct_v1",
                          "query": {"input_mode": "text", "text": "{{INPUT.query}}"}, "top_k": 50}],
            "final_top_k": 20}}}]}, ns)
ret = ret.get("retriever", ret)["retriever_id"]

while True:  # wait until all 704 categories are searchable
    if call("GET", f"/v1/tasks/{task}", ns=ns).get("status") == "FAILED":
        sys.exit(call("GET", f"/v1/tasks/{task}", ns=ns)["additional_data"]["error"])
    n = call("GET", f"/v1/collections/{col}/documents?page_size=1&include_total=true", ns=ns)["pagination"]["total"]
    print(f"{n}/{len(cats)} indexed")
    if n >= len(cats):
        break
    time.sleep(15)
print(json.dumps({"MIXPEEK_NAMESPACE": ns, "MIXPEEK_IAB_RETRIEVER_ID": ret}))

curl -s -X POST "https://api.mixpeek.com/v1/retrievers/$RETRIEVER_ID/execute" \
  -H "Authorization: Bearer $MIXPEEK_API_KEY" -H "X-Namespace: $NAMESPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{"inputs": {"query": "The Chiefs clinched the AFC West with a late field goal in Sunday night football."}}'

git clone https://github.com/IABTechLab/buyer-agent && cd buyer-agent
uv sync --all-extras
cat >> .env <<'EOF'
MIXPEEK_API_KEY=your_mixpeek_api_key
MIXPEEK_BASE_URL=https://api.mixpeek.com
MIXPEEK_NAMESPACE=your_namespace_id
EOF

import asyncio, json
from ad_buyer.interfaces import mcp_server as m

async def main():
    print(await m.classify_content(
        text="The Chiefs clinched the AFC West with a late field goal in Sunday night football."))
    print(await m.check_brand_safety(
        text="Claim a 200 percent deposit bonus at our online casino and play poker and slots tonight."))
    rid = await m._discover_iab_retriever(m._get_mixpeek_client())
    print(rid, await m.contextual_search(query="home renovation", retriever_id=rid, limit=3))

asyncio.run(main())

UNSAFE_3_1 = {
    "Poker and Professional Gambling", "Casinos & Gambling", "Casino and Gambling Video Games",
    "Lotteries and Scratchcards", "Sensitive Topics", "Adult & Explicit Sexual Content",
    "Adult Video Games", "Debated Sensitive Social Issues", "Terrorism", "Crime",
    "Crime & Harmful Acts to Individuals, Society & Human Right Violations",
    "Illegal Drugs, Tobacco, eCigarettes, Vaping, Alcohol", "Arms & Ammunition",
    "Death, Injury, or Military Conflict",
}
async def flagged_categories(page_text):
    result = json.loads(await m.check_brand_safety(text=page_text))
    return [c for c in result["categories"] if c["category"] in UNSAFE_3_1]