# SERP API trong Python: requests, httpx và asyncio

> Dùng SerpCrypto SERP API trong Python: một client requests có retry, httpx async với giới hạn concurrency, phân trang và xuất CSV. Code đã được kiểm thử.

- URL: https://serpcrypto.com/vi/docs/python/
- Cập nhật lần cuối: 2026-10-22
- Đơn vị phát hành: SerpCrypto (https://serpcrypto.com)

Một client tái sử dụng có retry, vòng lặp phân trang dừng khi hết kết quả, tìm kiếm async giới hạn theo concurrency của bạn, và file CSV hàng ngày cho thứ hạng. Code copy-paste, đã kiểm tra cú pháp, không cần cài package nào của chúng tôi.

**Trả lời nhanh — Làm sao gọi SERP API từ Python?**

Gửi GET request đến `https://serpcrypto.com/api/v1/search` bằng `requests` hoặc `httpx`, khóa của bạn trong header `Authorization: Bearer` và truy vấn trong `params`, rồi đọc `response.json()`. **SerpCrypto** tính 1 credit mỗi lần tìm kiếm thành công, từ 0,30 US$/1K lượt tìm kiếm, và retry các lần gọi thất bại là miễn phí. Thanh toán bằng crypto, không cần KYC. Đoạn code dưới đây chỉ cần `requests` hoặc `httpx`.

## Thiết lập Python trong nháy mắt

| Thuộc tính | Giá trị |
| --- | --- |
| Cài đặt | `pip install requests` hoặc `pip install httpx` |
| Khóa | Biến môi trường `SERPCRYPTO_API_KEY` |
| Header | `Authorization: Bearer <key>` |
| Timeout | 60 đến 90 giây: tìm kiếm chạy trực tiếp |
| Retry | Chỉ 429, 502 và 504, có backoff; lỗi thì miễn phí |
| Tìm kiếm song song | Tối đa `concurrency_limit` từ `GET /credits` |

## Làm sao thiết lập một client Python tái sử dụng?

Tạo một `requests.Session` với khóa trong header và chính sách retry cho ba lỗi tạm thời. Một session tái sử dụng kết nối, nên vòng lặp tìm kiếm tốn ít thời gian hơn cho TLS handshake. Lưu file là `serp_client.py`; các ví dụ sau sẽ import nó.

`Retry` từ urllib3 chờ lâu hơn trước mỗi lần thử mới và tuân theo header `Retry-After` nếu có. Nó không bao giờ retry `400`, `401` hay `402`, vì những lỗi này cần sửa chứ không phải thử lại.

*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())
```

## Làm sao thu thập nhiều hơn 10 kết quả?

Yêu cầu từng trang tiếp theo một. Khoảng 10 kết quả organic mỗi trang, đúng như Google đã trả về kể từ khi ngừng áp dụng num=100 vào tháng 9 năm 2025; mỗi trang thêm là thêm một credit. Dừng ngay khi thiếu `pagination.next_page`: không còn gì để lấy nữa, và bạn giữ được credit.

Với hầu hết công việc theo dõi thứ hạng, trang đầu hoặc ba trang đầu là đủ, vì click tập trung ở đó. Ba mươi kết quả cho một từ khóa tốn ba credit.

*Phân trang: tối đa ba trang, dừng sớm*

```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")
```

## Làm sao chạy nhiều tìm kiếm song song?

Dùng `httpx.AsyncClient` với `asyncio.Semaphore` được đặt theo giới hạn của tài khoản bạn. Script đọc `concurrency_limit` từ endpoint credit miễn phí, nên nó tự thích ứng khi support nâng giới hạn của bạn. Tìm kiếm vượt giới hạn dù sao cũng nhận `429`; semaphore tránh lãng phí những lần thử đó.

Mỗi task retry các lỗi tạm thời với khoảng chờ tăng dần và trả về `None` sau bốn lần thất bại, nên một truy vấn cứng đầu không bao giờ chặn cả batch.

*asyncio + httpx: tìm kiếm song song dưới giới hạn*

```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"]))
```

## Làm sao lưu thứ hạng hàng ngày vào file CSV?

Thêm một dòng cho mỗi kết quả, gồm ngày, từ khóa, vị trí, link và các cài đặt mà API đã áp dụng. Đọc `gl` và `device` từ `search_parameters` thay vì từ biến của riêng bạn nghĩa là file ghi lại những gì thực sự đã dùng.

Chạy script mỗi ngày một lần bằng cron hoặc Windows Task Scheduler. Ba từ khóa trên mobile tốn ba credit một ngày, khoảng 90 một tháng.

*Thêm thứ hạng hôm nay vào 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"],
            })
```

## Tôi nên dùng HTTP client Python nào?

Dùng `requests` cho script và cron job: đơn giản và adapter retry của nó đã chín muồi. Dùng `httpx` khi bạn cần code async, chẳng hạn web app, agent hoặc một lô hàng trăm từ khóa. Cả hai gửi cùng một request và nhận cùng một JSON; API không quan tâm bạn chọn cái nào.

Tránh đặt timeout quá ngắn với cả hai thư viện. Client từ bỏ sau năm giây có thể gửi lại một tìm kiếm vẫn đang chạy và trả tiền cho cả hai.

## Script của tôi sẽ tốn bao nhiêu?

Đếm số lần tìm kiếm, không đếm kết quả: một lần gọi là một credit khi thành công. Một job 500 từ khóa, mỗi từ khóa một trang, mỗi ngày trong 30 ngày, là 15.000 lượt tìm kiếm. Với pack 25.000 tìm kiếm (18,75 US$), vẫn còn dư chỗ, và Credit tìm kiếm trả trước không bao giờ hết hạn: các lượt tìm kiếm chưa dùng vẫn nằm trong tài khoản của bạn cho đến khi bạn dùng.

Kiểm tra số dư từ code bằng helper `credits()` ở trên; không tốn gì.

## Câu hỏi về Python

### Có package SerpCrypto trên PyPI không?

Không. Các ví dụ gọi endpoint HTTPS bằng `requests` hoặc `httpx`, hai thư viện được dùng rộng rãi, nên không có gì của chúng tôi để cài hay cập nhật. Client ở trên dài khoảng 40 dòng và bạn sở hữu nó.

### Đoạn code này hỗ trợ những phiên bản Python nào?

Python 3.9 trở lên. Ví dụ async dùng `asyncio.run`, có từ Python 3.7, phần còn lại chỉ dùng thư viện chuẩn cùng `requests` hoặc `httpx`.

### Tôi có thể dùng API từ Jupyter notebook không?

Có. Đặt `SERPCRYPTO_API_KEY` trước khi khởi động Jupyter, hoặc đọc khóa bằng `getpass` bên trong notebook để nó không bị lưu vào một cell đã lưu. Trong notebook, dùng `await main(...)` thay vì `asyncio.run(main(...))`.

### Làm sao chuyển kết quả thành pandas DataFrame?

Truyền danh sách các dictionary kết quả cho `pandas.DataFrame(data["organic_results"])`. Mỗi field thành một cột; các field tùy chọn bị thiếu như `date` thành ô trống.

---

SerpCrypto: SerpCrypto là SERP API cho lập trình viên và AI agent, trả về kết quả Google và Bing dưới dạng JSON, thanh toán bằng crypto mà không cần tải lên giấy tờ tùy thân. Credit tìm kiếm trả trước từ 0,30 US$/1K lượt tìm kiếm không bao giờ hết hạn. Không cần KYC — không ID, không selfie, không giấy tờ. Thanh toán bằng crypto.