MCP MITM Mem0

Instructions

Implementation Reference

• mcp_mitm_mem0/mcp_server.py:125-215 (handler)

  The main MCP tool handler for 'search_memories'. This is the decorated async function that defines the tool's logic, input schema via type hints, and registration via @mcp.tool decorator. It proxies the search to the memory_service and handles errors.

  @mcp.tool(
      name="search_memories",
      description="Search conversation history using natural language - USE AUTONOMOUSLY based on conversation triggers",
  )
  async def search_memories(
      query: str, user_id: str | None = None, limit: int = 10
  ) -> list[dict[str, Any]]:
      """
      Search memories using natural language queries to find relevant past conversations.
  
      ## AUTONOMOUS USAGE TRIGGERS
  
      ### HIGH Confidence (Execute Immediately)
      - User mentions: "again", "before", "last time", "previous", "remember", "we discussed"
      - Frustration indicators: "keeps happening", "still getting", "same error"
      - Reference patterns: "that thing", "like we did", "as mentioned"
      - Debugging: recurring errors, "this error again"
  
      ### MEDIUM Confidence (Execute with Brief Context)
      - Complex technical questions that might have prior context
      - User asks "how do I..." for non-trivial topics
      - Project-specific questions
      - User seems confused about topics you might have covered
  
      ### Examples of Autonomous Execution
  
      ```python
      # User: "That CORS error is back"
      # → AUTO: search_memories("CORS error cross-origin")
  
      # User: "How do we handle auth again?"
      # → AUTO: search_memories("authentication handling auth")
  
      # User: "Let's continue with the API"
      # → AUTO: search_memories("API development")
  
      # User: "I'm getting the same TypeScript error"
      # → AUTO: search_memories("TypeScript error")
      ```
  
      ## Query Construction Guidelines
  
      ### EFFECTIVE Queries (3-50 words, specific terms)
      ```python
      # GOOD: Extract key technical terms
      "JWT refresh token implementation error"
  
      "React useEffect dependency array warning"
      "Docker compose PostgreSQL connection"
  
      # BAD: Too vague
      "auth stuff"
      "that thing we talked about"
      "help"
      ```
  
      ### Auto-Query Building from User Input
      - Extract technical terms: frameworks, libraries, error types
      - Include context words: "error", "problem", "implementation", "setup"
      - Preserve specific names: API endpoints, file names, function names
  
      ## Graceful Handling
      - **Empty results**: Continue with current knowledge, don't mention search failure
      - **Search error**: Silent fallback, proceed without memory context
      - **Partial matches**: Use what you find, don't complain about incomplete results
  
      Args:
          query: Natural language search query
              - REQUIRED: 3-50 words for optimal results
              - INCLUDE: Technical terms, specific names, error types
              - AVOID: Vague references, pronouns without context
          user_id: User ID (optional, defaults to DEFAULT_USER_ID from settings)
          limit: Maximum results to return (default: 10, recommended: 5-15)
  
      Returns:
          List of memories sorted by relevance, each containing:
          - id: Unique memory identifier
          - memory/content: The stored conversation text
          - created_at: When the memory was created
          - metadata: Additional context about the memory
      """
      try:
          results = await memory_service.search_memories(
              query=query, user_id=user_id, limit=limit
          )
          logger.info("Memory search completed", result_count=len(results))
          return results
      except Exception as e:
          logger.error("Search failed", error=str(e))
          raise RuntimeError(f"Search failed: {str(e)}") from e

• mcp_mitm_mem0/memory_service.py:130-188 (helper)

  Helper method in MemoryService class that implements the actual memory search using Mem0's AsyncMemoryClient.search API. Handles user_id defaults, filters, and logging.

  async def search_memories(
      self, query: str, user_id: str | None = None, limit: int = 10
  ) -> list[dict[str, Any]]:
      """Search memories asynchronously.
  
      Args:
          query: Search query
          user_id: User identifier (defaults to settings.default_user_id)
          limit: Maximum number of results
  
      Returns:
          List of matching memories
      """
      user_id = user_id or settings.default_user_id
  
      try:
          self._logger.info("Searching memories", user_id=user_id, query=query[:50])
  
          # v2 API requires filters parameter - validate non-empty values
          filters = {}
          if user_id and user_id.strip():
              filters["user_id"] = user_id
          if settings.default_agent_id and settings.default_agent_id.strip():
              filters["agent_id"] = settings.default_agent_id
  
          # Ensure we have at least one filter
          if not filters:
              raise ValueError(
                  "At least one filter (user_id or agent_id) must be provided"
              )
  
          search_params = {
              "query": query,
              "filters": filters,
              "version": "v2",
              "top_k": limit,
          }
  
          results = await self.async_client.search(**search_params)
  
          self._logger.info(
              "Search completed", user_id=user_id, result_count=len(results)
          )
          return results
  
      except Exception as e:
          # Enhanced error logging
          error_details = str(e)
          if hasattr(e, "response") and hasattr(e.response, "text"):
              error_details = f"{str(e)} - Response: {e.response.text}"
  
          self._logger.error(
              "Failed to search memories",
              user_id=user_id,
              error=error_details,
              search_params=search_params,
          )
          raise