Practice and Insights Behind Building and Operating OpenManus and MCP Agents

Background

I wrote an article before introducing the OpenManus project, which mainly runs the examples provided in the OpenManus project through locally built open source model services. With in-depth thinking on engineering design in OpenManus project and the minimum MVP test introduced into our operation and maintenance scenarios, we found that there is still a lot of work to be done before the actual scenario is available. After implementing the business operation and maintenance scenarios for a period of time, we will sort it out again and record the practice and thinking of the OpenManus project and MCP construction operation and maintenance Agent.

Technical Analysis

OpenManus is a multi-agent system whose architecture contains multiple modular agents, such as planning, execution, and tool calls, each performs its own responsibilities and works together. It adopts a hierarchical architecture design and is clearly divided into different intelligent modules such as planning, execution, and tool call. The following are:

• Planning Agent: Responsible for task plugging and process arrangement;
• Execution Agent: Conduct code generation and tool calls;
• ToolCallAgent: Responsible for performing specific operations such as browsing web pages, file management, etc.

When talking about agents, they need to take a closer look at their basic elements. The AI ​​Agent definition that is generally familiar to everyone is the combination of "planning + memory + tool call + execution", where LLM serves as the core controller of the system.

• Planning(Planning)
  : Disassembly, generate subtasks, and self-reflection;
• Memory(Memory)
  : Manage short-term and long-term memory, provide a context for the problem, and can act on complex task scenarios;
• Tool Usage)
  : Enhance the Agent's ability by calling external tools and supplement additional information;
• Executor
  : Obtain the tool's return results through the execution tool, and then feed it back to the Agent for analysis or summary.

1. Planning mechanism

OpenManus Planning Process: User input task request -> Create initialization plan -> Decompose the task into executable steps -> Create structured plan. Among them, the function of the PlanningTool class is to create and manage tasks, generate linear plans, and assign tasks to the corresponding agents.

When the user enters a task request, the _create_initial_plan method will be called. LLM and PlanningTool will be called to generate an initial plan. The plan includes task goals, step decomposition and status tracking. PlanningTool breaks the task into executable linear steps and generates a structured plan (objectives, steps, states).

curl --location 'http://localhost:11434/api/chat' \
--header 'Content-Type: application/json' \
--data "content": "You are a planning assistant. Your task is to create a detailed plan with clear "Content": "Create a detailed plan to accomplish this task: I want to rent a one-bedroom apartment near Xinjiangwan City, Yangpu District, Shanghai. The rent is not more than 8,000 yuan, 2 bedrooms and one living room, more than 90 square meters, and the community environment is good. Please recommend it to me."
                "content": "Create a detailed plan to accomplish this task: I want to rent a one-bedroom apartment near Xinjiangwan City, Yangpu District, Shanghai. The rent is not more than 8,000 yuan, 2 bedrooms and one living room, more than 90 square meters, and the community environment is good. Please recommend it to me."
 }
    ],
    "options": {
        "temperature": 0.0,
        "stream": false
    },
    "tools": [
        {
"type": "function",
              "function": {
                "name": "planning",
"description": "\\nA planning tool that allows the agent to create and manage plans for solving complex tasks.\\nThe tool provides functionality for creating plans, updating plan steps, and tracking progress.\\n",
                "parameters": {
                    "type": "object",
 "properties": {
                        "command": {
                            "description": "The command to execute. Available commands: create, update, list, get, set_active, mark_step, delete.",
                            "enum": [
                                "create",
                                "update",
                                "list",
 "get",
                                "set_active",
                                "mark_step",
                                "delete"
 ],
                            "type": "string"
                        },
                        "plan_id": {
                            "description": "Unique identifier for the plan. Required for create, update, set_active, and delete commands. Optional for get and mark_step (uses active plan if not specified).",
                            "type": "string"
                        },
                        "title": {
 "description": "Title for the plan. Required for create command, optional for update command.",
                            "type": "string"
                        },
                        "steps": {
 "description": "List of plan steps. Required for create command, optional for update command.",
                            "type": "array",
                            "items": {
"description": "Index of the step to update (0-based). Required for mark_step command.",
                            "type": "integer"
                        },
                        "step_status": {
                            "description": "Status to set for a step. Used with mark_step command.",
                            "enum": [
                                "not_started",
                                "in_progress",
 "completed",
                                "blocked"
                            ],
                            "type": "string"
                        },
 "step_notes": {
                            "description": "Additional notes for a step. Optional for mark_step command.",
                            "type": "string"
 }
                    },
                    "required": [
                        "command"
                    ],
                    "additionalProperties": false
                }
            }
        }
    ],
    "tool_calls": "required"
}'

The result returned by the interface is as follows:

{
  "model": "qwen2.5:14b",
  "created_at": "2025-05-31T08:07:06.248964463Z",
"message": {
    "role": "assistant",
    "content": "",
    "tool_calls": [
      {
      "function": {
        "name": "planning",
"arguments": {
              "description": "determine the search scope and conditions, including rent limits, house type requirements, etc.",
              "step_index": 0
"Title": "Live inspection of potential housing,"
            "step_index": 3
}
      }
    ]
  },
  "done": false
}
  "model": "qwen2.5:14b",
  "created_at": "2025-05-31T08:07:06.359368765Z",
  "message": {
  "role": "assistant",
  "content": ""
  },
  "done_reason": "stop",
  "done": true,
"total_duration": 11605808012,
  "load_duration": 27693026,
  "prompt_eval_count": 454,
  "prompt_eval_duration": 218000000,
  "eval_duration": 2180000000,
  "eval_duration": 208,
  "eval_duration": 11343000000
}

Scaling and Challenges of Planning

• Planning Pattern: Current planning adopts linear planning, But the MetaGPT team plans to introduce a DAG (directed acyclic graph) structure to support more complex task dependencies. For example, the Data Interpreter task may require multi-step conditional judgment and dynamic adjustment;
• Implementation of dynamic programming: The system passes update_plan_status The method dynamically adjusts the plan based on the tool execution results. For example, if a step takes too long, the system may execute other subtasks in parallel to improve efficiency;
• User feedback-driven optimization: If the task fails or is not understood, the user can be supported to trigger re-planning.

2. Reflection mechanism

The reflection mechanism is mainly reflected in the use of a combination of ReAct mode and real-time feedback.

• ReAct mode
  : OpenManus will make dynamic decisions and iterative optimization through ReAct mode during task execution. Regarding the implementation of the core mechanism of the ReAct framework, the implementation of the ReActAgent class is specifically reflected, and two static methods are defined. Analyze the current state through LLM, generate thinking results, and select the tool to perform the next operation. In this process, the system will verify and repair previous decisions and actions through self-reflection, thereby improving the quality of task completion. For example, in the think method, LLM generates a response with tool options based on historical messages and tool states, which is essentially a reflective adjustment to past behavior.

class ReActAgent(BaseAgent, ABC):
    name: str
description: Optional[str] = None

    system_prompt: Optional[str] = None
    next_step_prompt: Optional[str] = None
    llm: Optional[LLM] = Field(default_factory=LLM)
  memory: Memory = Field(default_factory=Memory)
    state: AgentState = AgentState.IDLE

    max_steps: int = 10
    current_step: int = 0
    @abstractmethod
    async def think(self) -> bool:
      """Process current state and decide next action"""

    @abstractmethod
    async def act(self) -> str:
      """Execute decided actions"""
    async def step(self) -> str:
        """Execute a single step: think and act.""""
          should_act = await self.think()
        if not should_act:
        return "Thinking complete - no action needed"
      return await self.act()

• Real-time feedback and dynamic adjustment
  : OpenManus' real-time feedback mechanism not only shows the user the thinking process (such as task decomposition logic, tool call steps), but also allows the system to dynamically update memory and context based on the execution results. For example, when a repeated step is detected, the is_stuck method triggers an abnormal termination to avoid falling into an invalid loop. This mechanism optimizes follow-up by reflecting on historical steps.

Compared with traditional end-to-end models, OpenManus' reflection mechanism allows it to locate problems more accurately when the task fails, but there are still many to be optimized and modified. The current points are as follows:

• OpenManus project is currently a single agent system, which will cause fuzzy errors in the error attribution range. When a tool-level error occurs in the task chain (such as browser automation failure), the system can only locate the current execution node, but cannot trace the potential decision-making errors in the upstream planning stage. Therefore, it is necessary to introduce error propagation tracking of multi-agent architectures to build a task dependency map; the dynamic decision-making mechanism of
• The dynamic decision-making mechanism is insufficient: the state evaluation threshold of ReAct loop is set too rigid and the ability to cope with complex scenarios is insufficient.

Memory mechanism

The memory mechanism is through multiple rounds of dialogue and contextual association. For example, dynamic context management, each interaction will record the complete dialogue history (including user input, tool call results, LLM response, etc.) through the update_memory method, forming a traceable context chain. When initialized in the ReAgent class, two objects will be instantiated. One is used to implement short-term memory and maintain a fixed-length context window through MemoryBuffer to avoid performance degradation or errors caused by too long context; the other is long-term memory, which uses a vector database to store key information and supports calling historical knowledge through semantic retrieval.

OpenManus by Memory class stores and manages dialogue history. This component records interaction records between users, systems and tools in the form of a message list, and uses update_memory Dynamic update style="color: #24292F;font-size: 16px;"> . Specific implementations include:

• Message type support : Supports user input, system messages, assistant replies and tool execution results;
• Dynamic update mechanism : Automatically evaluate the importance of information after each interaction, decide whether to be included in the memory bank, and avoid redundant data accumulation.

class Memory(BaseModel):
    messages: List[Message] = Field(default_factory=list)
    max_messages: int = Field(default=100)

    def add_message(self, message: Message) -> None:

        self.messages.append(message)
        # Optional: Implement message limit
            if len(self.messages) > self.max_messages:
          self.messages = self.messages[-self.max_messages :]

    def add_messages(self, messages: List[Message]) -> None:
      """Add multiple messages to memory""""
      self.messages.extend(messages)
    def clear(self) -> None:
      """Clear all messages"""
        self.messages.clear()
      def get_recent_messages(self, n: int) -> List[Message]:
      """Get n most recent messages"""""
      return self.messages[-n:]

    def to_dict_list(self) -> List[dict]:
      """Convert messages to list of dicts""""
        return [msg.to_dict() for msg in self.messages]

return [msg.to_dict() for msg in self.messages]

style="font-size: 16px;"> About long-term memory, it is still under planning. The community mentioned the introduction of long-term memory storage verified knowledge or stable state to improve the efficiency of repetitive tasks. The source code shows that the current memory is only temporarily stored in memory, but the architecture design reserves the possibility of expansion through database or file persistence.

Practical cases

Inspired by the OpenManus project, we have also conducted pilot exploration in the operation and maintenance scenarios of the big data real-time engine side. Use the intelligent planning, tool call, execution and reflection process to build self-update and optimize operation and maintenance tools.

1.

• The root cause analysis is performed for Flink real-time job exceptions. The exception types mainly include: job failure, data accumulation, flow failure, and Checkpoint failure.

2.

• Solve the root cause analysis of job abnormalities (job failure, data accumulation, flow failure, Checkpoint failure) to improve automatic operation and maintenance and diagnosis effects.

3.

The specific process of the execution process is as follows:

class Memory:
    """Memory system that stores the interactive history of an agent"""
    def __init__(self):
self.messages: List[Message] = []  # Message Queue (Short-term Memory)
        self.knowledge_base = Elasticsearch()  # Knowledge Base (Long-term Memory)
    def add_message(self, message: Union[Message, dict]):
      """Add structured messages to memory stream"""
      if isinstance(message, dict):
            message = Message(**message)
        self.messages.append(message)
        self._update_base(message)  # 触发知识库更新

    def _update_base(self, message):
"""Build knowledge nodes based on message content"""

              properties={'name': entity['text'], 'context': message.context}
          )
          )
      def get_relevant_memories(self, query: str, top_k=5) -> List[Message]:
      """Get related memories based on hybrid search""
        query_embed = self.encoder.encode(query)
 return sorted(self.messages, 
                     key=lambda x: cosine_similarity(query_embed, x.embed))[:top_k]

FLINK_DIAGNOSIS_PLAN_PROMPT = """
You are a planning agent expert, and your task is to solve complex problems by creating structured plans.
Your requirements are:
1. If the idea is clear in the question, build the plan according to the idea; if the idea is provided in the context, build the plan according to the context; if the idea is not clear, generate a structured plan according to the content of the question.
Your job is:
1. Analyze the requests to understand the scope of the task.
2. Use the “plan” tool to create a clear and feasible plan.
3. Perform steps using the available tools as needed.
4. There are known error messages in the history and need to be corrected in the next generated result.
5. Track progress and adjust the plan dynamically.
6. The directive object in the output json format content is an array type, and the value in the array is a str type. The type of the `thoughts` object is str.

Output json list in the following format:
```json
{{
  "plan":
  {{
  "thoughts": str # ]
    "instructions": [
    str # Record the steps
    ]
  }}
  }}
}}