maxtheweb/MCP-IMAP-Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: maxtheweb:88fbde454278928cc07fbb55a645b9329e85af35
Branches Tags
maxtheweb:main
..
compare: maxtheweb:3c5a9a35e8f184d4a22fcaeae7b86808c3a104a6
Branches Tags
maxtheweb:main

No commits in common. "88fbde454278928cc07fbb55a645b9329e85af35" and "3c5a9a35e8f184d4a22fcaeae7b86808c3a104a6" have entirely different histories.
88fbde4542 ... 3c5a9a35e8

7 changed files with 0 additions and 977 deletions
Show Stats Expand all files Collapse all files

54
.gitignore vendored
View File

@ -1,54 +0,0 @@
# Environment variables
.env
.env.local
.env.production
.env.development
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
backend/venv/
backend/.deps_installed
# Node.js / Next.js
node_modules/
.next/
out/
frontend/node_modules/
frontend/.next/
frontend/out/
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
.pnpm-debug.log*
package-lock.json
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Build artifacts
dist/
build/
*.egg-info/
# Process IDs
.backend.pid
.frontend.pid
# Archived files
backend/archived/
# Next.js
next-env.d.ts

23
backend/.env.example
View File

@ -1,23 +0,0 @@
# MCP Server Configuration
MCP_PORT=8000
MCP_HOST=0.0.0.0
# Security Settings
MCP_API_KEYS=your_api_key_1,your_api_key_2,your_api_key_3
RATE_LIMIT_PER_MINUTE=60
MAX_EMAIL_FETCH_SIZE=100
SESSION_TIMEOUT_MINUTES=30
# Redis Configuration (for production)
REDIS_URL=redis://localhost:6379
REDIS_DB=0
# Logging
LOG_LEVEL=INFO
# CORS Settings (for frontend)
CORS_ORIGINS=http://localhost:3000,https://maxtheweb.ai
# Auth Settings (for magic links)
RESEND_API_KEY=your_resend_api_key
MAGIC_LINK_EXPIRY_MINUTES=15

253
backend/deployment/DEPLOY.md
View File

@ -1,253 +0,0 @@
# MCP IMAP Agent - Hetzner Deployment Guide
Deploy production-ready MCP IMAP server for N8N automation.
## Prerequisites
- Hetzner VPS (Ubuntu/Arch Linux)
- Domain: `imap.maxtheweb.ai` pointing to server IP
- Root or sudo access
## Installation Steps
### 1. System Setup
```bash
# Update system (Arch Linux)
sudo pacman -Syu
# Install dependencies
sudo pacman -S python python-pip nginx certbot certbot-nginx git
# Create app directory
sudo mkdir -p /opt/mcp-imap-agent
sudo chown $USER:$USER /opt/mcp-imap-agent
```
### 2. Deploy Application
```bash
# Clone repository
cd /opt/mcp-imap-agent
git clone <your-repo> .
# Create virtual environment
cd backend
python -m venv venv
source venv/bin/activate
# Install dependencies
pip install -r requirements.txt
```
### 3. Configure Environment
```bash
# Create production .env
cp .env.example .env
nano .env
```
**Required .env variables:**
```bash
MCP_PORT=8000
MCP_API_KEYS=generate_random_key_1,generate_random_key_2
RATE_LIMIT_PER_MINUTE=60
LOG_LEVEL=INFO
```
**Generate secure API keys:**
```bash
# Generate 3 random API keys for clients
openssl rand -hex 32
openssl rand -hex 32
openssl rand -hex 32
```
### 4. Install Systemd Service
```bash
# Copy service file
sudo cp deployment/mcp-imap.service /etc/systemd/system/
# Reload systemd
sudo systemctl daemon-reload
# Enable auto-start
sudo systemctl enable mcp-imap
# Start service
sudo systemctl start mcp-imap
# Check status
sudo systemctl status mcp-imap
```
### 5. Configure Nginx
```bash
# Copy nginx config
sudo cp deployment/nginx-mcp-imap.conf /etc/nginx/sites-available/mcp-imap
# Enable site
sudo ln -s /etc/nginx/sites-available/mcp-imap /etc/nginx/sites-enabled/
# Test config
sudo nginx -t
# Reload nginx
sudo systemctl reload nginx
```
### 6. Setup SSL with Certbot
```bash
# Get SSL certificate
sudo certbot --nginx -d imap.maxtheweb.ai
# Certbot auto-configures nginx and auto-renews
```
### 7. Test Deployment
```bash
# Health check (no auth)
curl https://imap.maxtheweb.ai/health
# MCP endpoint (needs API key)
curl -H "Authorization: Bearer YOUR_API_KEY" https://imap.maxtheweb.ai/mcp
```
## N8N Configuration
### Add MCP Server to N8N
N8N uses **MCP over SSE** (Server-Sent Events) protocol.
**Connection Settings:**
- **SSE URL**: `https://imap.maxtheweb.ai/mcp/sse`
- **Messages URL**: `https://imap.maxtheweb.ai/mcp/messages`
- **Authentication**: `Authorization: Bearer YOUR_API_KEY`
**In N8N:**
1. Add "HTTP Request" node or MCP-specific node
2. Configure SSE connection with API key in headers
3. MCP tools are called via JSON-RPC 2.0 protocol
### Example N8N MCP Tool Call
**Search emails:**
```json
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "search_emails",
"arguments": {
"username": "user@gmail.com",
"password": "app_password",
"host": "imap.gmail.com",
"folder": "INBOX",
"limit": 10
}
},
"id": 1
}
```
**Send email:**
```json
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "send_email",
"arguments": {
"username": "user@gmail.com",
"password": "app_password",
"smtp_host": "smtp.gmail.com",
"to": "recipient@example.com",
"subject": "Automated email",
"body": "This is sent via N8N + MCP"
}
},
"id": 2
}
```
### Available MCP Tools
1. **list_folders** - List IMAP folders
2. **search_emails** - Search with filters (sender, subject, date)
3. **get_email** - Fetch full email content
4. **send_email** - Send email via SMTP
5. **health_check** - Server health status
## Monitoring
```bash
# View logs
sudo journalctl -u mcp-imap -f
# Check nginx logs
sudo tail -f /var/log/nginx/mcp-imap-access.log
sudo tail -f /var/log/nginx/mcp-imap-error.log
# Check resource usage
htop
```
## Security Checklist
- [ ] Firewall configured (only 80, 443, 22 open)
- [ ] SSL certificate installed and auto-renewing
- [ ] Strong API keys generated (32+ chars)
- [ ] Rate limiting enabled (10 req/s)
- [ ] Logs monitored for unauthorized access
- [ ] Regular updates: `sudo pacman -Syu`
## Troubleshooting
### Service won't start
```bash
sudo journalctl -u mcp-imap -n 50
```
### SSL errors
```bash
sudo certbot renew --dry-run
```
### Connection refused
```bash
# Check if service is running
sudo systemctl status mcp-imap
# Check port binding
sudo netstat -tlnp | grep 8000
```
## Maintenance
### Update code
```bash
cd /opt/mcp-imap-agent/backend
git pull
sudo systemctl restart mcp-imap
```
### Rotate API keys
```bash
nano .env # Update MCP_API_KEYS
sudo systemctl restart mcp-imap
```
### Backup
```bash
# Backup .env (contains API keys)
sudo cp /opt/mcp-imap-agent/backend/.env ~/mcp-backup.env.$(date +%Y%m%d)
```
## Support
Issues: https://github.com/yourusername/mcp-imap-agent/issues
Docs: https://modelcontextprotocol.io

25
backend/deployment/mcp-imap.service
View File

@ -1,25 +0,0 @@
[Unit]
Description=MCP IMAP Agent - Email automation server for N8N
After=network.target
[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/mcp-imap-agent/backend
Environment="PATH=/opt/mcp-imap-agent/backend/venv/bin"
EnvironmentFile=/opt/mcp-imap-agent/backend/.env
ExecStart=/opt/mcp-imap-agent/backend/venv/bin/python src/mcp_server.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
# Security hardening
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/mcp-imap-agent/backend
[Install]
WantedBy=multi-user.target

94
backend/deployment/nginx-mcp-imap.conf
View File

@ -1,94 +0,0 @@
# Nginx configuration for MCP IMAP Agent
# Place in: /etc/nginx/sites-available/mcp-imap
# Enable with: ln -s /etc/nginx/sites-available/mcp-imap /etc/nginx/sites-enabled/
# Get SSL cert: certbot --nginx -d imap.maxtheweb.ai
upstream mcp_backend {
server 127.0.0.1:8000;
keepalive 32;
}
server {
listen 80;
server_name imap.maxtheweb.ai;
# Redirect HTTP to HTTPS
return 301 https://$server_name$request_uri;
}
server {
listen 443 ssl http2;
server_name imap.maxtheweb.ai;
# SSL certificates (certbot will populate these)
ssl_certificate /etc/letsencrypt/live/imap.maxtheweb.ai/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/imap.maxtheweb.ai/privkey.pem;
# SSL configuration (Mozilla Intermediate)
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers off;
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
# Security headers
add_header X-Frame-Options "DENY" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
# Logging
access_log /var/log/nginx/mcp-imap-access.log;
error_log /var/log/nginx/mcp-imap-error.log;
# MCP SSE endpoint for N8N (streaming)
location /mcp/sse {
proxy_pass http://mcp_backend;
# Proxy headers
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# SSE support (critical for MCP protocol)
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 86400s;
proxy_send_timeout 86400s;
# Chunked transfer encoding for streaming
chunked_transfer_encoding on;
# Rate limiting
limit_req zone=api_limit burst=10 nodelay;
}
# MCP messages endpoint for N8N (posting)
location /mcp/messages {
proxy_pass http://mcp_backend;
# Proxy headers
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Standard HTTP
proxy_http_version 1.1;
# Rate limiting
limit_req zone=api_limit burst=10 nodelay;
}
# Health check endpoint (no auth required)
location /health {
proxy_pass http://mcp_backend;
access_log off;
}
}
# Rate limiting zone (10 req/sec per IP)
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;

33
backend/requirements.txt
View File

@ -1,33 +0,0 @@
# Core MCP dependencies
fastmcp>=0.5.0
# IMAP email handling
imapclient>=3.0.0
email-validator>=2.1.0
# SMTP support
aiosmtplib>=3.0.0
# Security and authentication
python-dotenv>=1.0.0
cryptography>=41.0.0
# Async support and performance
aiofiles>=23.2.1
aioimaplib>=1.0.1
# Rate limiting
slowapi>=0.1.9
# Web server
fastapi>=0.100.0
uvicorn[standard]>=0.30.0
# Utilities
pydantic>=2.5.0
python-multipart>=0.0.6
# Testing
pytest>=8.0.0
pytest-asyncio>=0.23.0
pytest-cov>=4.1.0

495
backend/src/mcp_server.py
View File

@ -1,495 +0,0 @@
"""
MCP IMAP Agent Server - AI-powered email automation via MCP protocol
Built with FastMCP for production-ready performance
"""
import os
import asyncio
import logging
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
from functools import lru_cache
import email
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from fastmcp import FastMCP, Context
from pydantic import BaseModel, Field, EmailStr
from dotenv import load_dotenv
import aioimaplib
from email_validator import validate_email, EmailNotValidError
from slowapi import Limiter
from slowapi.util import get_remote_address
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP("mcp-imap-agent")
# Authentication middleware for N8N multi-tenant access
@mcp.add_middleware
async def auth_middleware(request, call_next):
"""Validate API key from N8N clients"""
# Get allowed API keys from environment
allowed_keys = os.getenv("MCP_API_KEYS", "").split(",")
# Extract API key from Authorization header
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
logger.warning(f"Missing or invalid Authorization header from {request.client.host}")
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=401,
content={"error": "Missing or invalid Authorization header"}
)
api_key = auth_header.replace("Bearer ", "")
if api_key not in allowed_keys or not api_key:
logger.warning(f"Invalid API key attempt from {request.client.host}")
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=403,
content={"error": "Invalid API key"}
)
# Valid API key - proceed with request
return await call_next(request)
# Rate limiter for security
limiter = Limiter(key_func=get_remote_address)
# Configuration models
class IMAPConfig(BaseModel):
"""IMAP connection configuration"""
host: str = Field(..., description="IMAP server hostname")
port: int = Field(993, description="IMAP port (993 for SSL)")
username: EmailStr = Field(..., description="Email address")
password: str = Field(..., description="App-specific password")
use_ssl: bool = Field(True, description="Use SSL connection")
class EmailSearchParams(BaseModel):
"""Email search parameters"""
folder: str = Field("INBOX", description="Folder to search in")
sender: Optional[str] = Field(None, description="Filter by sender email")
subject: Optional[str] = Field(None, description="Filter by subject")
body: Optional[str] = Field(None, description="Search in email body")
since: Optional[str] = Field(None, description="Emails since date (YYYY-MM-DD)")
limit: int = Field(50, description="Maximum emails to return", ge=1, le=100)
class IMAPSession:
"""Manages IMAP connection with connection pooling"""
def __init__(self, config: IMAPConfig):
self.config = config
self.client: Optional[aioimaplib.IMAP4_SSL] = None
self._lock = asyncio.Lock()
async def connect(self):
"""Establish IMAP connection with retry logic"""
async with self._lock:
if self.client is None:
try:
self.client = aioimaplib.IMAP4_SSL(
host=self.config.host,
port=self.config.port
)
await self.client.wait_hello_from_server()
# Authenticate
response = await self.client.login(
self.config.username,
self.config.password
)
if response.result != 'OK':
raise Exception(f"Login failed: {response}")
logger.info(f"Connected to {self.config.host} as {self.config.username}")
except Exception as e:
logger.error(f"Connection failed: {e}")
self.client = None
raise
async def disconnect(self):
"""Close IMAP connection gracefully"""
async with self._lock:
if self.client:
try:
await self.client.logout()
except:
pass
self.client = None
async def ensure_connected(self):
"""Ensure connection is active"""
if self.client is None:
await self.connect()
async def execute(self, command: str, *args):
"""Execute IMAP command with automatic reconnection"""
await self.ensure_connected()
try:
return await getattr(self.client, command)(*args)
except Exception as e:
logger.warning(f"Command failed, reconnecting: {e}")
await self.disconnect()
await self.connect()
return await getattr(self.client, command)(*args)
# Global session cache (in production, use Redis)
sessions: Dict[str, IMAPSession] = {}
@lru_cache(maxsize=128)
def get_session(username: str, password: str, host: str) -> IMAPSession:
"""Get or create IMAP session with caching"""
key = f"{username}@{host}"
if key not in sessions:
config = IMAPConfig(
host=host,
username=username,
password=password
)
sessions[key] = IMAPSession(config)
return sessions[key]
# MCP Tools Implementation
@mcp.tool()
async def list_folders(
ctx: Context,
username: EmailStr = Field(..., description="Email address"),
password: str = Field(..., description="App-specific password"),
host: str = Field(..., description="IMAP server hostname")
) -> Dict[str, Any]:
"""
List all email folders in the account.
Returns folder names and their attributes.
"""
try:
session = get_session(username, password, host)
await session.ensure_connected()
# List folders
response = await session.execute('list')
if response.result != 'OK':
return {
"success": False,
"error": "Failed to list folders"
}
folders = []
for line in response.lines:
# Parse IMAP LIST response
if isinstance(line, bytes):
line = line.decode('utf-8')
# Extract folder name (simplified parsing)
parts = line.split('"')
if len(parts) >= 3:
folder_name = parts[-2]
folders.append({
"name": folder_name,
"path": folder_name
})
return {
"success": True,
"folders": folders,
"count": len(folders)
}
except Exception as e:
logger.error(f"list_folders error: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def search_emails(
ctx: Context,
username: EmailStr = Field(..., description="Email address"),
password: str = Field(..., description="App-specific password"),
host: str = Field(..., description="IMAP server hostname"),
folder: str = Field("INBOX", description="Folder to search"),
sender: Optional[str] = Field(None, description="Filter by sender"),
subject: Optional[str] = Field(None, description="Filter by subject"),
body: Optional[str] = Field(None, description="Search in body"),
since: Optional[str] = Field(None, description="Since date YYYY-MM-DD"),
limit: int = Field(20, description="Max results", ge=1, le=100)
) -> Dict[str, Any]:
"""
Search emails with multiple filters.
Returns email metadata and snippets.
"""
try:
session = get_session(username, password, host)
await session.ensure_connected()
# Select folder
response = await session.execute('select', folder)
if response.result != 'OK':
return {
"success": False,
"error": f"Cannot access folder: {folder}"
}
# Build search criteria
criteria = ['ALL']
if sender:
criteria.append(f'FROM "{sender}"')
if subject:
criteria.append(f'SUBJECT "{subject}"')
if body:
criteria.append(f'BODY "{body}"')
if since:
date_obj = datetime.strptime(since, '%Y-%m-%d')
criteria.append(f'SINCE {date_obj.strftime("%d-%b-%Y")}')
search_string = ' '.join(criteria) if len(criteria) > 1 else criteria[0]
# Execute search
response = await session.execute('search', None, search_string)
if response.result != 'OK':
return {
"success": False,
"error": "Search failed"
}
# Parse message IDs
message_ids = response.lines[0].split() if response.lines else []
message_ids = message_ids[-limit:] if len(message_ids) > limit else message_ids
emails = []
for msg_id in message_ids:
# Fetch email headers
response = await session.execute(
'fetch',
msg_id.decode() if isinstance(msg_id, bytes) else str(msg_id),
'(BODY.PEEK[HEADER] FLAGS)'
)
if response.result == 'OK' and response.lines:
# Parse email headers (simplified)
header_data = b''.join(
line if isinstance(line, bytes) else line.encode()
for line in response.lines if line
)
msg = email.message_from_bytes(header_data)
emails.append({
"id": msg_id.decode() if isinstance(msg_id, bytes) else str(msg_id),
"subject": msg.get('Subject', '(no subject)'),
"from": msg.get('From', ''),
"to": msg.get('To', ''),
"date": msg.get('Date', ''),
"folder": folder
})
return {
"success": True,
"emails": emails,
"count": len(emails),
"search_criteria": search_string
}
except Exception as e:
logger.error(f"search_emails error: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def get_email(
ctx: Context,
username: EmailStr = Field(..., description="Email address"),
password: str = Field(..., description="App-specific password"),
host: str = Field(..., description="IMAP server hostname"),
email_id: str = Field(..., description="Email ID to retrieve"),
folder: str = Field("INBOX", description="Email folder")
) -> Dict[str, Any]:
"""
Get full email content including body and attachments info.
"""
try:
session = get_session(username, password, host)
await session.ensure_connected()
# Select folder
await session.execute('select', folder)
# Fetch full email
response = await session.execute('fetch', email_id, '(RFC822)')
if response.result != 'OK':
return {
"success": False,
"error": "Failed to fetch email"
}
# Parse email
raw_email = b''.join(
line if isinstance(line, bytes) else line.encode()
for line in response.lines if line
)
msg = email.message_from_bytes(raw_email)
# Extract body
body_plain = ""
body_html = ""
attachments = []
for part in msg.walk():
content_type = part.get_content_type()
content_disposition = str(part.get("Content-Disposition", ""))
if "attachment" in content_disposition:
filename = part.get_filename()
if filename:
attachments.append({
"filename": filename,
"content_type": content_type,
"size": len(part.get_payload())
})
elif content_type == "text/plain":
body_plain = part.get_payload(decode=True).decode('utf-8', errors='ignore')
elif content_type == "text/html":
body_html = part.get_payload(decode=True).decode('utf-8', errors='ignore')
return {
"success": True,
"email": {
"id": email_id,
"subject": msg.get('Subject', ''),
"from": msg.get('From', ''),
"to": msg.get('To', ''),
"cc": msg.get('Cc', ''),
"date": msg.get('Date', ''),
"body_plain": body_plain,
"body_html": body_html,
"attachments": attachments,
"folder": folder
}
}
except Exception as e:
logger.error(f"get_email error: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def send_email(
ctx: Context,
username: EmailStr = Field(..., description="Email address"),
password: str = Field(..., description="App-specific password"),
smtp_host: str = Field(..., description="SMTP server hostname"),
to: EmailStr = Field(..., description="Recipient email"),
subject: str = Field(..., description="Email subject"),
body: str = Field(..., description="Email body"),
cc: Optional[str] = Field(None, description="CC recipients"),
is_html: bool = Field(False, description="Send as HTML email")
) -> Dict[str, Any]:
"""
Send an email via SMTP.
Supports plain text and HTML formats.
"""
try:
import aiosmtplib
# Validate email
try:
validate_email(to)
except EmailNotValidError as e:
return {
"success": False,
"error": f"Invalid recipient email: {e}"
}
# Create message
msg = MIMEMultipart('alternative')
msg['From'] = username
msg['To'] = to
msg['Subject'] = subject
if cc:
msg['Cc'] = cc
# Add body
if is_html:
msg.attach(MIMEText(body, 'html'))
else:
msg.attach(MIMEText(body, 'plain'))
# Send via SMTP
async with aiosmtplib.SMTP(
hostname=smtp_host,
port=587,
start_tls=True
) as smtp:
await smtp.login(username, password)
await smtp.send_message(msg)
return {
"success": True,
"message": "Email sent successfully",
"details": {
"to": to,
"subject": subject,
"timestamp": datetime.now().isoformat()
}
}
except Exception as e:
logger.error(f"send_email error: {e}")
return {
"success": False,
"error": str(e)
}
# Health check endpoint
@mcp.tool()
async def health_check(ctx: Context) -> Dict[str, Any]:
"""Check server health and status"""
return {
"status": "healthy",
"version": "1.0.0",
"timestamp": datetime.now().isoformat(),
"active_sessions": len(sessions)
}
if __name__ == "__main__":
# Run the MCP server with SSE transport (N8N compatibility)
from fastmcp.server.http import create_sse_app
import uvicorn
app = create_sse_app(
mcp,
message_path="/mcp/messages",
sse_path="/mcp/sse"
)
uvicorn.run(
app,
host="0.0.0.0",
port=int(os.getenv("MCP_PORT", 8000)),
log_level="info"
)