"""Advanced usage examples for the Mem.ai MCP server.
This script demonstrates advanced patterns and workflows including:
- Batch operations
- Error handling
- Complex content processing
- Collection-based organization
"""
import asyncio
from datetime import datetime, timezone
from typing import List
from dotenv import load_dotenv
from mcp_mem import MemClient
from mcp_mem.client import MemAPIError, MemNotFoundError
# Load environment variables
load_dotenv()
async def batch_save_meeting_notes(client: MemClient, meetings: List[dict]) -> None:
"""Save multiple meeting notes in batch.
Args:
client: MemClient instance
meetings: List of meeting dictionaries with 'title', 'content', 'date'
"""
print(f"\n📝 Batch saving {len(meetings)} meeting notes...")
tasks = []
for meeting in meetings:
# Create markdown content with timestamp
content = f"""# {meeting['title']}
**Date:** {meeting['date']}
{meeting['content']}
"""
# Create async task for each note
task = client.create_note(
content=content,
collection_titles=["Team Meetings"],
)
tasks.append(task)
# Execute all tasks concurrently
results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results
success_count = 0
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f" ✗ Failed to save '{meetings[i]['title']}': {result}")
else:
print(f" ✓ Saved '{result.title}'")
success_count += 1
print(f" → {success_count}/{len(meetings)} notes saved successfully")
async def process_research_articles(client: MemClient, articles: List[dict]) -> None:
"""Process and categorize research articles.
Args:
client: MemClient instance
articles: List of article dictionaries with 'url', 'content', 'category'
"""
print(f"\n🔬 Processing {len(articles)} research articles...")
for article in articles:
try:
response = await client.mem_it(
input=article["content"],
instructions=f"Extract key findings and methodology. Categorize as {article['category']} research.",
context=f"Research - {article['category']}",
timestamp=datetime.now(timezone.utc).isoformat(),
)
print(f" ✓ Processed article: {article.get('title', 'Untitled')}")
print(f" Request ID: {response.request_id}")
except MemAPIError as e:
print(f" ✗ Failed to process article: {e}")
async def organize_project_knowledge(client: MemClient, project_name: str) -> None:
"""Create a structured knowledge base for a project.
Args:
client: MemClient instance
project_name: Name of the project
"""
print(f"\n📚 Organizing knowledge base for '{project_name}'...")
# Create main project collection
try:
main_collection = await client.create_collection(
title=f"{project_name} - Main",
description=f"""# {project_name} Knowledge Base
This collection contains all documentation and knowledge related to the {project_name} project.
## Contents
- Technical documentation
- Meeting notes
- Design decisions
- User research
- Development logs
""",
)
print(f" ✓ Created main collection: {main_collection.id}")
# Create project overview note
overview = await client.create_note(
content=f"""# {project_name} - Project Overview
## Mission
Build an innovative solution that solves real user problems.
## Key Objectives
1. Deliver high-quality, maintainable code
2. Create excellent user experience
3. Build scalable architecture
4. Maintain comprehensive documentation
## Timeline
- **Start Date:** {datetime.now().strftime('%Y-%m-%d')}
- **Target Launch:** Q2 2024
## Team
- Engineering: 5 developers
- Product: 2 product managers
- Design: 2 designers
## Technology Stack
- Backend: Python, FastAPI
- Frontend: React, TypeScript
- Database: PostgreSQL
- Infrastructure: AWS, Docker, Kubernetes
## Success Metrics
- User satisfaction score > 4.5/5
- System uptime > 99.9%
- API response time < 200ms
- Test coverage > 90%
""",
collection_ids=[str(main_collection.id)],
)
print(f" ✓ Created overview note: {overview.title}")
# Create technical decisions log
tech_decisions = await client.create_note(
content=f"""# {project_name} - Technical Decisions Log
## Architecture Decisions
### ADR-001: Use FastAPI for Backend
**Date:** {datetime.now().strftime('%Y-%m-%d')}
**Status:** Accepted
**Context:**
We need a modern, fast Python web framework for our API.
**Decision:**
Use FastAPI for the backend API implementation.
**Rationale:**
- Native async/await support
- Automatic API documentation
- Type hints and validation with Pydantic
- Excellent performance
- Great developer experience
**Consequences:**
- Team needs to learn async Python patterns
- Requires Python 3.10+
- Excellent ecosystem support
### ADR-002: PostgreSQL for Primary Database
**Date:** {datetime.now().strftime('%Y-%m-%d')}
**Status:** Accepted
**Context:**
We need a reliable, scalable relational database.
**Decision:**
Use PostgreSQL as our primary database.
**Rationale:**
- Proven reliability and performance
- Rich feature set (JSON, full-text search, etc.)
- Strong ACID compliance
- Excellent tooling and community support
**Consequences:**
- Need to set up backup and replication
- Requires database migration strategy
- Team has strong PostgreSQL expertise
""",
collection_ids=[str(main_collection.id)],
)
print(f" ✓ Created technical decisions: {tech_decisions.title}")
print(f" → Knowledge base for '{project_name}' organized successfully")
except MemAPIError as e:
print(f" ✗ Failed to organize knowledge base: {e}")
async def safe_read_note(client: MemClient, note_id: str) -> None:
"""Demonstrate safe note reading with error handling.
Args:
client: MemClient instance
note_id: UUID of the note to read
"""
print(f"\n📖 Reading note {note_id}...")
try:
note = await client.read_note(note_id)
print(f" ✓ Successfully read note: '{note.title}'")
print(f" Created: {note.created_at}")
print(f" Updated: {note.updated_at}")
print(f" Collections: {len(note.collection_ids)}")
print(f" Content preview: {note.content[:100]}...")
except MemNotFoundError:
print(f" ⚠ Note {note_id} not found")
except MemAPIError as e:
print(f" ✗ Error reading note: {e}")
async def cleanup_old_notes(
client: MemClient, note_ids: List[str], dry_run: bool = True
) -> None:
"""Safely delete multiple notes with dry-run support.
Args:
client: MemClient instance
note_ids: List of note UUIDs to delete
dry_run: If True, only simulate deletion
"""
mode = "DRY RUN" if dry_run else "LIVE"
print(f"\n🗑️ Cleanup mode: {mode}")
print(f" Processing {len(note_ids)} notes...")
deleted_count = 0
failed_count = 0
for note_id in note_ids:
try:
if dry_run:
# In dry run, just check if note exists
await client.read_note(note_id)
print(f" ✓ Would delete note: {note_id}")
deleted_count += 1
else:
# Actually delete the note
await client.delete_note(note_id)
print(f" ✓ Deleted note: {note_id}")
deleted_count += 1
except MemNotFoundError:
print(f" ⚠ Note {note_id} not found (already deleted?)")
failed_count += 1
except MemAPIError as e:
print(f" ✗ Failed to delete {note_id}: {e}")
failed_count += 1
print(f" → {deleted_count} processed, {failed_count} failed")
async def main():
"""Run advanced usage examples."""
async with MemClient() as client:
print("=== Advanced Mem.ai MCP Usage Examples ===")
# Example 1: Batch save meeting notes
meetings = [
{
"title": "Daily Standup - Monday",
"date": "2024-01-15",
"content": "Quick sync on progress. Everyone on track.",
},
{
"title": "Daily Standup - Tuesday",
"date": "2024-01-16",
"content": "Discussed blockers. API integration needs attention.",
},
{
"title": "Daily Standup - Wednesday",
"date": "2024-01-17",
"content": "Good progress. Planning for demo on Friday.",
},
]
await batch_save_meeting_notes(client, meetings)
# Example 2: Process research articles
articles = [
{
"title": "Async Python Best Practices",
"content": "Article about async/await patterns, asyncio, and best practices...",
"category": "Python",
},
{
"title": "Microservices Architecture",
"content": "Deep dive into microservices patterns, challenges, and solutions...",
"category": "Architecture",
},
]
await process_research_articles(client, articles)
# Example 3: Organize project knowledge base
await organize_project_knowledge(client, "Project Phoenix")
# Example 4: Safe note reading (with non-existent note)
await safe_read_note(client, "00000000-0000-0000-0000-000000000000")
# Example 5: Cleanup with dry run
# await cleanup_old_notes(
# client,
# ["note-id-1", "note-id-2", "note-id-3"],
# dry_run=True
# )
print("\n✨ All advanced examples completed!")
if __name__ == "__main__":
asyncio.run(main())
