Mem0 MCP Server

Instructions

Implementation Reference

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

  The handler function decorated with @server.tool, implementing the core logic for adding memories to Mem0, including input processing, validation with AddMemoryArgs, and calling the Mem0 client.add method.
  @server.tool(description="Store a new preference, fact, or conversation snippet. Requires at least one: user_id, agent_id, or run_id.") 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/schemas.py:28-47 (schema)

  Pydantic BaseModel defining the input arguments schema for the add_memory tool, used for validation and serialization.
  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." )
• src/mem0_mcp_server/server.py:170-170 (registration)

  The @server.tool decorator registers the add_memory function as an MCP tool on the FastMCP server.
  @server.tool(description="Store a new preference, fact, or conversation snippet. Requires at least one: user_id, agent_id, or run_id.")