Memos MCP Server

Instructions

Implementation Reference

• server.py:31-128 (handler)

  The main handler function for the 'search_memos' tool. It constructs filters from input parameters, makes an HTTP GET request to the Memos API, processes the response, and returns a formatted JSON string of matching memos. Includes error handling.

  @mcp.tool()
  async def search_memos(
      query: Optional[str] = None,
      creator_id: Optional[int] = None,
      tag: Optional[str] = None,
      visibility: Optional[str] = None,
      limit: int = 10,
      offset: int = 0
  ) -> str:
      """
      Search for memos with optional filters.
      
      Args:
          query: Text to search for in memo content
          creator_id: Filter by creator user ID
          tag: Filter by tag name
          visibility: Filter by visibility (PUBLIC, PROTECTED, PRIVATE)
          limit: Maximum number of results to return (default: 10)
          offset: Number of results to skip (default: 0)
      
      Returns:
          JSON string containing the list of matching memos
      """
      # Build filter expression
      filters = []
      
      if creator_id is not None:
          filters.append(f"creator_id == {creator_id}")
      
      if query:
          # Escape quotes in query
          escaped_query = query.replace('"', '\\"')
          filters.append(f'content.contains("{escaped_query}")')
      
      if tag:
          escaped_tag = tag.replace('"', '\\"')
          filters.append(f'tag in ["{escaped_tag}"]')
      
      if visibility:
          filters.append(f'visibility == "{visibility.upper()}"')
      
      # Combine filters with AND operator
      filter_str = " && ".join(filters) if filters else ""
      
      # Build request parameters
      params = {
          "pageSize": limit,
      }
      
      if filter_str:
          params["filter"] = filter_str
      
      # Calculate page token for pagination
      if offset > 0:
          # For simplicity, we'll use offset/limit approach
          # In production, you'd want to use proper page tokens
          page = offset // limit
          if page > 0:
              params["pageToken"] = f"offset={offset}"
      
      try:
          async with httpx.AsyncClient() as client:
              response = await client.get(
                  f"{MEMOS_BASE_URL}/api/v1/memos",
                  params=params,
                  headers=get_headers(),
                  timeout=30.0
              )
              response.raise_for_status()
              data = response.json()
              
              # Format the response nicely
              memos = data.get("memos", [])
              result = {
                  "count": len(memos),
                  "memos": [
                      {
                          "name": memo.get("name"),
                          "uid": memo.get("uid"),
                          "creator": memo.get("creator"),
                          "content": memo.get("content"),
                          "visibility": memo.get("visibility"),
                          "pinned": memo.get("pinned", False),
                          "createTime": memo.get("createTime"),
                          "updateTime": memo.get("updateTime"),
                          "displayTime": memo.get("displayTime"),
                      }
                      for memo in memos
                  ],
                  "nextPageToken": data.get("nextPageToken", "")
              }
              
              return str(result)
              
      except httpx.HTTPError as e:
          return f"Error searching memos: {str(e)}"
      except Exception as e:
          return f"Unexpected error: {str(e)}"

• server.py:21-28 (helper)

  Helper utility function used by search_memos (and other tools) to generate HTTP headers for authenticated API requests to the Memos server.

  def get_headers() -> dict:
      """Get headers for API requests including authentication"""
      headers = {
          "Content-Type": "application/json",
      }
      if MEMOS_API_TOKEN:
          headers["Authorization"] = f"Bearer {MEMOS_API_TOKEN}"
      return headers

• server.py:31-31 (registration)

  The @mcp.tool() decorator registers the search_memos function as an MCP tool.

  @mcp.tool()