Bi-Temporal Knowledge Graph MCP Server
README.md
# Bi-Temporal Knowledge Graph MCP Server
A production-ready MCP (Model Context Protocol) server that gives your AI agents persistent memory with full temporal tracking. Save facts, extract entities using AI, and query historical data with time-travel capabilities.
**Build intelligent AI agents with persistent memory that understands time and context**
## Architecture
This server uses a **single-file "Database-Blind" architecture**:
- **main.py** - Everything in one file: FalkorDB driver, session management, entity extraction, memory tools, and your custom automation tools
**Structure:**
1. Configuration & Database Driver
2. Session Store & Entity Extractor
3. Graphiti Memory Core
4. Core MCP Memory Tools
5. **CUSTOM AUTOMATION TOOLS** section (add your webhook tools here!)
6. Server Startup
**Note:** This server focuses solely on memory operations. For advanced workflow orchestration, see the optional [Automation Engine OS](#-optional-automation-engine-os) section.
---
## ⭐ Star This Repo
If you find this project useful, please give it a star! It helps others discover the project and motivates continued development.
---
## 🔗 Links
- 🎁 **[Get Started](#-quick-start)** - Ready in 5 minutes
- 🎥 **[Video Tutorial](YOUR_YOUTUBE_VIDEO_LINK)** - Watch how to set it up
- ❓ **[FAQs](#-frequently-asked-questions)** - Common questions answered
- 🐛 **[Report Bugs](https://github.com/YOUR_USERNAME/bitemporal-mcp-server/issues)** - Found an issue?
- 🆕 **[Request Features](https://github.com/YOUR_USERNAME/bitemporal-mcp-server/issues)** - Have an idea?
### Resources
- 💬 **[Community](https://www.skool.com/knowledge-engineering-hub-9993)** - High Ticket AI Builders community
- 📚 **[Full Documentation](./README.md)** - Complete guide
- 🚀 **[Deployment Guide](./DEPLOYMENT.md)** - Deploy anywhere
- 🧪 **[Examples](./examples.py)** - Interactive scenarios
---
## 📑 Table of Contents
- [Features](#-features)
- [How It Works](#-how-it-works)
- [Screenshots](#-screenshots)
- [Video Tutorial](#-video-tutorial)
- [Quick Start](#-quick-start)
- [Adding Custom Tools](#-adding-custom-automation-tools)
- [API Reference - Memory Tools](#-api-reference---memory-tools)
- [Use Cases](#-use-cases)
- [FAQ](#-frequently-asked-questions)
- [Changelog](#-changelog)
- [Support](#-support)
- [Optional: Automation Engine OS](#-optional-automation-engine-os)
- [License](#-license)
---
## ✨ Features
### 🧠 Bi-Temporal Knowledge Graph
- **Smart Memory**: Automatically tracks when facts were created AND when they became true in reality
- **Conflict Resolution**: When you move locations or change jobs, old facts are automatically invalidated
- **Time Travel Queries**: Ask "Where did John live in March 2024?" and get accurate historical answers
- **Session Tracking**: Maintains context across conversations with automatic cleanup
### 🤖 AI-Powered Entity Extraction
- **Natural Language Understanding**: Just tell it in plain English - "Alice moved to San Francisco and started working at Google"
- **Automatic Relationship Discovery**: AI extracts entities and relationships without manual input
- **OpenAI Integration**: Uses GPT-4 for intelligent entity extraction
- **Graceful Degradation**: Works without AI - just add facts manually
### 🛠️ Simple Tool Extension
- **Single-File Architecture**: Everything in one `main.py` file for easy customization
- **Direct @mcp.tool() Pattern**: Add tools with a simple decorator - no config files needed
- **Single & Multi-Webhook**: Execute one webhook or fire multiple in parallel
- **Clear Custom Section**: Marked section in main.py shows exactly where to add your tools
### 🚀 Production Ready
- **Docker Support**: Complete docker-compose setup included
- **Replit Optimized**: Built specifically for Replit Autoscale environments
- **Resource Management**: Automatic session cleanup and connection pooling
- **Health Checks**: Built-in monitoring and status endpoints
- **100% Privacy-Friendly**: Your data stays in your database
---
## 🎬 How It Works
```
┌─────────────────────────────────────────────────────────┐
│ 1. Natural Language Input │
│ "Bob moved to NYC and joined Google as a PM" │
└────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 2. AI Entity Extraction (OpenAI) │
│ • Bob -> lives in -> NYC │
│ • Bob -> works at -> Google │
│ • Bob -> has role -> PM │
└────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 3. Bi-Temporal Storage (FalkorDB) │
│ • Fact: Bob works at Google │
│ • created_at: 2024-12-19T10:00:00Z │
│ • valid_at: 2024-12-19T10:00:00Z │
│ • invalid_at: null (still true) │
└────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 4. Query Anytime │
│ • "Where does Bob work now?" → Google │
│ • "What was Bob's job history?" → All past jobs │
│ • "Where did Bob live in 2023?" → Historical data │
└─────────────────────────────────────────────────────────┘
```
---
## 📸 Screenshots
### Memory in Action
### AI Entity Extraction
### Dynamic Tool Generation
### Temporal Queries
---
## 🎥 Video Tutorial
**Watch the complete setup and usage guide:**
[](YOUR_YOUTUBE_VIDEO_LINK)
**Topics covered:**
- Installation & setup (0:00)
- Adding your first facts (2:30)
- Using AI entity extraction (5:15)
- Creating automation tools (8:45)
- Temporal queries (12:20)
- Deployment to production (15:00)
---
## 🚀 Quick Start
### Option 1: Docker Compose (Recommended)
```bash
# 1. Download and extract
wget https://github.com/YOUR_USERNAME/bitemporal-mcp-server/archive/main.zip
unzip main.zip
cd bitemporal-mcp-server-main
# 2. Configure
echo "OPENAI_API_KEY=sk-your-key" > .env
# 3. Start everything (FalkorDB + MCP Server)
docker-compose up -d
# 4. Verify it's running
curl http://localhost:8080/health
```
**That's it! 🎉** Your server is now running at `http://localhost:8080/sse`
### Option 2: Python (Local Development)
```bash
# 1. Install dependencies
pip install -r requirements.txt
# 2. Configure
cp .env.example .env
# Edit .env with your settings
# 3. Start FalkorDB (Docker)
docker run -d -p 6379:6379 falkordb/falkordb:latest
# 4. Run the server
python main.py
```
### Option 3: One-Click Deploy
[](https://replit.com/github/YOUR_USERNAME/bitemporal-mcp-server)
---
## 🛠️ Adding Custom Automation Tools
Add your custom automation tools directly in `main.py` in the **CUSTOM AUTOMATION TOOLS** section.
### Step 1: Find the Custom Tools Section
Open `main.py` and scroll to around **line 800** - look for this clearly marked section:
```python
# =============================================================================
#
# ██████╗ ██╗ ██╗ ███████╗ ████████╗ ██████╗ ███╗ ███╗
# ██╔════╝ ██║ ██║ ██╔════╝ ╚══██╔══╝ ██╔═══██╗ ████╗ ████║
# ██║ ██║ ██║ ███████╗ ██║ ██║ ██║ ██╔████╔██║
# ██║ ██║ ██║ ╚════██║ ██║ ██║ ██║ ██║╚██╔╝██║
# ╚██████╗ ╚██████╔╝ ███████║ ██║ ╚██████╔╝ ██║ ╚═╝ ██║
# ╚═════╝ ╚═════╝ ╚══════╝ ╚═╝ ╚═════╝ ╚═╝ ╚═╝
#
# AUTOMATION TOOLS
#
# ADD YOUR CUSTOM AUTOMATION TOOLS BELOW
#
# =============================================================================
```
This is where you'll add your webhook tools using the `@mcp.tool()` decorator.
### Step 2: Add Your Tool
Add a decorated async function with `@mcp.tool()`:
```python
@mcp.tool()
async def send_slack_notification(message: str, channel: str = "#general") -> str:
"""Send a notification to Slack."""
import httpx
payload = {"text": message, "channel": channel}
url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
async with httpx.AsyncClient() as client:
try:
resp = await client.post(url, json=payload)
return f"Success: Slack notification sent ({resp.status_code})"
except Exception as e:
return f"Error: {str(e)}"
```
The function's **docstring becomes the tool description** that the AI sees.
### Step 3: Restart the Server
Restart the MCP server to load your new tools.
### Example: LinkedIn Poster Tools
The Automation Engine App generates tools like this:
```python
@mcp.tool()
async def linkedin_post_image(caption: str, imageurl: str) -> str:
"""Posts an image with a caption to your LinkedIn page."""
import httpx
payload = {"caption": caption, "imageUrl": imageurl}
url = "https://webhook.latenode.com/YOUR/WEBHOOK/URL"
async with httpx.AsyncClient() as client:
try:
resp = await client.post(url, json=payload)
return f"Success: LinkedIn image posted ({resp.status_code})"
except Exception as e:
return f"Error: {str(e)}"
```
### Example: Multi-Webhook Broadcast
Fire multiple webhooks in parallel:
```python
@mcp.tool()
async def broadcast_alert(message: str) -> str:
"""Send alerts to multiple platforms in parallel."""
import httpx
import asyncio
webhooks = [
("https://hooks.slack.com/...", {"text": message}),
("https://discord.com/api/webhooks/...", {"content": message}),
]
async def send(url, data):
async with httpx.AsyncClient() as client:
return await client.post(url, json=data)
results = await asyncio.gather(*[send(url, data) for url, data in webhooks])
return f"Broadcast complete: {len(results)} webhooks fired"
```
---
## 📖 API Reference - Memory Tools
All available MCP tools for managing your knowledge graph:
### Core Memory Operations
#### `add_fact`
Add a new fact to the knowledge graph with bi-temporal tracking.
```python
await add_fact(
source_entity="John",
relation="works at",
target_entity="Google",
group_id="my_org", # Optional
session_id="session_123", # Optional
valid_at="2024-01-15T00:00:00Z" # Optional - when fact became true
)
```
**Smart Conflict Resolution:** When adding location or employment facts, previous facts of the same type are automatically invalidated.
#### `add_message`
Add a natural language message and automatically extract entities using AI.
```python
await add_message(
content="Alice moved to San Francisco and started working at OpenAI",
session_id="session_123",
group_id="my_org", # Optional
extract_entities=True # Uses OpenAI for extraction
)
```
**Returns:** Extracted entities and relationships as facts.
#### `query_facts`
Query facts from the knowledge graph.
```python
await query_facts(
entity_name="John", # Optional - filter by entity
group_id="my_org", # Optional
include_invalid=False, # Include invalidated facts
max_facts=20
)
```
#### `query_at_time`
Time-travel query - get facts valid at a specific point in time.
```python
await query_at_time(
timestamp="2024-01-15T00:00:00Z",
entity_name="John", # Optional
group_id="my_org", # Optional
max_facts=20
)
```
**Use Case:** "Where did John work in January 2024?"
#### `get_episodes`
Get recent conversation sessions/episodes.
```python
await get_episodes(
group_ids=["my_org"], # Optional
max_episodes=10
)
```
#### `clear_graph`
Clear all data for specified groups. **Warning: Permanent deletion!**
```python
await clear_graph(
group_ids=["my_org"] # Optional - defaults to DEFAULT_GROUP_ID
)
```
### Server Management
#### `get_status`
Get comprehensive server status and statistics.
```python
await get_status()
# Returns: node counts, relationship types, session stats, connection status
```
#### `force_cleanup`
Manually trigger cleanup of expired sessions and idle connections.
```python
await force_cleanup()
# Returns: cleanup statistics
```
---
## 💡 Use Cases
### Personal Knowledge Management
Track your life events, relationships, and locations with full history:
```python
await add_message(
"I met Sarah at the tech conference. She works at OpenAI.",
session_id="my_life"
)
# Later: "Where did I meet Sarah?" → "At the tech conference"
```
### Customer Relationship Management
Monitor customer interactions with automatic conflict resolution:
```python
await add_fact("CustomerA", "status", "premium")
# Automatically invalidates previous "status" facts
# Query history: "What was CustomerA's status in January?"
```
### AI Agent Memory
Give your AI agents persistent, queryable memory:
```python
# Agent learns from conversation
await add_message(
"User prefers morning meetings and uses Slack",
session_id="agent_123"
)
# Agent recalls later: "What are the user's preferences?"
```
### Workflow Automation
Combine knowledge with actions:
```python
# When fact changes, trigger automation
if customer_upgraded_to_premium:
await notify_sales_team(customer_name=name)
await update_crm(customer_id=id, tier="premium")
```
---
## ❓ Frequently Asked Questions
### Q: Does this require OpenAI?
**A:** No! OpenAI is optional for AI entity extraction. You can add facts manually without it.
### Q: Can I use this with Claude Desktop?
**A:** Yes! Add the server URL to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"knowledge-graph": {
"url": "http://localhost:8080/sse"
}
}
}
```
### Q: How do I query historical data?
**A:** Use the `query_at_time` tool:
```python
await query_at_time(
timestamp="2024-01-15T00:00:00Z",
entity_name="John"
)
```
### Q: Can I deploy this to production?
**A:** Absolutely! See [DEPLOYMENT.md](./DEPLOYMENT.md) for guides on:
- Replit Autoscale
- Railway
- Render
- Fly.io
- Docker
- VPS
### Q: How does fact invalidation work?
**A:** When you add a fact about location or employment, the system automatically finds previous facts of the same type and marks them as `invalid_at: current_time`. Your query results only show current facts unless you specifically request historical data.
### Q: Can I create multi-webhook tools?
**A:** Yes! Add a tool to the Custom Tools section in `main.py` using `asyncio.gather()` to fire multiple webhooks simultaneously. See the [Adding Custom Tools](#-adding-custom-automation-tools) section for examples.
### Q: Is my data secure?
**A:** Yes! Everything runs in your infrastructure. No data is sent anywhere except:
- OpenAI (only if you use entity extraction)
- Your configured webhooks (only when you call them)
### Q: How much does it cost to run?
**A:** Free for self-hosting! Only costs:
- FalkorDB hosting (free tier available)
- OpenAI API usage (optional, ~$0.001 per extraction)
---
## 📋 Changelog
### [1.0.0] - 2024-12-19
#### Added
- ✅ Full bi-temporal tracking (created_at, valid_at, invalid_at, expired_at)
- ✅ Smart conflict resolution for location and employment changes
- ✅ Session-aware episodic memory with 30-minute TTL
- ✅ OpenAI-powered entity extraction from natural language
- ✅ Dynamic tool generator for automation workflows
- ✅ Single webhook tool template
- ✅ Multi-webhook parallel execution template
- ✅ Docker and Docker Compose support
- ✅ Replit Autoscale optimization
- ✅ Background cleanup manager
- ✅ Comprehensive documentation and examples
#### Supported Features
| Feature | Status | Notes |
|---------|--------|-------|
| Bi-Temporal Tracking | ✅ | Full implementation |
| AI Entity Extraction | ✅ | OpenAI GPT-4 |
| Smart Invalidation | ✅ | Location, employment, relationships |
| Session Management | ✅ | Auto-cleanup after 30 min |
| Custom Tools | ✅ | Single & multi-webhook via @mcp.tool() |
| Parallel Webhooks | ✅ | asyncio.gather |
| Docker Support | ✅ | Complete stack included |
| Health Checks | ✅ | Built-in monitoring |
---
## 🆘 Support
### Need Help?
1. **Check Documentation**: Start with [QUICKSTART.md](./QUICKSTART.md)
2. **Join Community**: [High Ticket AI Builders](https://www.skool.com/knowledge-engineering-hub-9993) - Free access!
3. **Watch Tutorial**: [Video Guide](YOUR_YOUTUBE_VIDEO_LINK)
4. **Report Bugs**: [GitHub Issues](https://github.com/YOUR_USERNAME/bitemporal-mcp-server/issues)
---
## 🔧 Optional: Automation Engine OS
**Need a visual tool to orchestrate your workflows?**
If you want to manage webhook configurations, generate tools automatically, and orchestrate complex workflows without writing code, check out **Automation Engine OS** - it's free when you join our community!
**What Automation Engine OS provides:**
- Visual webhook configuration builder
- Automatic MCP tool code generation
- Workflow orchestration dashboard
- Multi-webhook template management
- One-click tool deployment to your MCP server
**Get free access:** [Join High Ticket AI Builders](https://www.skool.com/knowledge-engineering-hub-9993)
**Note:** Automation Engine OS is completely optional. This MCP server works standalone - you can manually add tools to the Custom Tools section in `main.py` as shown in the [Adding Custom Automation Tools](#-adding-custom-automation-tools) section.
---
## 🤝 Contributing
Contributions are welcome! Areas for improvement:
- 🔍 Additional temporal query operators
- 🧠 Enhanced entity extraction prompts
- 🔧 More webhook authentication methods
- 📊 Performance optimizations
- 🌐 Additional deployment platforms
- 📖 More examples and tutorials
**To contribute:**
1. Fork the repository
2. Create your feature branch (`git checkout -b feature/AmazingFeature`)
3. Commit your changes (`git commit -m 'Add some AmazingFeature'`)
4. Push to the branch (`git push origin feature/AmazingFeature`)
5. Open a Pull Request
---
## 📄 License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
**TL;DR:** You can use this commercially, modify it, distribute it. Just keep the license notice.
---
## 🙏 Acknowledgments
- Built with [FastMCP](https://github.com/jlowin/fastmcp)
- Powered by [FalkorDB](https://www.falkordb.com/)
- AI features via [OpenAI](https://openai.com/)
- Inspired by the High Ticket AI Builders community
---
## ⭐ Star History
[](https://star-history.com/#YOUR_USERNAME/bitemporal-mcp-server&Date)
---
## 📞 Connect
- 💬 **Community**: [High Ticket AI Builders](https://www.skool.com/knowledge-engineering-hub-9993)
- 📅 **Want this implemented for your business?** [Book a Meeting](https://calendly.com/aiagentready/meeting)
---
<div align="center">
**Built with ❤️ for the High Ticket AI Builders ecosystem**
If this project helps you, please consider giving it a ⭐!
[⬆ Back to Top](#bi-temporal-knowledge-graph-mcp-server)
</div>