"""MCP server implementation for Mem.ai API.
This module provides a FastMCP-based server that exposes Mem.ai's
intelligent memory platform to AI assistants via the Model Context Protocol.
"""
import logging
import os
from typing import Optional
from dotenv import load_dotenv
from fastmcp import FastMCP
from .client import MemClient
# Load environment variables
load_dotenv()
# Configure logging
log_level = logging.DEBUG if os.getenv("MEM_DEBUG", "false").lower() == "true" else logging.INFO
logging.basicConfig(
level=log_level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# Initialize MCP server
mcp = FastMCP("Mem.ai Memory Platform")
# Global client instance
_client: Optional[MemClient] = None
async def get_client() -> MemClient:
"""Get or create the global MemClient instance.
Returns:
MemClient instance
Raises:
RuntimeError: If MEM_API_KEY is not set
"""
global _client
if _client is None:
_client = MemClient()
return _client
@mcp.tool()
async def mem_it(
input: str,
instructions: str | None = None,
context: str | None = None,
timestamp: str | None = None,
) -> str:
"""Save and intelligently process content with Mem It.
Mem It is the most powerful way to save information to your Mem.ai account.
It automatically processes, organizes, and extracts knowledge from various
content types including notes, web pages, meeting transcripts, emails, and documents.
Args:
input: The content to save. Can be plain text, HTML, markdown, or any text content.
Examples: meeting notes, article text, research findings, email content.
instructions: Optional instructions for how to process and organize the content.
Examples: "Extract key findings", "Save as research note",
"Summarize main points".
context: Optional additional context about the content to help with organization.
Examples: "Project research", "Customer feedback", "Weekly meeting".
timestamp: Optional ISO 8601 timestamp for when the content was created.
Example: "2024-01-15T14:30:00Z"
Returns:
Success message with the request ID from Mem.ai
Example:
Save a research article:
```
mem_it(
input="<article content>",
instructions="Extract key findings and save as research note",
context="AI Research 2024"
)
```
"""
client = await get_client()
try:
response = await client.mem_it(
input=input,
instructions=instructions,
context=context,
timestamp=timestamp,
)
return f"Successfully saved content to Mem.ai (request_id: {response.request_id})"
except Exception as e:
logger.error(f"Failed to save content: {e}")
return f"Error: {str(e)}"
@mcp.tool()
async def create_note(
content: str,
collection_ids: list[str] | None = None,
collection_titles: list[str] | None = None,
) -> str:
"""Create a new structured note in Mem.ai.
Notes are markdown-formatted documents that can be organized into collections.
Use this when you want explicit control over the note content and organization,
rather than automatic processing via Mem It.
Args:
content: The note content in markdown format. Can include headings, lists,
code blocks, and other markdown formatting.
collection_ids: Optional list of collection UUIDs to add this note to.
Example: ["3c90c3cc-0d44-4b50-8888-8dd25736052a"]
collection_titles: Optional list of collection titles to add this note to.
Example: ["Work Projects", "Research"]
Returns:
Success message with note ID and title, or error message
Example:
Create a meeting notes:
```
create_note(
content=\"\"\"# Team Standup - Jan 15, 2024
## Attendees
- Alice, Bob, Charlie
## Discussion
- Project X milestone reached
- Planning sprint for next week
\"\"\",
collection_titles=["Team Meetings"]
)
```
"""
client = await get_client()
try:
note = await client.create_note(
content=content,
collection_ids=collection_ids,
collection_titles=collection_titles,
)
collections_info = f" in {len(note.collection_ids)} collection(s)" if note.collection_ids else ""
return f"Successfully created note: '{note.title}' (ID: {note.id}){collections_info}"
except Exception as e:
logger.error(f"Failed to create note: {e}")
return f"Error: {str(e)}"
@mcp.tool()
async def read_note(note_id: str) -> str:
"""Read and retrieve a note from Mem.ai by its ID.
Use this to fetch the full content and metadata of a specific note.
Args:
note_id: The UUID of the note to read.
Example: "01961d40-7a67-7049-a8a6-d5638cbaaeb9"
Returns:
The note's title, content, and metadata formatted as text, or error message
Example:
```
read_note("01961d40-7a67-7049-a8a6-d5638cbaaeb9")
```
"""
client = await get_client()
try:
note = await client.read_note(note_id)
collections_info = f"\nCollections: {len(note.collection_ids)}" if note.collection_ids else ""
return f"""Note: {note.title}
ID: {note.id}
Created: {note.created_at}
Updated: {note.updated_at}{collections_info}
Content:
{note.content}
"""
except Exception as e:
logger.error(f"Failed to read note: {e}")
return f"Error: {str(e)}"
@mcp.tool()
async def delete_note(note_id: str) -> str:
"""Delete a note from Mem.ai by its ID.
This permanently removes the note. This action cannot be undone.
Args:
note_id: The UUID of the note to delete.
Example: "01961d40-7a67-7049-a8a6-d5638cbaaeb9"
Returns:
Success message with request ID, or error message
Example:
```
delete_note("01961d40-7a67-7049-a8a6-d5638cbaaeb9")
```
"""
client = await get_client()
try:
response = await client.delete_note(note_id)
return f"Successfully deleted note {note_id} (request_id: {response.request_id})"
except Exception as e:
logger.error(f"Failed to delete note: {e}")
return f"Error: {str(e)}"
@mcp.tool()
async def create_collection(
title: str,
description: str | None = None,
) -> str:
"""Create a new collection to organize notes.
Collections are containers for organizing related notes. Examples include
"Work Projects", "Research Papers", "Meeting Notes", etc.
Args:
title: The collection title. Should be descriptive and unique.
Examples: "Acme Corp", "AI Research 2024", "Customer Feedback"
description: Optional markdown-formatted description explaining what
the collection is for and what types of notes it contains.
Returns:
Success message with collection ID and title, or error message
Example:
Create a project collection:
```
create_collection(
title="Project Apollo",
description=\"\"\"# Project Apollo
A collection for all notes related to Project Apollo including:
- Meeting notes
- Technical specifications
- Customer feedback
\"\"\"
)
```
"""
client = await get_client()
try:
collection = await client.create_collection(
title=title,
description=description,
)
return f"Successfully created collection: '{collection.title}' (ID: {collection.id})"
except Exception as e:
logger.error(f"Failed to create collection: {e}")
return f"Error: {str(e)}"
@mcp.tool()
async def delete_collection(collection_id: str) -> str:
"""Delete a collection from Mem.ai by its ID.
This removes the collection but does not delete the notes within it.
The notes will simply no longer be associated with this collection.
Args:
collection_id: The UUID of the collection to delete.
Example: "5e29c8a2-c73b-476b-9311-e2579712d4b1"
Returns:
Success message with request ID, or error message
Example:
```
delete_collection("5e29c8a2-c73b-476b-9311-e2579712d4b1")
```
"""
client = await get_client()
try:
response = await client.delete_collection(collection_id)
return f"Successfully deleted collection {collection_id} (request_id: {response.request_id})"
except Exception as e:
logger.error(f"Failed to delete collection: {e}")
return f"Error: {str(e)}"
if __name__ == "__main__":
# Run the MCP server
mcp.run()
