Finance MCP

Instructions

Implementation Reference

• finance_mcp/core/agent/react_agent_op.py:204-258 (handler)

  Core handler function implementing the ReAct agent loop: initializes tools and messages, performs iterative reasoning (LLM chat) and acting (parallel tool execution) steps until no more tool calls or max_steps reached.

  async def async_execute(self):
      """
      Main execution loop implementing the ReAct (Reasoning + Acting) pattern.
  
      The agent alternates between:
      1. Reasoning: Invoking the LLM to analyze the situation and decide on actions
      2. Acting: Executing the chosen tools and incorporating their results
  
      This loop continues until:
      - The agent decides no more tools are needed (final answer reached)
      - The maximum number of steps is reached
      """
      from .think_tool_op import ThinkToolOp
  
      # Initialize think tool operator if needed
      think_op = ThinkToolOp(language=self.language)
  
      # Build dictionary of available tool operators from context
      tool_op_dict = await self.build_tool_op_dict()
  
      # Optionally add think_tool to available tools
      if self.add_think_tool:
          tool_op_dict["think_tool"] = think_op
  
      # Initialize conversation message history
      messages = await self.build_messages()
  
      # Main ReAct loop: alternate between reasoning and acting
      for step in range(self.max_steps):
          # Reasoning step: LLM analyzes context and decides on actions
          assistant_message, should_continue = await self._reasoning_step(
              messages,
              tool_op_dict,
              step,
          )
  
          # If no tool calls, the agent has reached a final answer
          if not should_continue:
              break
  
          # Acting step: execute tools and collect results
          tool_result_messages = await self._acting_step(
              assistant_message,
              tool_op_dict,
              think_op,
              step,
          )
  
          # Append tool results to message history for next reasoning step
          messages.extend(tool_result_messages)
  
      # Set final output and store full conversation history in metadata
      self.set_output(messages[-1].content)
      self.context.response.metadata["messages"] = messages

• finance_mcp/core/agent/react_agent_op.py:33-46 (schema)

  Defines the tool's input schema (required 'query' string) and description for MCP tool calling.

  def build_tool_call(self) -> ToolCall:
      """Expose metadata describing how to invoke the agent."""
      return ToolCall(
          **{
              "description": "A React agent that answers user queries.",
              "input_schema": {
                  "query": {
                      "type": "string",
                      "description": "query",
                      "required": True,
                  },
              },
          },
      )

• finance_mcp/core/agent/react_agent_op.py:14-15 (registration)

  @C.register_op() decorator registers ReactAgentOp as an available operation/tool in the FlowLLM context, making it discoverable as 'react_agent'.

  @C.register_op()
  class ReactAgentOp(BaseAsyncToolOp):

• finance_mcp/core/agent/react_agent_op.py:58-79 (helper)

  Helper to construct initial LLM conversation messages from input query or messages list, adding system prompt with timestamp.

  async def build_messages(self) -> List[Message]:
      """Build the initial message history for the LLM."""
      if "query" in self.input_dict and self.input_dict["query"]:
          query: str = self.input_dict["query"]
          now_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
          messages = [
              Message(role=Role.SYSTEM, content=self.prompt_format(prompt_name="system_prompt", time=now_time)),
              Message(role=Role.USER, content=query),
          ]
          logger.info(f"round0.system={messages[0].model_dump_json()}")
          logger.info(f"round0.user={messages[1].model_dump_json()}")
  
      elif "messages" in self.input_dict:
          messages = self.input_dict["messages"]
          messages = [Message(**x) for x in messages]
  
          logger.info(f"round0.user={messages[-1].model_dump_json()}")
      else:
          raise ValueError("input_dict must contain either 'query' or 'messages'")
  
      return messages

• finance_mcp/core/agent/react_agent_op.py:48-57 (helper)

  Helper to build dictionary of available tool ops from context, keyed by tool_call.name, setting language.

  async def build_tool_op_dict(self) -> dict:
      """Collect available tool operators from the execution context."""
      assert isinstance(self.ops, BaseContext), "self.ops must be BaseContext"
      tool_op_dict: Dict[str, BaseAsyncToolOp] = {
          op.tool_call.name: op for op in self.ops.values() if isinstance(op, BaseAsyncToolOp)
      }
      for op in tool_op_dict.values():
          op.language = self.language
      return tool_op_dict