OpenAI / Claude エージェントにリアルタイム Web 検索を 4 ステップで追加する

curl -X POST https://www.apipick.com/api/search/web \
  -H "x-api-key: pk_yourkey" \
  -H "Content-Type: application/json" \
  -d '{"query": "what is retrieval augmented generation"}'

curl https://www.apipick.com/api/search/web/tool-schema

from openai import OpenAI
import requests

client = OpenAI()
schema = requests.get("https://www.apipick.com/api/search/web/tool-schema").json()

assistant = client.beta.assistants.create(
    name="Research Agent",
    model="gpt-4o",
    instructions="Use web_search whenever the user asks about current events or anything that changes after your training cutoff.",
    tools=[{"type": "function", "function": schema["openai"]}],
)

import anthropic
import requests

schema = requests.get("https://www.apipick.com/api/search/web/tool-schema").json()
client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=[schema["claude"]],
    messages=[{"role": "user", "content": "Summarise this week's RAG research."}],
)

from langchain.tools import tool
import requests

@tool
def web_search(query: str) -> dict:
    """Real-time web search. Use for any question that needs current information."""
    return requests.post(
        "https://www.apipick.com/api/search/web",
        headers={"x-api-key": "pk_yourkey"},
        json={"query": query},
    ).json()

import json, time, requests
from openai import OpenAI

client = OpenAI()
KEY = "pk_yourkey"

def run_with_tool(thread_id: str, assistant_id: str):
    run = client.beta.threads.runs.create(thread_id=thread_id, assistant_id=assistant_id)
    while True:
        run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
        if run.status == "requires_action":
            outputs = []
            for call in run.required_action.submit_tool_outputs.tool_calls:
                if call.function.name == "web_search":
                    args = json.loads(call.function.arguments)
                    result = requests.post(
                        "https://www.apipick.com/api/search/web",
                        headers={"x-api-key": KEY},
                        json=args,
                    ).json()
                    outputs.append({"tool_call_id": call.id, "output": json.dumps(result)})
            client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id, run_id=run.id, tool_outputs=outputs,
            )
        elif run.status in ("completed", "failed", "cancelled", "expired"):
            return run
        time.sleep(0.5)