tavily_search
Search the internet for financial information to support research and analysis, including stock data, fund entities, and structured investigations.
Instructions
Use search keywords to retrieve relevant information from the internet.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | search keyword |
Input Schema (JSON Schema)
{
"properties": {
"query": {
"description": "search keyword",
"type": "string"
}
},
"required": [
"query"
],
"type": "object"
}
Implementation Reference
- The handler function that executes the tavily_search tool. It performs a web search using the Tavily API client, handles caching, optional content extraction with character limits, and outputs the results as JSON.async def async_execute(self): """Execute the Tavily web search for the given query. The query is read from ``input_dict['query']`` and the result is either the raw Tavily search output or a post-processed mapping with optional extracted content, depending on ``enable_extract``. """ query: str = self.input_dict["query"] logger.info(f"tavily.query: {query}") if self.enable_cache: cached_result = self.cache.load(query) if cached_result: self.set_output(json.dumps(cached_result, ensure_ascii=False, indent=2)) return response = await self.client.search(query=query) logger.info(f"tavily.response: {response}") if not self.enable_extract: # 如果不需要 extract，直接返回 search 的结果 if not response.get("results"): raise RuntimeError("tavily return empty result") final_result = {item["url"]: item for item in response["results"]} if self.enable_cache and final_result: self.cache.save(query, final_result, expire_hours=self.cache_expire_hours) self.set_output(json.dumps(final_result, ensure_ascii=False, indent=2)) return # enable_extract=True 时的原有逻辑 url_info_dict = {item["url"]: item for item in response["results"]} response_extract = await self.client.extract(urls=[item["url"] for item in response["results"]]) logger.info(f"tavily.response_extract: {response_extract}") final_result = {} all_char_count = 0 for item in response_extract["results"]: url = item["url"] raw_content: str = item["raw_content"] if len(raw_content) > self.item_max_char_count: raw_content = raw_content[: self.item_max_char_count] if all_char_count + len(raw_content) > self.all_max_char_count: raw_content = raw_content[: self.all_max_char_count - all_char_count] if raw_content: final_result[url] = url_info_dict[url] final_result[url]["raw_content"] = raw_content all_char_count += len(raw_content) if not final_result: raise RuntimeError("tavily return empty result") if self.enable_cache and final_result: self.cache.save(query, final_result, expire_hours=self.cache_expire_hours) self.set_output(json.dumps(final_result, ensure_ascii=False, indent=2))
- Defines the tool schema with a description and input schema requiring a single 'query' string parameter.def build_tool_call(self) -> ToolCall: """Build the tool call schema for the Tavily web search tool.""" return ToolCall( **{ "description": "Use search keywords to retrieve relevant information from the internet.", "input_schema": { "query": { "type": "string", "description": "search keyword", "required": True, }, }, }, )
- finance_mcp/core/search/tavily_search_op.py:17-21 (registration)Registers the TavilySearchOp class as an MCP tool operation using the @C.register_op() decorator. This is the tavily_search tool implementation.@C.register_op() class TavilySearchOp(BaseAsyncToolOp): """Asynchronous web search operation backed by the Tavily API.""" file_path: str = __file__
- Helper property that lazily initializes and provides the Tavily AsyncTavilyClient instance used for search operations.@property def client(self): """Get or create the Tavily async client instance. Returns: AsyncTavilyClient: The Tavily async client instance. """ if self._client is None: from tavily import AsyncTavilyClient self._client = AsyncTavilyClient(api_key=os.environ.get("TAVILY_API_KEY", "")) return self._client
- finance_mcp/core/search/__init__.py:13-17 (registration)Imports the TavilySearchOp for use in the search package, facilitating its registration and exposure.from .tavily_search_op import TavilySearchOp __all__ = [ "DashscopeSearchOp", "TavilySearchOp",