Use PartFinder as an agent tool

# partfinder_tools.py
import json
import os
import time
from urllib.parse import quote

import requests

API = "https://api.partfinder.dev"
HEADERS = {"Authorization": f"Bearer {os.environ['PARTFINDER_API_KEY']}"}

doc = requests.get(f"{API}/v1/openapi.json").json()


# Inline every "$ref" so each tool's schema stands alone.
def inline(s):
    if isinstance(s, list):
        return [inline(v) for v in s]
    if isinstance(s, dict) and "$ref" in s:
        return inline(doc["components"]["schemas"][s["$ref"].split("/")[-1]])
    if isinstance(s, dict):
        return {k: inline(v) for k, v in s.items()}
    return s


# One tool per operation. Keys and webhooks stay with your own code, and so does the file upload.
operations = []
for path, methods in doc["paths"].items():
    for method, op in methods.items():
        if "Keys" in op["tags"] or "Webhooks" in op["tags"]:
            continue
        content = op.get("requestBody", {}).get("content")
        if content and "application/json" not in content:
            continue
        params = [p for p in op.get("parameters", []) if p["in"] != "header"]
        body = inline(content["application/json"]["schema"]) if content else {}
        operations.append({
            "name": op["operationId"],
            "description": op.get("description") or op.get("summary"),
            "method": method.upper(),
            "path": path,
            "body_fields": list(body.get("properties", {})),
            "parameters": {
                "type": "object",
                "properties": {
                    **{p["name"]: {**p["schema"], "description": p["description"]} for p in params},
                    **body.get("properties", {}),
                },
                "required": [p["name"] for p in params if p.get("required")] + body.get("required", []),
            },
        })


def status_of(result):
    return result.get("status") or (result.get("bom") or {}).get("status")


# Make the call a tool call names. A job answers 202: read it until it is no longer processing.
def call_tool(name, args):
    op = next(o for o in operations if o["name"] == name)
    path = op["path"]
    query, body = {}, {}
    for key, value in (args or {}).items():
        if value is None:
            continue
        if f"{{{key}}}" in path:
            path = path.replace(f"{{{key}}}", quote(str(value), safe=""))
        elif key in op["body_fields"]:
            body[key] = value
        else:
            query[key] = str(value).lower() if isinstance(value, bool) else str(value)
    res = requests.request(
        op["method"],
        f"{API}{path}",
        headers=HEADERS,
        params=query,
        json=body if op["body_fields"] else None,
    )
    result = res.json()
    location = res.headers.get("location", res.url)
    while res.status_code == 202 or status_of(result) == "processing":
        time.sleep(float(res.headers.get("retry-after", 5)))
        res = requests.get(location, headers=HEADERS)
        result = res.json()
    return json.dumps(result)

// partfinder-tools.mjs
const API = "https://api.partfinder.dev";
const headers = { Authorization: `Bearer ${process.env.PARTFINDER_API_KEY}` };

const doc = await (await fetch(`${API}/v1/openapi.json`)).json();

// Inline every "$ref" so each tool's schema stands alone.
const inline = (s) =>
  Array.isArray(s) ? s.map(inline)
  : s?.$ref ? inline(doc.components.schemas[s.$ref.split("/").pop()])
  : s && typeof s === "object" ? Object.fromEntries(Object.entries(s).map(([k, v]) => [k, inline(v)]))
  : s;

// One tool per operation. Keys and webhooks stay with your own code, and so does the file upload.
export const operations = [];
for (const [path, methods] of Object.entries(doc.paths)) {
  for (const [method, op] of Object.entries(methods)) {
    if (op.tags.includes("Keys") || op.tags.includes("Webhooks")) continue;
    const content = op.requestBody?.content;
    if (content && !content["application/json"]) continue;
    const params = (op.parameters ?? []).filter((p) => p.in !== "header");
    const body = inline(content?.["application/json"].schema) ?? {};
    operations.push({
      name: op.operationId,
      description: op.description ?? op.summary,
      method: method.toUpperCase(),
      path,
      bodyFields: Object.keys(body.properties ?? {}),
      parameters: {
        type: "object",
        properties: {
          ...Object.fromEntries(params.map((p) => [p.name, { ...p.schema, description: p.description }])),
          ...body.properties,
        },
        required: [...params.filter((p) => p.required).map((p) => p.name), ...(body.required ?? [])],
      },
    });
  }
}

const statusOf = (result) => result.status ?? result.bom?.status;

// Make the call a tool call names. A job answers 202: read it until it is no longer processing.
export async function callTool(name, args) {
  const op = operations.find((o) => o.name === name);
  let path = op.path;
  const query = new URLSearchParams();
  const body = {};
  for (const [key, value] of Object.entries(args ?? {})) {
    if (value == null) continue;
    if (path.includes(`{${key}}`)) path = path.replace(`{${key}}`, encodeURIComponent(value));
    else if (op.bodyFields.includes(key)) body[key] = value;
    else query.set(key, String(value));
  }
  const search = query.toString();
  let res = await fetch(`${API}${path}${search ? `?${search}` : ""}`, {
    method: op.method,
    headers: op.bodyFields.length ? { ...headers, "Content-Type": "application/json" } : headers,
    body: op.bodyFields.length ? JSON.stringify(body) : undefined,
  });
  let result = await res.json();
  const location = res.headers.get("location") ?? res.url;
  while (res.status === 202 || statusOf(result) === "processing") {
    await new Promise((r) => setTimeout(r, Number(res.headers.get("retry-after") ?? 5) * 1000));
    res = await fetch(location, { headers });
    result = await res.json();
  }
  return JSON.stringify(result);
}