dashscope_search
Search the internet for financial information to support research and analysis, enabling structured investigations and entity extraction for stocks and funds.
Instructions
Use search keywords to retrieve relevant information from the internet. If you have multiple keywords, please call this tool separately for each one.
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
- finance_mcp/core/search/dashscope_search_op.py:16-17 (registration)Registers DashscopeSearchOp as an MCP tool operation using the @C.register_op decorator. This is the primary registration point for the 'dashscope_search' tool.@C.register_op() class DashscopeSearchOp(BaseAsyncToolOp):
- Defines the input schema for the tool: a required string 'query' parameter.def build_tool_call(self) -> ToolCall: """Build the tool call schema for the Dashscope search op.""" return ToolCall( **{ "description": self.get_prompt("tool_description"), "input_schema": { "query": { "type": "string", "description": "search keyword", "required": True, }, }, }, )
- The main handler function that performs the Dashscope web search by calling the Dashscope AioGeneration API with enable_search=True, extracts search results and response content, handles caching, and sets the output.async def async_execute(self): """Execute a Dashscope web search for the given query. The method prepares the user query (optionally using a role prompt), calls the Dashscope generation API with search enabled, and returns the natural-language answer produced by the model. """ query: str = self.input_dict["query"] if self.enable_cache: cached_result = self.cache.load(query) if cached_result: self.set_output(cached_result["response_content"]) return if self.enable_role_prompt: user_query = self.prompt_format(prompt_name="role_prompt", query=query) else: user_query = query logger.info(f"user_query={user_query}") messages: list = [{"role": "user", "content": user_query}] import dashscope response = await dashscope.AioGeneration.call( api_key=self.api_key, model=self.model, messages=messages, enable_search=True, # Enable web search search_options={ "forced_search": True, # Force web search "enable_source": True, # Include search source information "enable_citation": False, # Enable citation markers "search_strategy": self.search_strategy, # Search strategy }, result_format="message", ) search_results = [] response_content = "" if hasattr(response, "output") and response.output: if hasattr(response.output, "search_info") and response.output.search_info: search_results = response.output.search_info.get("search_results", []) if hasattr(response.output, "choices") and response.output.choices and len(response.output.choices) > 0: response_content = response.output.choices[0].message.content final_result = { "query": query, "search_results": search_results, "response_content": response_content, "model": self.model, "search_strategy": self.search_strategy, } if self.enable_cache: self.cache.save(query, final_result, expire_hours=self.cache_expire_hours) self.set_output(final_result["response_content"])