OpenAI MCP Server
by arthurcolle
- claude_code
- lib
- tools
#!/usr/bin/env python3
# claude_code/lib/tools/search_tools.py
"""Web search and information retrieval tools."""
import os
import logging
import json
import urllib.parse
import requests
from typing import Dict, List, Optional, Any
from .base import tool, ToolRegistry
logger = logging.getLogger(__name__)
@tool(
name="WebSearch",
description="Search the web for information using various search engines",
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"engine": {
"type": "string",
"description": "Search engine to use (google, bing, duckduckgo)",
"enum": ["google", "bing", "duckduckgo"]
},
"num_results": {
"type": "integer",
"description": "Number of results to return (max 10)"
}
},
"required": ["query"]
},
category="search"
)
def web_search(query: str, engine: str = "google", num_results: int = 5) -> str:
"""Search the web for information.
Args:
query: Search query
engine: Search engine to use
num_results: Number of results to return
Returns:
Search results as formatted text
"""
logger.info(f"Searching web for: {query} using {engine}")
# Validate inputs
if num_results > 10:
num_results = 10 # Cap at 10 results
# Get API key based on engine
api_key = None
if engine == "google":
api_key = os.getenv("GOOGLE_SEARCH_API_KEY")
cx = os.getenv("GOOGLE_SEARCH_CX")
if not api_key or not cx:
return "Error: Google Search API key or CX not configured. Please set GOOGLE_SEARCH_API_KEY and GOOGLE_SEARCH_CX environment variables."
elif engine == "bing":
api_key = os.getenv("BING_SEARCH_API_KEY")
if not api_key:
return "Error: Bing Search API key not configured. Please set BING_SEARCH_API_KEY environment variable."
# Perform search based on engine
try:
if engine == "google":
return _google_search(query, api_key, cx, num_results)
elif engine == "bing":
return _bing_search(query, api_key, num_results)
elif engine == "duckduckgo":
return _duckduckgo_search(query, num_results)
else:
return f"Error: Unsupported search engine: {engine}"
except Exception as e:
logger.exception(f"Error during web search: {str(e)}")
return f"Error performing search: {str(e)}"
def _google_search(query: str, api_key: str, cx: str, num_results: int) -> str:
"""Perform Google search using Custom Search API."""
url = "https://www.googleapis.com/customsearch/v1"
params = {
"key": api_key,
"cx": cx,
"q": query,
"num": min(num_results, 10)
}
response = requests.get(url, params=params)
if response.status_code != 200:
return f"Error: Google search failed with status code {response.status_code}: {response.text}"
data = response.json()
if "items" not in data:
return f"No results found for '{query}'"
results = []
for i, item in enumerate(data["items"], 1):
title = item.get("title", "No title")
link = item.get("link", "No link")
snippet = item.get("snippet", "No description").replace("\n", " ")
results.append(f"{i}. {title}\n URL: {link}\n {snippet}\n")
return f"Google Search Results for '{query}':\n\n" + "\n".join(results)
def _bing_search(query: str, api_key: str, num_results: int) -> str:
"""Perform Bing search using Bing Web Search API."""
url = "https://api.bing.microsoft.com/v7.0/search"
headers = {"Ocp-Apim-Subscription-Key": api_key}
params = {
"q": query,
"count": min(num_results, 10),
"responseFilter": "Webpages"
}
response = requests.get(url, headers=headers, params=params)
if response.status_code != 200:
return f"Error: Bing search failed with status code {response.status_code}: {response.text}"
data = response.json()
if "webPages" not in data or "value" not in data["webPages"]:
return f"No results found for '{query}'"
results = []
for i, item in enumerate(data["webPages"]["value"], 1):
title = item.get("name", "No title")
link = item.get("url", "No link")
snippet = item.get("snippet", "No description").replace("\n", " ")
results.append(f"{i}. {title}\n URL: {link}\n {snippet}\n")
return f"Bing Search Results for '{query}':\n\n" + "\n".join(results)
def _duckduckgo_search(query: str, num_results: int) -> str:
"""Perform DuckDuckGo search using their API."""
# DuckDuckGo doesn't have an official API, but we can use their instant answer API
url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": 1,
"skip_disambig": 1
}
response = requests.get(url, params=params)
if response.status_code != 200:
return f"Error: DuckDuckGo search failed with status code {response.status_code}: {response.text}"
data = response.json()
results = []
# Add the abstract if available
if data.get("Abstract"):
results.append(f"Summary: {data['Abstract']}\n")
# Add related topics
if data.get("RelatedTopics"):
topics = data["RelatedTopics"][:num_results]
for i, topic in enumerate(topics, 1):
if "Text" in topic:
text = topic.get("Text", "No description")
url = topic.get("FirstURL", "No URL")
results.append(f"{i}. {text}\n URL: {url}\n")
if not results:
return f"No results found for '{query}'"
return f"DuckDuckGo Search Results for '{query}':\n\n" + "\n".join(results)
@tool(
name="WikipediaSearch",
description="Search Wikipedia for information on a topic",
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The topic to search for"
},
"language": {
"type": "string",
"description": "Language code (e.g., 'en', 'es', 'fr')",
"default": "en"
}
},
"required": ["query"]
},
category="search"
)
def wikipedia_search(query: str, language: str = "en") -> str:
"""Search Wikipedia for information on a topic.
Args:
query: Topic to search for
language: Language code
Returns:
Wikipedia article summary
"""
logger.info(f"Searching Wikipedia for: {query} in {language}")
try:
# Wikipedia API endpoint
url = f"https://{language}.wikipedia.org/api/rest_v1/page/summary/{urllib.parse.quote(query)}"
response = requests.get(url)
if response.status_code != 200:
# Try search API if direct lookup fails
search_url = f"https://{language}.wikipedia.org/w/api.php"
search_params = {
"action": "query",
"list": "search",
"srsearch": query,
"format": "json"
}
search_response = requests.get(search_url, params=search_params)
if search_response.status_code != 200:
return f"Error: Wikipedia search failed with status code {search_response.status_code}"
search_data = search_response.json()
if "query" not in search_data or "search" not in search_data["query"] or not search_data["query"]["search"]:
return f"No Wikipedia articles found for '{query}'"
# Get the first search result
first_result = search_data["query"]["search"][0]
title = first_result["title"]
# Get the summary for the first result
url = f"https://{language}.wikipedia.org/api/rest_v1/page/summary/{urllib.parse.quote(title)}"
response = requests.get(url)
if response.status_code != 200:
return f"Error: Wikipedia article lookup failed with status code {response.status_code}"
data = response.json()
# Format the response
title = data.get("title", "Unknown")
extract = data.get("extract", "No information available")
url = data.get("content_urls", {}).get("desktop", {}).get("page", "")
result = f"Wikipedia: {title}\n\n{extract}\n"
if url:
result += f"\nSource: {url}"
return result
except Exception as e:
logger.exception(f"Error during Wikipedia search: {str(e)}")
return f"Error searching Wikipedia: {str(e)}"
def register_search_tools(registry: ToolRegistry) -> None:
"""Register all search tools with the registry.
Args:
registry: Tool registry to register with
"""
from .base import create_tools_from_functions
search_tools = [
web_search,
wikipedia_search
]
create_tools_from_functions(registry, search_tools)