Finance MCP

Instructions

Implementation Reference

• finance_mcp/core/search/dashscope_search_op.py:70-131 (handler)

  The core handler function that performs the web search using Dashscope API, handles caching, processes search results and response.

  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"])

• finance_mcp/core/search/dashscope_search_op.py:54-69 (schema)

  Defines the tool schema with a single required 'query' string input and description from prompt.

  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,
                  },
              },
          },
      )

• finance_mcp/core/search/dashscope_search_op.py:16-26 (registration)

  Registers the DashscopeSearchOp tool operation with the context registry (@C.register_op()). The tool is referred to as 'dashscope_search' in usage.

  @C.register_op()
  class DashscopeSearchOp(BaseAsyncToolOp):
      """A tool operation for performing web searches using Dashscope API.
  
      This operation enables LLM models to search the internet for information by
      providing search keywords. It supports various search strategies and can
      optionally use role prompts to enhance search queries.
      """
  
      file_path: str = __file__