identity:
name: google_search
author: YourName
label:
en_US: Google Search
ja_JP: Google検索
description:
human:
en_US: Search Google using SerpApi
ja_JP: SerpApiを使用してGoogle検索を行います
llm: A tool for performing a Google search.
parameters:
- name: query
type: string
required: true
label:
en_US: Query
ja_JP: 検索クエリ
form: llm # LLMが自動的に入力するパラメータ
extra:
python:
source: tools/google_search.py
3. 認証ロジックの実装 (provider/google.py)
ユーザーが入力したAPIキーが正しいかどうかを検証するロジックを実装します。
from typing import Any
from dify_plugin import ToolProvider
from dify_plugin.errors.tool import ToolProviderCredentialValidationError
from tools.google_search import GoogleSearchTool
class GoogleProvider(ToolProvider):
def _validate_credentials(self, credentials: dict[str, Any]) -> None:
try:
# テストクエリを実行してAPIキーの有効性を確認
for _ in GoogleSearchTool.from_credentials(credentials).invoke(
tool_parameters={"query": "test", "result_type": "link"},
):
pass
except Exception as e:
raise ToolProviderCredentialValidationError(str(e))
4. ツールロジックの実装 (tools/google_search.py)
実際にSerpApiを呼び出し、検索結果を整形して返すメインの処理です。
from collections.abc import Generator
from typing import Any
import requests
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage
SERP_API_URL = "https://serpapi.com/search"
class GoogleSearchTool(Tool):
def _parse_response(self, response: dict) -> dict:
result = {}
if "knowledge_graph" in response:
result["title"] = response["knowledge_graph"].get("title", "")
result["description"] = response["knowledge_graph"].get("description", "")
if "organic_results" in response:
result["organic_results"] = [
{
"title": item.get("title", ""),
"link": item.get("link", ""),
"snippet": item.get("snippet", ""),
}
for item in response["organic_results"]
]
return result
def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
params = {
"api_key": self.runtime.credentials["serpapi_api_key"],
"q": tool_parameters["query"],
"engine": "google",
"google_domain": "google.com",
"gl": "us",
"hl": "en",
}
response = requests.get(url=SERP_API_URL, params=params, timeout=5)
response.raise_for_status()
valuable_res = self._parse_response(response.json())
yield self.create_json_message(valuable_res)
コメント