OpenAI Agents SDK — Hosted tools（WebSearchTool）をStreamlitに組み込む

LLM アプリで「最新の天気」「今日のニュース」のような質問には、モデル単体の知識だけでは足りません。OpenAI Agents SDK では Hosted tools を使うと、OpenAI 側で動く組み込みツール（Web 検索など）をエージェントに渡せます。

今回は Hosted tools の WebSearchTool を main.py に載せ、検索中の UI 表示 と SQLiteSession への記録 まで一通り扱います。

---

Hosted tools とは

Agents SDK のツールは大きく分かれます。

種類	実行場所	例
Hosted tools	OpenAI サーバー	WebSearchTool, FileSearchTool, CodeInterpreterTool
Function tools	自分の Python コード	@function_tool で包んだ関数
Agents as tools	別 Agent を tool 化	agent.as_tool(...)

Hosted tools = インフラを自分で用意せず、OpenAI がホストするツールを Agent の tools=[...] に渡す方式です。

公式ドキュメント より:

The WebSearchTool lets an agent search the web.

ノートブックで試した get_weather のような function_tool が「自分で書いた関数」なら、WebSearchTool() は Web 検索 API を SDK が用意 してくれます。

---

WebSearchTool の最小構成

from agents import Agent, WebSearchTool

agent = Agent(
    name="Assistant",
    instructions="""
    あなたはChatGPTのようなAIアシスタントです。

    あなたは以下のツールにアクセスできます:
    - WebSearchTool: インターネットから情報を検索するツール
    """,
    tools=[WebSearchTool()],
)

instructions に いつ Web 検索を使うか を書いておくと、モデルが tool を選びやすくなります。

オプション（公式）:

• filters — 検索対象の絞り込み
• user_location — 地域に応じた検索
• search_context_size — 検索コンテキスト量

今回は引数なしの WebSearchTool() から始めます。

---

Streamlit との全体構成

ユーザー入力（chat_input）
        ↓
Runner.run_streamed(agent, message, session=session)
        ↓
モデルが WebSearchTool を呼ぶ（Hosted）
        ↓
raw_response_event → status 更新 + delta ストリーミング
        ↓
SQLiteSession に user / web_search_call / assistant が保存
        ↓
paint_history + サイドバー Memory で表示

レイヤー	役割
WebSearchTool	Web 検索（Hosted）
Runner.run_streamed	エージェント実行 + ストリーム
update_status	検索・回答の進行表示
SQLiteSession	会話 + tool 呼び出し履歴の永続化

---

ストリーミング — raw_response_event で status を更新

Web 検索付きの実行では、raw_response_event の event.data.type に Hosted tool の状態が流れてきます。

def update_status(status_container, event: str) -> None:
    status_messages = {
        "response.web_search_call.in_progress": ("🔍️ Starting web search", "running"),
        "response.web_search_call.searching": ("🔍️ Web search in progress", "running"),
        "response.web_search_call.completed": ("✅ Web Search Completed", "complete"),
        "response.completed": ("✅ Response Completed", "complete"),
    }

    if event in status_messages:
        label, state = status_messages[event]
        status_container.update(label=label, state=state)


async def run_agent(message: str) -> str:
    with st.chat_message("assistant"):
        status_container = st.status("⏳", expanded=False)
        text_placeholder = st.empty()
        response = ""

        stream = Runner.run_streamed(agent, message, session=session)
        async for event in stream.stream_events():
            if event.type == "raw_response_event":
                update_status(status_container, event.data.type)
                if event.data.type == "response.output_text.delta":
                    response += event.data.delta
                    text_placeholder.write(response)

    return response

event.data.type	UI
response.web_search_call.in_progress	検索開始
response.web_search_call.searching	検索中
response.web_search_call.completed	検索完了
response.output_text.delta	回答テキストのストリーミング

function_tool 名をパースするより、API イベント型にマップする 方が Hosted tools と相性が良いです。

---

SQLiteSession — web_search_call の記録

Web 検索後、get_items() には role のない item も含まれます。

{
  "type": "web_search_call",
  "action": {
    "type": "search",
    "query": "weather: Busan, South Korea"
  },
  "status": "completed"
}

message["role"] 前提のコードは KeyError: 'role' になります。

安全な読み取り

def web_search_query(message: dict) -> str | None:
    if message.get("type") != "web_search_call":
        return None
    action = message.get("action") or {}
    return action.get("query")


def message_text(message: dict) -> str | None:
    role = message.get("role")

    if role == "user":
        return message.get("content")

    if role == "assistant" and message.get("type") == "message":
        content = message.get("content")
        if isinstance(content, list) and content:
            return content[0].get("text")

    return None

履歴の描画

def render_history(messages: list[dict]) -> None:
    for message in messages:
        query = web_search_query(message)
        if query:
            with st.chat_message("assistant"):
                st.caption(f"Searched the web for: {query}")
            continue

        text = message_text(message)
        role = message.get("role")
        if text is None or role not in ("user", "assistant"):
            continue

        with st.chat_message(role):
            st.write(text)

item の type	表示
user	ユーザーの質問
web_search_call	Searched the web for: ...
assistant + message	AI の回答

---

実行画像

関連記事（シリーズ）

• 参考: Tools — Hosted tools（公式）