Mem0 MCP Server

Instructions

Implementation Reference

• src/mem0_mcp_server/server.py:171-245 (handler)

  The main handler function that executes the 'add_memory' tool logic. It processes inputs, constructs the payload using AddMemoryArgs, handles defaults, and calls the Mem0 MemoryClient's add method.

  def add_memory(
      text: Annotated[
          str,
          Field(
              description="Plain sentence summarizing what to store. Required even if `messages` is provided."
          ),
      ],
      messages: Annotated[
          Optional[list[Dict[str, str]]],
          Field(
              default=None,
              description="Structured conversation history with `role`/`content`. "
              "Use when you have multiple turns.",
          ),
      ] = None,
      user_id: Annotated[
          Optional[str],
          Field(default=None, description="Override the default user scope for this write."),
      ] = None,
      agent_id: Annotated[
          Optional[str], Field(default=None, description="Optional agent identifier.")
      ] = None,
      app_id: Annotated[
          Optional[str], Field(default=None, description="Optional app identifier.")
      ] = None,
      run_id: Annotated[
          Optional[str], Field(default=None, description="Optional run identifier.")
      ] = None,
      metadata: Annotated[
          Optional[Dict[str, Any]],
          Field(default=None, description="Attach arbitrary metadata JSON to the memory."),
      ] = None,
      enable_graph: Annotated[
          Optional[bool],
          Field(
              default=None,
              description="Set true only if the caller explicitly wants Mem0 graph memory.",
          ),
      ] = None,
      ctx: Context | None = None,
  ) -> str:
      """Write durable information to Mem0."""
  
      api_key, default_user, graph_default = _resolve_settings(ctx)
      args = AddMemoryArgs(
          text=text,
          messages=[ToolMessage(**msg) for msg in messages] if messages else None,
          user_id=user_id if user_id else (default_user if not (agent_id or run_id) else None),
          agent_id=agent_id,
          app_id=app_id,
          run_id=run_id,
          metadata=metadata,
          enable_graph=_default_enable_graph(enable_graph, graph_default),
      )
      payload = args.model_dump(exclude_none=True)
      payload.setdefault("enable_graph", graph_default)
      conversation = payload.pop("messages", None)
      if not conversation:
          derived_text = payload.pop("text", None)
          if derived_text:
              conversation = [{"role": "user", "content": derived_text}]
          else:
              return json.dumps(
                  {
                      "error": "messages_missing",
                      "detail": "Provide either `text` or `messages` so Mem0 knows what to store.",
                  },
                  ensure_ascii=False,
              )
      else:
          payload.pop("text", None)
  
      client = _mem0_client(api_key)
      return _mem0_call(client.add, conversation, **payload)

• src/mem0_mcp_server/server.py:170-170 (registration)

  Registers the add_memory function as an MCP tool using FastMCP's @server.tool decorator, with the tool name derived from the function name.

  @server.tool(description="Store a new preference, fact, or conversation snippet. Requires at least one: user_id, agent_id, or run_id.")

• src/mem0_mcp_server/schemas.py:28-47 (schema)

  Pydantic BaseModel defining the input schema and validation for the add_memory tool arguments.

  class AddMemoryArgs(BaseModel):
      text: Optional[str] = Field(
          None, description="Simple sentence to remember; converted into a user message when set."
      )
      messages: Optional[list[ToolMessage]] = Field(
          None,
          description=(
              "Explicit role/content history for durable storage. Provide this OR `text`; defaults "
              "to the server user_id."
          ),
      )
      user_id: Optional[str] = Field(None, description="Override for the Mem0 user ID.")
      agent_id: Optional[str] = Field(None, description="Optional agent identifier.")
      app_id: Optional[str] = Field(None, description="Optional app identifier.")
      run_id: Optional[str] = Field(None, description="Optional run identifier.")
      metadata: Optional[Dict[str, Any]] = Field(None, description="Opaque metadata to persist.")
      enable_graph: Optional[bool] = Field(
          None, description="Only set True if the user explicitly opts into graph storage."
      )