-
-
Notifications
You must be signed in to change notification settings - Fork 118
09 Examples
Henry edited this page Aug 23, 2025
·
1 revision
Practical examples and code snippets for common MCP Memory Service use cases.
# Store a basic memory
await memory_service.store(
content="Fixed authentication timeout by increasing JWT expiration to 24 hours",
tags=["bug-fix", "authentication", "jwt"]
)
# Store detailed information
memory_content = """
# Database Migration - User Profiles
Date: 2025-08-23
Status: Completed
## Changes:
- Added `profile_image_url` column
- Added `bio` text field (max 500 chars)
- Created index on `updated_at`
## Impact:
- Affects user registration flow
- Requires frontend updates
- Migration took 2.3 seconds
"""
await memory_service.store(
content=memory_content,
tags=["database", "migration", "user-profiles", "completed"]
)
# Find memories about authentication
results = await memory_service.search("authentication JWT token")
# Find recent bug fixes
results = await memory_service.search("bug fix", limit=10)
# Find all Python-related memories
python_memories = await memory_service.search_by_tags(["python"])
# Find completed database work
db_work = await memory_service.search_by_tags(["database", "completed"])
# Get yesterday's work
yesterday = await memory_service.recall("yesterday")
# Get this week's decisions
week_decisions = await memory_service.recall("this week decisions")
# Get last month's bug fixes
bugs = await memory_service.recall("last month bug-fix")
# 1. Document the bug
await memory_service.store(
content="""
Bug Report: Login fails with 500 error
- Occurs with email addresses containing '+' symbol
- Error: "Invalid email format" from validation regex
- Affects ~5% of user registrations
""",
tags=["bug", "login", "email-validation", "investigating"]
)
# 2. Track investigation progress
await memory_service.store(
content="Investigation: Email regex doesn't handle '+' in local part correctly",
tags=["bug", "login", "email-validation", "root-cause-found"]
)
# 3. Document the solution
await memory_service.store(
content="""
Fix: Updated email validation regex
- Changed from: /^[a-zA-Z0-9._-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$/
- To: /^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$/
- Added unit tests for emails with '+', '_', and '%' characters
""",
tags=["bug", "login", "email-validation", "fixed", "tested"]
)
# Planning phase
await memory_service.store(
content="""
Feature Planning: Real-time Notifications
- Use WebSocket for live updates
- Fallback to Server-Sent Events for older browsers
- Store notifications in Redis (TTL: 30 days)
- UI: Toast notifications + notification center
""",
tags=["feature", "notifications", "planning", "websocket"]
)
# Implementation milestones
await memory_service.store(
content="WebSocket connection manager implemented with auto-reconnect",
tags=["feature", "notifications", "websocket", "completed"]
)
await memory_service.store(
content="Notification UI components created and tested",
tags=["feature", "notifications", "ui", "completed"]
)
# Store code review feedback
await memory_service.store(
content="""
Code Review: User Authentication Module
Feedback:
- Add input validation for password strength
- Implement rate limiting for login attempts
- Consider using bcrypt rounds = 12 for production
- Add logging for failed authentication attempts
""",
tags=["code-review", "authentication", "security", "feedback"]
)
#!/bin/bash
# .git/hooks/post-commit
COMMIT_MSG=$(git log -1 --pretty=%B)
BRANCH=$(git branch --show-current)
FILES=$(git diff-tree --no-commit-id --name-only -r HEAD | head -5 | tr '\n' ', ')
# Store commit info in memory service
curl -X POST "https://localhost:8443/api/memories" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $MCP_API_KEY" \
-d "{
\"content\": \"Commit on $BRANCH: $COMMIT_MSG\\nFiles: $FILES\",
\"tags\": [\"git\", \"commit\", \"$BRANCH\"]
}"
# .github/workflows/deploy.yml
- name: Store Deployment Memory
run: |
MEMORY="Deployment to ${{ github.event.inputs.environment }}
- Version: ${{ github.sha }}
- Status: ${{ job.status }}
- Duration: ${{ steps.deploy.outputs.duration }}
- Timestamp: $(date -u +'%Y-%m-%dT%H:%M:%SZ')"
curl -X POST "https://your-memory-service:8443/api/memories" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer ${{ secrets.MEMORY_API_KEY }}" \
-d "{
\"content\": \"$MEMORY\",
\"tags\": [\"deployment\", \"${{ github.event.inputs.environment }}\", \"automated\"]
}"
// Store code snippets from VS Code
const vscode = require('vscode');
const storeCodeSnippet = async () => {
const editor = vscode.window.activeTextEditor;
if (!editor) return;
const selection = editor.document.getText(editor.selection);
const language = editor.document.languageId;
const fileName = path.basename(editor.document.fileName);
const memory = {
content: `Code snippet from ${fileName}:\n\`\`\`${language}\n${selection}\n\`\`\``,
tags: ['code-snippet', language, 'vscode', fileName]
};
await fetch('https://localhost:8443/api/memories', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${apiKey}`
},
body: JSON.stringify(memory)
});
};
import schedule
import asyncio
from datetime import datetime, timedelta
async def generate_daily_summary():
today = datetime.now().date()
memories = await memory_service.recall("today")
# Categorize memories
bugs_fixed = [m for m in memories if 'bug-fix' in m.tags and 'completed' in m.tags]
features_worked = [m for m in memories if 'feature' in m.tags]
decisions_made = [m for m in memories if 'decision' in m.tags]
summary = f"""Daily Summary - {today}
Total activities: {len(memories)}
🐛 Bugs Fixed: {len(bugs_fixed)}
{chr(10).join(f"- {bug.content[:100]}..." for bug in bugs_fixed[:3])}
🚀 Feature Work: {len(features_worked)}
{chr(10).join(f"- {feat.content[:100]}..." for feat in features_worked[:3])}
📋 Decisions Made: {len(decisions_made)}
{chr(10).join(f"- {dec.content[:100]}..." for dec in decisions_made[:3])}
"""
await memory_service.store(
content=summary,
tags=['daily-summary', str(today), 'automated']
)
# Schedule daily at 6 PM
schedule.every().day.at("18:00").do(lambda: asyncio.run(generate_daily_summary()))
class TeamMemorySharing:
def __init__(self, memory_service, team_members):
self.memory = memory_service
self.team = team_members
async def share_learning(self, content, topic, urgency='normal'):
# Store the learning
memory = await self.memory.store(
content=f"Team Learning: {content}",
tags=['team-learning', topic, urgency, 'shared']
)
# Notify team based on urgency
if urgency == 'urgent':
await self.notify_all_immediately(content, memory.id)
else:
await self.add_to_daily_digest(content, memory.id)
async def find_expert(self, topic):
# Find team member with most experience in topic
memories = await self.memory.search_by_tags([topic, 'team-learning'])
expert_scores = {}
for memory in memories:
author = memory.metadata.get('author')
if author in self.team:
expert_scores[author] = expert_scores.get(author, 0) + 1
return max(expert_scores.items(), key=lambda x: x[1]) if expert_scores else None
async def generate_project_docs(project_name):
# Get all project-related memories
memories = await memory_service.search_by_tags([project_name])
# Categorize by type
docs = {
'architecture': [],
'decisions': [],
'bugs': [],
'features': [],
'deployment': []
}
for memory in memories:
if 'architecture' in memory.tags:
docs['architecture'].append(memory)
elif 'decision' in memory.tags:
docs['decisions'].append(memory)
elif 'bug' in memory.tags or 'bug-fix' in memory.tags:
docs['bugs'].append(memory)
elif 'feature' in memory.tags:
docs['features'].append(memory)
elif 'deployment' in memory.tags:
docs['deployment'].append(memory)
# Generate markdown documentation
doc_content = f"""# {project_name.title()} Project Documentation
## Architecture Decisions
{chr(10).join(f"- {mem.content}" for mem in docs['architecture'])}
## Key Features
{chr(10).join(f"- {mem.content}" for mem in docs['features'])}
## Bug Fixes & Issues
{chr(10).join(f"- {mem.content}" for mem in docs['bugs'])}
## Deployment History
{chr(10).join(f"- {mem.content}" for mem in docs['deployment'])}
"""
return doc_content
async def analyze_development_patterns():
# Get last month's memories
memories = await memory_service.recall("last month")
# Analyze patterns
tag_frequency = {}
daily_activity = {}
for memory in memories:
# Tag analysis
for tag in memory.tags:
tag_frequency[tag] = tag_frequency.get(tag, 0) + 1
# Daily activity
day = memory.created_at.date()
daily_activity[day] = daily_activity.get(day, 0) + 1
# Generate insights
top_tags = sorted(tag_frequency.items(), key=lambda x: x[1], reverse=True)[:10]
most_active_day = max(daily_activity.items(), key=lambda x: x[1])
insights = f"""Development Patterns Analysis
Top Technologies: {', '.join(tag for tag, count in top_tags[:5])}
Most Active Day: {most_active_day[0]} ({most_active_day[1]} activities)
Total Memories: {len(memories)}
Average per day: {len(memories) / 30:.1f}
"""
await memory_service.store(
content=insights,
tags=['analytics', 'insights', 'monthly-report']
)
These examples demonstrate the versatility of MCP Memory Service across different development scenarios and workflows. Adapt them to your specific needs and use cases.