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()