中文资讯掘金 AI Sun, 19 Jul 2026 正文 ok 775 words

SERP API + Claude 函数调用：从工具使用到 Agent 的完整实现

英文标题：SERP API + Claude function calling:从 tool use 到 agent 的完整实现

本文介绍如何结合 SerpBase 的 SERP API 与 Claude 的函数调用能力，构建一个能查询实时数据的 AI Agent。文章给出了完整的工具模式设计、调用循环、错误处理和上下文压缩方案，并分享了监控指标与实际运行效果，为开发者快速搭建可实现降级与重试的智能 Agent 提供了参考。

打开原文相关 8 · 机会 4

3.2 KB

589 B

5.4 KB

中文

场景

做一个 AI 助手，用户问“2026 年 SERP API 价格趋势”，助手需要查实时数据，不能只靠训练数据。下面是完整的工程实现，包含工具模式设计 + 调用 + 上下文管理。

1. 选择 API

我用 SerpBase 的 /google/search，原因：

返回结构化 JSON，不用解析 HTML
POST 单端点简单
失败自动退款（不浪费 credit）
1 credit / call，价格友好

2. 工具模式设计

Claude 函数调用的核心是工具描述。模式写得好，模型才知道何时调用。

python import anthropic

tools = [ { "name": "search_google", "description": """Search Google for real-time information. Use when:

The question asks about recent events (last 6 months)
The question asks about pricing, statistics, or current state
The question involves specific companies, products, or current facts
The user asks "what is the latest" or "current state"

Returns structured Google search results including title, link, snippet, and related questions. Max 5 organic results per call.""", "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query, in English, 2-6 words. Do not include date." }, "gl": { "type": "string", "description": "Country code for results. Examples: us, uk, cn, jp", "default": "us" }, "num": { "type": "integer", "description": "Number of results to return. 1-10.", "default": 5, "minimum": 1, "maximum": 10 } }, "required": ["query"] } } ]

说明中写得很具体：何时用、返回什么、限制。模型依靠这个判断何时调用工具。

3. 工具调用循环

python import requests

def execute_search(query: str, gl: str = "us", num: int = 5) -> str: """调 SerpBase 查 SERP，返回裁剪后的文本""" try: r = requests.post( "https://api.serpbase.dev/google/search", headers={"X-API-Key": SERPBASE_KEY}, json={"q": query, "gl": gl, "hl": "en", "page": 1, "num": num}, timeout=10, ) r.raise_for_status() data = r.json()

    results = []
    for item in data.get("organic", [])[:num]:
        results.append(
            f"[{item['rank']}] {item['title']}\n"
            f"    {item['link']}\n"
            f"    {item.get('snippet', '')}"
        )
    return "\n\n".join(results)
except requests.exceptions.Timeout:
    return f"SERP API timeout for query: {query}"
except requests.exceptions.HTTPError as e:
    return f"SERP API error: {e.response.status_code}"
except Exception as e:
    return f"SERP API failed: {str(e)}"

def run_agent(user_query: str, max_iterations: int = 3) -> str: """Run Claude agent with search tool""" client = anthropic.Anthropic() messages = [{"role": "user", "content": user_query}]

for _ in range(max_iterations):
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        tools=tools,
        messages=messages,
    )

    tool_uses = [b for b in response.content if b.type == "tool_use"]

    if not tool_uses:
        return next(b.text for b in response.content if b.type == "text")

    messages.append({"role": "assistant", "content": response.content})

    tool_results = []
    for tool_use in tool_uses:
        if tool_use.name == "search_google":
            result = execute_search(
                tool_use.input["query"],
                tool_use.input.get("gl", "us"),
                tool_use.input.get("num", 5),
            )
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": result,
            })

    messages.append({"role": "user", "content": tool_results})

return "Agent reached max iterations"

4. 关键设计

max_iterations 设为 3：防止 agent 死循环（Claude 一直调工具不出结论）。

tool_use_id 必填：每个 tool_result 必须关联到对应 tool_use，否则模型分不清。

工具失败返回文本而非抛出异常：模型能看到失败信息，可以调整策略（换个 query 重试）。

client 不要复用：Anthropic client 线程安全，但 messages 列表要保持顺序（append assistant → tool_result）。

5. 错误处理细节

SERP API 调用可能失败：

网络超时 → 返回 "SERP API timeout, query: X"
HTTP 4xx → 返回 status code
HTTP 5xx → 返回 status code
解析失败 → 返回 "SERP API failed: ..."
QPS 超限 → 返回 status code

模型收到这些错误后，会：

换 query 重试（对 5xx）
跳过这个问题（对 4xx）
给出降级答案（对所有失败）

不要让工具失败 throw 异常传到模型，那会中断整个 agent 循环。

6. 上下文压缩

agent 跑多轮后 messages 会很长（每次工具调用 + 结果都进 messages）。Sonnet 200K context 不怕长，但成本和延迟会上去。

python def compress_history(messages: list, keep_recent: int = 5) -> list: """保留最近 N 条，把早期工具结果压缩""" if len(messages) <= keep_recent: return messages

compressed = messages[:2]  

for msg in messages[2:-keep_recent]:
    if msg["role"] == "user" and isinstance(msg["content"], list):
        for block in msg["content"]:
            if block.get("type") == "tool_result":
                block["content"] = "[previous results omitted for brevity]"
    compressed.append(msg)

compressed.extend(messages[-keep_recent:])
return compressed

跑长 agent 时，在每 10 轮调一次 compress_history，省 token 30-50%。

7. 完整示例

python result = run_agent("2026 年 SERP API 的价格区间") print(result)

8. 监控指标

tool_call_rate：每次 session 调 SERP 几次（1-3 正常，5+ 可能死循环）
latency_p50 / p99：工具调用延迟（SERP API 1-1.5s + 网络 + Claude 处理）
failure_rate：工具失败比例（超 5% 查 vendor 状态）
cache_hit_rate：相同 query 5 分钟内复用次数

小结

SERP API + Claude 函数调用的关键不是调 API，而是：

工具模式写得具体，模型才知道何时调用
失败返文本不抛异常，模型可以降级
上下文压缩，长 agent 不爆 token
max_iterations 防死循环

SerpBase 的 auto-refund 帮了大忙：失败不扣 credit，放心调用。

实际项目里，我用这套模式搭建了 3 个 agent：

SEO 监控 agent（查 SERP 排名）
竞品分析 agent（查 SERP + 提取数据）
内容生成 agent（SERP PAA 当提示）

代码不到 200 行，跑了 3 个月没出问题。

English / 原文

场景

做一个 AI 助手,用户问"2026 年 SERP API 价格趋势",助手需要查实时数据,不能只靠训练数据。下面是完整的工程实现,包含 tool schema 设计 + 调用 + 上下文管理。

1. 选 API

我用 SerpBase 的 /google/search,原因:

返回结构化 JSON,不用解析 HTML
POST 单 endpoint 简单
失败自动 refund(不浪费 credit)
1 credit / call,价格友好

2. Tool Schema 设计

Claude function calling 的核心是工具描述。Schema 写得好,模型才知道何时调。

import anthropic

tools = [
    {
        "name": "search_google",
        "description": """Search Google for real-time information. Use when:
- The question asks about recent events (last 6 months)
- The question asks about pricing, statistics, or current state
- The question involves specific companies, products, or current facts
- The user asks "what is the latest" or "current state"

Returns structured Google search results including title, link, snippet, and related questions. Max 5 organic results per call.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query, in English, 2-6 words. Do not include date."
                },
                "gl": {
                    "type": "string",
                    "description": "Country code for results. Examples: us, uk, cn, jp",
                    "default": "us"
                },
                "num": {
                    "type": "integer",
                    "description": "Number of results to return. 1-10.",
                    "default": 5,
                    "minimum": 1,
                    "maximum": 10
                }
            },
            "required": ["query"]
        }
    }
]

description 写得具体:何时用、返回什么、限制。模型靠这个判断何时调工具。

3. Tool 调用循环

import requests

def execute_search(query: str, gl: str = "us", num: int = 5) -> str:
    """调 SerpBase 查 SERP,返回裁剪后的文本"""
    try:
        r = requests.post(
            "https://api.serpbase.dev/google/search",
            headers={"X-API-Key": SERPBASE_KEY},
            json={"q": query, "gl": gl, "hl": "en", "page": 1, "num": num},
            timeout=10,
        )
        r.raise_for_status()
        data = r.json()

        
        results = []
        for item in data.get("organic", [])[:num]:
            results.append(
                f"[{item['rank']}] {item['title']}\n"
                f"    {item['link']}\n"
                f"    {item.get('snippet', '')}"
            )
        return "\n\n".join(results)
    except requests.exceptions.Timeout:
        return f"SERP API timeout for query: {query}"
    except requests.exceptions.HTTPError as e:
        return f"SERP API error: {e.response.status_code}"
    except Exception as e:
        return f"SERP API failed: {str(e)}"

def run_agent(user_query: str, max_iterations: int = 3) -> str:
    """Run Claude agent with search tool"""
    client = anthropic.Anthropic()
    messages = [{"role": "user", "content": user_query}]

    for _ in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            tools=tools,
            messages=messages,
        )

        
        tool_uses = [b for b in response.content if b.type == "tool_use"]

        
        if not tool_uses:
            return next(b.text for b in response.content if b.type == "text")

        
        messages.append({"role": "assistant", "content": response.content})

        
        tool_results = []
        for tool_use in tool_uses:
            if tool_use.name == "search_google":
                result = execute_search(
                    tool_use.input["query"],
                    tool_use.input.get("gl", "us"),
                    tool_use.input.get("num", 5),
                )
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": result,
                })

        
        messages.append({"role": "user", "content": tool_results})

    return "Agent reached max iterations"

4. 关键设计

max_iterations 设 3 :防止 agent 死循环(Claude 一直调工具不出结论)。

tool_use_id 必填 :每个 tool_result 必须关联到对应 tool_use,否则模型分不清。

tool 失败返文本而不是抛异常 :模型看得到失败信息,可以调整策略(换个 query 重试)。

client 不要复用 :Anthropic client 线程安全,但 messages 列表要保持顺序(append assistant → tool_result)。

5. 错误处理细节

SERP API 调用可能失败:

网络超时 → 返 "SERP API timeout,query: X"
HTTP 4xx → 返 status code
HTTP 5xx → 返 status code
解析失败 → 返 "SERP API failed: ..."
QPS 超限 → 返 status code

模型收到这些错误后,会:

换 query 重试(对 5xx)
跳过这个问题(对 4xx)
给出降级答案(对所有失败)

不要让工具失败 throw 异常传到 model,那会中断整个 agent 循环。

6. 上下文压缩

agent 跑多轮后 messages 会很长(每次 tool 调 + result 都进 messages)。Sonnet 200K context 不怕长,但成本和延迟会上去。

def compress_history(messages: list, keep_recent: int = 5) -> list:
    """保留最近 N 条,把早期 tool 结果压缩"""
    if len(messages) <= keep_recent:
        return messages

    
    
    compressed = messages[:2]  

    for msg in messages[2:-keep_recent]:
        if msg["role"] == "user" and isinstance(msg["content"], list):
            for block in msg["content"]:
                if block.get("type") == "tool_result":
                    block["content"] = "[previous results omitted for brevity]"
        compressed.append(msg)

    compressed.extend(messages[-keep_recent:])
    return compressed

跑长 agent 时,在每 10 轮调一次 compress_history,省 token 30-50%。

7. 一个完整 demo

result = run_agent("2026 年 SERP API 的价格区间")




print(result)

8. 监控指标

tool_call_rate :每次 session 调 SERP 几次(1-3 正常,5+ 可能死循环)
latency_p50 / p99 :tool call 延迟(SERP API 1-1.5s + 网络 + Claude 处理)
failure_rate :tool 失败比例(超 5% 查 vendor 状态)
cache_hit_rate :相同 query 5 分钟内复用次数

小结

SERP API + Claude function calling 的关键不是调 API,而是:

Tool schema 写得具体,模型才知道何时调
失败返文本不抛异常,模型可以降级
上下文压缩,长 agent 不爆 token
max_iterations 防死循环

SerpBase 的 auto-refund 帮了大忙:失败不扣 credit,放心调。

实际项目里,我用这套模式搭了 3 个 agent:

SEO 监控 agent(查 SERP 排名)
竞品分析 agent(查 SERP + 提取数据)
内容生成 agent(SERP PAA 当提示)

代码不到 200 行,跑了3个月没出问题。