Andi Search API：使用 Python 搜索并读取结果

Andi Search：GET|POST https://api.andiai.com/api/v1/search，请求头 x-api-key。format=context 可返回给 LLM 用的 markdown。

先看清成品

• mock 返回 results。
• 官方无效密钥 HTTP 401。
• 1 项 pytest 通过。

合规提醒：Andi 回答与搜索结果可能不完整，重要信息须打开原文核对。API 密钥仅放服务端；metrics.cost_dollars 反映计费，注意配额。遵守目标站点 robots/使用条款，勿用 fetch 做未授权批量抓取。本文不编造价格表。

运行说明：无正式 ANDI_API_KEY。A/B 用 mock；对官方 api.andiai.com 无效密钥实测 HTTP 401。C 为参数本地校验。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

107-andi-A-src/
├── andi_search.py
├── mock_andi.py
└── test_107a.py

第 1 步：客户端

#!/usr/bin/env python3
"""Andi Search API client: GET/POST /api/v1/search with x-api-key."""
from __future__ import annotations
import argparse, json, os, sys, urllib.error, urllib.parse, urllib.request

def search(base, key, q, *, format="json", limit=None, method="GET"):
    headers = {"x-api-key": key}
    if method.upper() == "POST":
        payload = {"q": q, "format": format}
        if limit is not None:
            payload["limit"] = limit
        req = urllib.request.Request(
            f"{base.rstrip('/')}/api/v1/search",
            data=json.dumps(payload).encode(),
            headers={**headers, "Content-Type": "application/json"},
            method="POST",
        )
    else:
        params = {"q": q, "format": format}
        if limit is not None:
            params["limit"] = str(limit)
        req = urllib.request.Request(
            f"{base.rstrip('/')}/api/v1/search?{urllib.parse.urlencode(params)}",
            headers=headers,
            method="GET",
        )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            raw = resp.read()
            ctype = resp.headers.get("Content-Type", "")
            if "markdown" in ctype or format == "context":
                return raw.decode()
            return json.loads(raw.decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e

def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("query")
    ap.add_argument("--base", default=os.environ.get("ANDI_BASE_URL", "https://api.andiai.com"))
    ap.add_argument("--key", default=os.environ.get("ANDI_API_KEY", ""))
    ap.add_argument("--format", default="json", choices=["json", "context"])
    ap.add_argument("--limit", type=int, default=None)
    ap.add_argument("--post", action="store_true")
    args = ap.parse_args(argv)
    if not args.key:
        print("设置 ANDI_API_KEY（API Console）", file=sys.stderr); return 2
    if not args.query.strip():
        print("q 不能为空", file=sys.stderr); return 2
    r = search(args.base, args.key, args.query, format=args.format, limit=args.limit, method="POST" if args.post else "GET")
    if isinstance(r, str):
        print(r)
    else:
        print(json.dumps(r, ensure_ascii=False, indent=2))
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Andi AI Search API (search + fetch)."""
from __future__ import annotations
import json
from urllib.parse import parse_qs, urlparse
from http.server import BaseHTTPRequestHandler, HTTPServer

API_KEY = "andi-test-key"

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _auth(self):
        return self.headers.get("x-api-key") == API_KEY
    def _send(self, code, obj, ctype="application/json"):
        b = obj if isinstance(obj, (bytes, bytearray)) else json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(b)))
        self.end_headers()
        self.wfile.write(b)
    def do_GET(self):
        if not self._auth():
            self._send(401, {"error": "Unauthorized"}); return
        u = urlparse(self.path)
        qs = parse_qs(u.query)
        if u.path == "/api/v1/search":
            q = (qs.get("q") or [""])[0]
            fmt = (qs.get("format") or ["json"])[0]
            if fmt == "context":
                md = f"---\nquery: {q}\n---\n\n# {q}\n\nMock context for LLM.\n"
                self._send(200, md.encode(), "text/markdown"); return
            self._send(200, {
                "results_type": "Search", "title": q, "answer": "",
                "results": [{"title": f"About {q}", "link": "https://example.com/a", "desc": "mock", "source": "example.com"}],
                "metrics": {"query": q, "cost_dollars": 0.01, "results_returned": 1},
            }); return
        if u.path == "/api/v1/fetch":
            url = (qs.get("url") or [""])[0]
            self._send(200, {"title": "Mock Page", "url": url, "word_count": 42, "content": "Clean extracted mock content."}); return
        self.send_error(404)
    def do_POST(self):
        if not self._auth():
            self._send(401, {"error": "Unauthorized"}); return
        if self.path.split("?")[0] != "/api/v1/search":
            self.send_error(404); return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        q = body.get("q") or ""
        self._send(200, {
            "results_type": "Search", "title": q,
            "results": [{"title": f"POST {q}", "link": "https://example.com/p", "desc": "mock", "source": "example.com"}],
            "metrics": {"query": q, "cost_dollars": 0.02, "results_returned": 1},
        })

def main():
    import argparse
    ap = argparse.ArgumentParser(); ap.add_argument("--port", type=int, default=9107)
    args = ap.parse_args()
    HTTPServer(("127.0.0.1", args.port), H).serve_forever()

if __name__ == "__main__":
    main()

第 3 步：测试

import subprocess, sys, time
from pathlib import Path
HERE = Path(__file__).resolve().parent

def test_search():
    port = 19170
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_andi.py"), "--port", str(port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run([sys.executable, str(HERE/"andi_search.py"), "RAG", "--base", f"http://127.0.0.1:{port}", "--key", "andi-test-key"], capture_output=True, text=True, check=True)
        assert "example.com" in r.stdout
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest mock search
.                                                                        [100%]
1 passed in 0.36s

$ invalid key api.andiai.com
HTTP 401
{"error":"Unauthorized","message":"Invalid API key"}

怎么检查结果？

• pytest 绿。
• 正式调用前先跑 C 篇参数校验。

调试示例

401 Invalid API key。 到 API Console 创建密钥，只显示一次；确认头名是 x-api-key 不是 Authorization。

常见问题

模板规则

• 密钥保密。
• 结果核对原文。
• 控制搜索成本。

项目资源

先搜得到源，再谈把 context 喂给模型。