# PythonでのSERP API: requests、httpx、asyncio > PythonでSerpCrypto SERP APIを使う: リトライ付きのrequestsクライアント、同時実行数を制限したasync httpx、ページネーション、CSVエクスポート。動作確認済みのコード。 - URL: https://serpcrypto.com/ja/docs/python/ - 最終更新: 2026-10-22 - 発行者: SerpCrypto (https://serpcrypto.com) リトライ付きの再利用可能なクライアント、結果が終わったら停止するページネーションループ、同時実行数の上限内に制限したasync検索、ランキング用の日次CSV。コピー&ペースト可能なコードで、構文チェック済み、当社のパッケージをインストールする必要はありません。 **簡潔な回答 — PythonからSERP APIを呼び出すには？** `https://serpcrypto.com/api/v1/search`にGETリクエストを送信し、`requests`または`httpx`を使い、`Authorization: Bearer`ヘッダーにキー、`params`に検索クエリを設定して、`response.json()`を読み取ります。**SerpCrypto**は成功した検索1回につき1クレジットを$0.30/1K検索から課金し、失敗した呼び出しのリトライは無料です。暗号資産で支払い、KYC不要。以下のコードは`requests`または`httpx`のみ必要です。 ## Pythonのセットアップ概要 | 項目 | 値 | | --- | --- | | インストール | `pip install requests`または`pip install httpx` | | キー | 環境変数`SERPCRYPTO_API_KEY` | | ヘッダー | `Authorization: Bearer ` | | タイムアウト | 60〜90秒: 検索はライブで実行されます | | リトライ | 429、502、504のみ、バックオフ付き。エラーは無料 | | 並列検索 | `GET /credits`から最大`concurrency_limit` | ## 再利用可能なPythonクライアントをセットアップするには？ ヘッダーにキーを設定し、3つの一時的エラーに対するリトライポリシーを持つ`requests.Session`を1つ作成します。セッションは接続を再利用するので、検索のループでTLSハンドシェイクに費やす時間が減ります。ファイルを`serp_client.py`として保存してください。次の例ではこれをインポートします。 urllib3の`Retry`は、各試行前に長く待機し、`Retry-After`ヘッダーが送信された場合はそれに従います。`400`、`401`、`402`はリトライしません。これらは再試行ではなく修正が必要です。 *serp_client.py* ```python """serp_client.py: a small, reusable client for the SerpCrypto SERP API (requests).""" import os import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry API_URL = "https://serpcrypto.com/api/v1/search" CREDITS_URL = "https://serpcrypto.com/api/v1/credits" def make_session(): session = requests.Session() session.headers["Authorization"] = "Bearer " + os.environ["SERPCRYPTO_API_KEY"] retry = Retry( total=4, backoff_factor=1.0, # waits about 1 s, 2 s, 4 s between attempts status_forcelist=(429, 502, 504), allowed_methods=("GET",), respect_retry_after_header=True, ) session.mount("https://", HTTPAdapter(max_retries=retry)) return session SESSION = make_session() def search(q, **params): """Run one search (1 credit if it succeeds) and return the parsed JSON.""" r = SESSION.get(API_URL, params={"q": q, **params}, timeout=90) r.raise_for_status() return r.json() def credits(): """Free: credits left, used today and the concurrency limit.""" r = SESSION.get(CREDITS_URL, timeout=30) r.raise_for_status() return r.json() if __name__ == "__main__": data = search("project management software", gl="us", hl="en") for item in data.get("organic_results", []): print(item["position"], item["link"]) print(credits()) ``` ## 10件を超える結果を取得するには？ 次のページを1つずつ要求します。1ページあたり約10件のオーガニック結果です。Googleは2025年9月にnum=100を尊重しなくなって以降、この件数を返しています。追加の各ページにはさらに1クレジットが必要です。`pagination.next_page`が欠落したらすぐに停止します。これ以上取得するものはなく、クレジットを節約できます。 ほとんどのランキング追跡作業では、1ページ目または最初の3ページで十分です。クリックがそこに集中するためです。1つのキーワードで30件の結果は3クレジットかかります。 *ページネーション: 最大3ページ、早期停止* ```python from serp_client import search def top_results(q, pages=3, **params): """Collect organic results over several pages: each page is 1 credit.""" rows = [] for page in range(1, pages + 1): data = search(q, page=page, **params) for item in data.get("organic_results", []): rows.append({"page": page, "position": item["position"], "title": item["title"], "link": item["link"]}) if not data.get("pagination", {}).get("next_page"): break # no further page: stop early and save credits return rows rows = top_results("crm for small business", pages=3, gl="gb", hl="en") print(len(rows), "results") ``` ## 多数の検索を並列で実行するには？ `httpx.AsyncClient`を`asyncio.Semaphore`とともに使用し、アカウントの制限に合わせてサイズを設定します。スクリプトは無料クレジットのエンドポイントから`concurrency_limit`を読み取るので、サポートが制限を引き上げるとそれに適応します。制限を超える検索はどのみち`429`になるため、セマフォでそれらの試行を無駄にしません。 各タスクは一時的エラーを徐々に長くなる間隔でリトライし、4回失敗すると`None`を返すので、1つの厄介なクエリがバッチをブロックすることはありません。 *asyncio + httpx: 制限内での並列検索* ```python """Run many searches in parallel without exceeding the account's concurrency limit.""" import asyncio import os import httpx API_URL = "https://serpcrypto.com/api/v1/search" CREDITS_URL = "https://serpcrypto.com/api/v1/credits" HEADERS = {"Authorization": "Bearer " + os.environ["SERPCRYPTO_API_KEY"]} async def search_one(client, gate, q): async with gate: # at most N searches at the same time for attempt in range(4): r = await client.get(API_URL, params={"q": q, "gl": "us", "hl": "en"}) if r.status_code in (429, 502, 504): await asyncio.sleep(2 ** attempt) continue r.raise_for_status() return q, r.json() return q, None async def main(queries): async with httpx.AsyncClient(headers=HEADERS, timeout=90) as client: limits = (await client.get(CREDITS_URL)).json() gate = asyncio.Semaphore(limits["concurrency_limit"]) results = await asyncio.gather(*(search_one(client, gate, q) for q in queries)) for q, data in results: first = (data or {}).get("organic_results", [{}])[0].get("link") print(f"{q!r}: {first}") asyncio.run(main(["best vpn router", "home espresso machine", "4k monitor for coding"])) ``` ## 日次ランキングをCSVファイルに保存するには？ 結果ごとに1行を追加し、日付、キーワード、順位、リンク、APIが適用した設定を記録します。独自の変数ではなく`search_parameters`から`gl`と`device`を読み取ることで、ファイルには実際に使用された内容が記録されます。 cronまたはWindowsタスクスケジューラで1日1回スクリプトを実行します。モバイルで3つのキーワードは1日3クレジット、月約90クレジットです。 *本日のランキングをrankings.csvに追加* ```python import csv from datetime import date from serp_client import search KEYWORDS = ["standing desk", "ergonomic chair", "monitor arm"] FIELDS = ["day", "keyword", "position", "title", "link", "gl", "device"] with open("rankings.csv", "a", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=FIELDS) if fh.tell() == 0: writer.writeheader() for kw in KEYWORDS: data = search(kw, gl="gb", hl="en", device="mobile") applied = data["search_parameters"] for item in data.get("organic_results", []): writer.writerow({ "day": date.today().isoformat(), "keyword": kw, "position": item["position"], "title": item["title"], "link": item["link"], "gl": applied["gl"], "device": applied["device"], }) ``` ## どのPython HTTPクライアントを使うべきですか？ スクリプトやcronジョブには`requests`を使いましょう。シンプルで、リトライアダプターが成熟しています。非同期コードが必要な場合、たとえばWebアプリ、エージェント、数百キーワードのバッチ処理には`httpx`を使いましょう。どちらも同じリクエストを送り、同じJSONを受け取ります。APIはどちらを選んでも関係ありません。 どちらのライブラリでも、極端に短いタイムアウトは避けてください。5秒で諦めるクライアントは、まだ実行中だった検索を再送してしまい、二重に支払う可能性があります。 ## スクリプトのコストはいくらになりますか？ 結果ではなく検索回数を数えます。成功した1回の呼び出しが1クレジットです。500キーワードを1ページずつ、毎日30日間実行するジョブは15,000回の検索になります。25,000検索パック（$18.75）なら余裕があり、プリペイドの検索クレジットに有効期限はありません。未使用の検索は、使用するまでアカウントに残ります。 コードからは上の`credits()`ヘルパーで残高を確認できます。これは無料です。 ## Pythonに関する質問 ### PyPIにSerpCryptoのパッケージはありますか？ いいえ。サンプルは広く使われている2つのライブラリ、`requests`または`httpx`でHTTPS endpointを呼び出すので、インストールや更新が必要なものはありません。上のクライアントは約40行で、あなた自身が管理できます。 ### このコードはどのPythonバージョンに対応していますか？ Python 3.9以降です。asyncのサンプルはPython 3.7から利用可能な`asyncio.run`を使い、それ以外は標準ライブラリと`requests`または`httpx`のみを使います。 ### Jupyter notebookからAPIを使えますか？ はい。Jupyterを起動する前に`SERPCRYPTO_API_KEY`を設定するか、notebook内で`getpass`を使ってキーを読み込めば、保存されたセルに残りません。notebook内では`asyncio.run(main(...))`の代わりに`await main(...)`を使ってください。 ### 結果をpandas DataFrameに変換するには？ 結果の辞書のリストを`pandas.DataFrame(data["organic_results"])`に渡します。各フィールドが列になり、`date`のような省略可能なフィールドが欠けている場合は空のセルになります。 --- SerpCrypto: SerpCryptoは、開発者とAIエージェント向けのSERP APIで、GoogleとBingの検索結果をJSONで返し、身分証明書のアップロードなしで暗号資産で支払えます。前払いの検索クレジットは$0.30/1K検索から購入でき、有効期限はありません。KYC不要 — 身分証明書もセルフィーも書類も不要。暗号資産で支払い。