← back to Goodquestion
MCP_SERVER_GUIDE.md
681 lines
# MCP Server Implementation Guide
Complete guide for building and deploying your Model Context Protocol (MCP) server for the SWOT Analysis Platform.
## What is MCP?
The Model Context Protocol (MCP) is a standardized way to connect AI models with external data sources. For this platform, the MCP server:
1. **Stores** local business data in a vector database
2. **Searches** through documents using semantic similarity
3. **Retrieves** full document content on demand
4. **Provides** context to Claude for analysis
## Architecture Overview
```
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Next.js App │────────>│ MCP Server │────────>│ Vector Store │
│ (Frontend) │ HTTP │ (FastMCP) │ Query │ (ChromaDB) │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
│ Ingest
▼
┌──────────────────┐
│ Data Sources │
│ - PDFs │
│ - Directories │
│ - Web Scraping │
└──────────────────┘
```
## Quick Start
### Option 1: Basic Python Server
```python
# mcp_server.py
from fastmcp import FastMCP
import chromadb
from chromadb.config import Settings
import os
# Initialize FastMCP server
mcp = FastMCP("SWOT Business Data")
# Initialize ChromaDB (vector database)
client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./chroma_data"
))
# Create or get collection
collection = client.get_or_create_collection(
name="business_data",
metadata={"description": "Local business and economic data"}
)
@mcp.tool()
def search(
query: str,
filters: dict = None,
limit: int = 10
) -> list:
"""
Search the vector database for relevant documents
Args:
query: Search query string
filters: Optional metadata filters
limit: Maximum number of results
Returns:
List of search results with relevance scores
"""
# Build where clause from filters
where = None
if filters:
where = {}
if 'documentType' in filters:
where['document_type'] = {'$in': filters['documentType']}
if 'location' in filters:
where['location'] = filters['location']
# Perform semantic search
results = collection.query(
query_texts=[query],
n_results=limit,
where=where
)
# Format results
formatted_results = []
for i in range(len(results['ids'][0])):
formatted_results.append({
'id': results['ids'][0][i],
'title': results['metadatas'][0][i].get('title', 'Untitled'),
'snippet': results['documents'][0][i][:200] + '...',
'relevance': 1 - results['distances'][0][i], # Convert distance to similarity
'source': results['metadatas'][0][i].get('source', 'Unknown'),
'datePublished': results['metadatas'][0][i].get('date_published')
})
return formatted_results
@mcp.tool()
def fetch(document_id: str) -> dict:
"""
Fetch the full content of a document by ID
Args:
document_id: Unique document identifier
Returns:
Complete document with metadata
"""
result = collection.get(
ids=[document_id],
include=['documents', 'metadatas']
)
if not result['ids']:
raise ValueError(f"Document {document_id} not found")
return {
'id': result['ids'][0],
'content': result['documents'][0],
'metadata': result['metadatas'][0]
}
# Health check endpoint
@mcp.tool()
def health() -> dict:
"""Check server health and database stats"""
count = collection.count()
return {
'status': 'healthy',
'document_count': count,
'version': '1.0.0'
}
if __name__ == "__main__":
port = int(os.getenv("PORT", 8000))
mcp.run(host="0.0.0.0", port=port)
```
### Installation
```bash
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install dependencies
pip install fastmcp chromadb uvicorn python-dotenv
# Run server
python mcp_server.py
```
Server runs on `http://localhost:8000`
---
## Data Ingestion
### Step 1: Prepare Data Sources
Create a data ingestion script:
```python
# ingest_data.py
import chromadb
import os
import glob
from pathlib import Path
import PyPDF2
import json
client = chromadb.Client(Settings(
persist_directory="./chroma_data"
))
collection = client.get_or_create_collection("business_data")
def ingest_pdfs(directory: str, document_type: str):
"""Ingest PDF files into vector database"""
pdf_files = glob.glob(f"{directory}/**/*.pdf", recursive=True)
for pdf_path in pdf_files:
try:
# Extract text from PDF
with open(pdf_path, 'rb') as file:
reader = PyPDF2.PdfReader(file)
text = ""
for page in reader.pages:
text += page.extract_text()
# Generate unique ID
doc_id = f"{document_type}_{Path(pdf_path).stem}"
# Add to collection
collection.add(
ids=[doc_id],
documents=[text],
metadatas=[{
'title': Path(pdf_path).stem,
'source': pdf_path,
'document_type': document_type,
'file_type': 'pdf'
}]
)
print(f"✓ Ingested: {pdf_path}")
except Exception as e:
print(f"✗ Error processing {pdf_path}: {e}")
def ingest_json(file_path: str, document_type: str):
"""Ingest structured JSON data"""
with open(file_path, 'r') as f:
data = json.load(f)
for item in data:
doc_id = f"{document_type}_{item['id']}"
# Convert item to text representation
text = f"{item.get('title', '')}\n{item.get('description', '')}\n{item.get('content', '')}"
collection.add(
ids=[doc_id],
documents=[text],
metadatas=[{
'title': item.get('title', 'Untitled'),
'source': file_path,
'document_type': document_type,
**{k: v for k, v in item.items() if k not in ['id', 'title', 'description', 'content']}
}]
)
# Example usage
if __name__ == "__main__":
# Ingest different data types
ingest_pdfs('./data/historical', 'historical-document')
ingest_pdfs('./data/zoning', 'zoning-map')
ingest_pdfs('./data/economic', 'economic-report')
ingest_json('./data/competitors.json', 'business-directory')
print(f"\nTotal documents: {collection.count()}")
```
### Step 2: Data Sources to Collect
#### Business Directories
```python
# scrape_businesses.py
import requests
from bs4 import BeautifulSoup
import json
def scrape_yelp_businesses(location: str, category: str):
"""Example: Scrape business data from Yelp"""
# Note: Use Yelp API in production
url = f"https://api.yelp.com/v3/businesses/search"
headers = {"Authorization": f"Bearer {YELP_API_KEY}"}
params = {
"location": location,
"categories": category,
"limit": 50
}
response = requests.get(url, headers=headers, params=params)
businesses = response.json()['businesses']
# Save to JSON
with open(f'data/competitors_{location}.json', 'w') as f:
json.dump(businesses, f)
return businesses
```
#### Historical Documents
- Local library archives (digitized)
- Chamber of Commerce reports
- Historical society publications
- City planning documents
#### Economic Data
- Census Bureau data
- Bureau of Labor Statistics
- Local economic development reports
- Real estate market reports
### Step 3: Automated Web Scraping
```python
# scrape_economic_data.py
import requests
from datetime import datetime
def fetch_census_data(location: str):
"""Fetch demographic data from Census API"""
url = "https://api.census.gov/data/2021/acs/acs5"
params = {
"get": "B01001_001E,B19013_001E,B25077_001E", # Population, Income, Home Value
"for": f"place:{location}",
"key": CENSUS_API_KEY
}
response = requests.get(url, params=params)
data = response.json()
# Save to vector database
collection.add(
ids=[f"census_{location}_{datetime.now().year}"],
documents=[json.dumps(data)],
metadatas=[{
'title': f"Census Data - {location}",
'document_type': 'census-data',
'location': location,
'date_published': str(datetime.now().date())
}]
)
```
---
## Advanced Features
### Semantic Search Optimization
```python
# Enhanced search with re-ranking
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
@mcp.tool()
def search_with_reranking(query: str, limit: int = 10):
"""Search with neural re-ranking for better relevance"""
# Initial retrieval (get more results)
initial_results = collection.query(
query_texts=[query],
n_results=limit * 3
)
# Re-rank using cross-encoder
pairs = [[query, doc] for doc in initial_results['documents'][0]]
scores = reranker.predict(pairs)
# Sort by score and take top results
ranked_indices = scores.argsort()[::-1][:limit]
return [initial_results['documents'][0][i] for i in ranked_indices]
```
### Hybrid Search (Keyword + Semantic)
```python
from chromadb.utils import embedding_functions
# Use custom embedding function
embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2"
)
collection = client.create_collection(
name="business_data_hybrid",
embedding_function=embed_fn,
metadata={"hnsw:space": "cosine"}
)
@mcp.tool()
def hybrid_search(query: str, filters: dict = None):
"""Combine semantic and keyword search"""
# Semantic search
semantic_results = collection.query(
query_texts=[query],
n_results=10
)
# Keyword search (simple text matching)
keyword_results = collection.get(
where_document={"$contains": query}
)
# Merge and deduplicate
all_ids = set(semantic_results['ids'][0] + keyword_results['ids'])
return list(all_ids)
```
### Caching Layer
```python
import redis
import json
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
@mcp.tool()
def cached_search(query: str, limit: int = 10):
"""Search with Redis caching"""
cache_key = f"search:{query}:{limit}"
# Check cache
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
# Perform search
results = search(query, limit=limit)
# Cache for 1 hour
redis_client.setex(cache_key, 3600, json.dumps(results))
return results
```
---
## Security & Authentication
### API Key Authentication
```python
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
VALID_API_KEYS = {
os.getenv("MCP_API_KEY"): "admin",
os.getenv("MCP_READ_KEY"): "read-only"
}
def verify_api_key(credentials: HTTPAuthorizationCredentials = Security(security)):
"""Verify API key from Authorization header"""
api_key = credentials.credentials
if api_key not in VALID_API_KEYS:
raise HTTPException(status_code=403, detail="Invalid API key")
return VALID_API_KEYS[api_key]
# Add to tool decorator
@mcp.tool(dependencies=[verify_api_key])
def search(query: str):
# ... search implementation
```
### Rate Limiting
```python
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@mcp.tool()
@limiter.limit("10/minute")
def search(query: str):
# ... search implementation
```
---
## Deployment
### Docker Deployment
```dockerfile
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application
COPY mcp_server.py .
COPY ingest_data.py .
COPY data/ ./data/
# Create data directory
RUN mkdir -p /app/chroma_data
# Ingest data on build
RUN python ingest_data.py
# Expose port
EXPOSE 8000
# Run server
CMD ["python", "mcp_server.py"]
```
Build and run:
```bash
docker build -t mcp-server .
docker run -p 8000:8000 -e MCP_API_KEY=your_key mcp-server
```
### Kubernetes Deployment
```yaml
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-server
spec:
replicas: 3
selector:
matchLabels:
app: mcp-server
template:
metadata:
labels:
app: mcp-server
spec:
containers:
- name: mcp-server
image: your-registry/mcp-server:latest
ports:
- containerPort: 8000
env:
- name: MCP_API_KEY
valueFrom:
secretKeyRef:
name: mcp-secrets
key: api-key
volumeMounts:
- name: chroma-data
mountPath: /app/chroma_data
volumes:
- name: chroma-data
persistentVolumeClaim:
claimName: chroma-pvc
---
apiVersion: v1
kind: Service
metadata:
name: mcp-service
spec:
selector:
app: mcp-server
ports:
- port: 80
targetPort: 8000
type: LoadBalancer
```
---
## Monitoring & Maintenance
### Health Checks
```python
@mcp.tool()
def health_check():
"""Comprehensive health check"""
try:
# Check database
count = collection.count()
# Check disk space
import shutil
disk = shutil.disk_usage('/app/chroma_data')
return {
'status': 'healthy',
'documents': count,
'disk_free_gb': disk.free / (1024**3),
'uptime_seconds': time.time() - start_time
}
except Exception as e:
return {
'status': 'unhealthy',
'error': str(e)
}
```
### Logging
```python
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mcp_server.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@mcp.tool()
def search(query: str):
logger.info(f"Search query: {query}")
results = collection.query(query_texts=[query])
logger.info(f"Found {len(results['ids'][0])} results")
return results
```
---
## Testing
```python
# test_mcp_server.py
import pytest
from mcp_server import search, fetch
def test_search():
results = search("coffee shops Portland", limit=5)
assert len(results) <= 5
assert all('id' in r for r in results)
def test_fetch():
# Assuming we know a document ID
doc = fetch("historical-document_portland_1990")
assert 'content' in doc
assert 'metadata' in doc
def test_search_with_filters():
results = search(
"economic growth",
filters={'documentType': ['economic-report']},
limit=10
)
assert all(r['metadata']['document_type'] == 'economic-report' for r in results)
```
Run tests:
```bash
pytest test_mcp_server.py -v
```
---
## Troubleshooting
### Common Issues
**Issue**: `chromadb.errors.ChromaError: Collection already exists`
```python
# Solution: Use get_or_create_collection
collection = client.get_or_create_collection("business_data")
```
**Issue**: Slow search performance
```python
# Solution: Reduce embedding dimensions or use faster model
embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2" # Faster, smaller model
)
```
**Issue**: Memory errors with large documents
```python
# Solution: Chunk documents
def chunk_text(text: str, chunk_size: int = 500):
words = text.split()
return [' '.join(words[i:i+chunk_size]) for i in range(0, len(words), chunk_size)]
```
---
## Next Steps
1. **Populate your database**: Run ingestion scripts with real data
2. **Test endpoints**: Use curl or Postman to verify functionality
3. **Deploy**: Choose deployment option (Replit, Railway, AWS)
4. **Monitor**: Set up logging and health checks
5. **Scale**: Add caching and load balancing as needed
For integration with the Next.js frontend, see the main README.md.