MCP Server Setup
The MCP (Model Context Protocol) server exposes all Hive tools to AI agents. It can run in HTTP mode for production or STDIO mode for local testing.
Quick Start
Install Dependencies
cd tools
uv pip install -e .
Set Environment Variables
export ANTHROPIC_API_KEY=your_key_here
export BRAVE_SEARCH_API_KEY=your_key_here # Optional
The server starts on http://0.0.0.0:4001 by default.
Server Modes
HTTP Mode (Production)
HTTP mode is used for production deployments and Docker containers.
python mcp_server.py
# Starts on port 4001
STDIO Mode (Local Testing)
STDIO mode is used for local testing and direct integration with MCP clients.
python mcp_server.py --stdio
In STDIO mode, only JSON-RPC messages go to stdout. All logs go to stderr.
Configuration
Environment Variables
Anthropic API key (required at startup)
Brave Search API key (validated at agent load time)
Google API key (for Custom Search, Maps, etc.)
Google Custom Search Engine ID
GitHub personal access token
Example Configuration
# Required
ANTHROPIC_API_KEY=sk-ant-...
# Search (pick one)
BRAVE_SEARCH_API_KEY=BSA...
# OR
GOOGLE_API_KEY=AIza...
GOOGLE_CSE_ID=abc123...
# Communication
SLACK_BOT_TOKEN=xoxb-...
DISCORD_BOT_TOKEN=...
# Cloud Services
GITHUB_TOKEN=ghp_...
GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
# Server
MCP_PORT=4001
Credential Validation
The server uses a two-tier credential validation system:
Tier 1: Startup Validation
Required credentials checked at server startup:
ANTHROPIC_API_KEY - Must be set before server starts
try:
credentials.validate_startup()
logger.info("Startup credentials validated")
except CredentialError as e:
logger.warning(str(e))
Tier 2: Agent Load Validation
Tool-specific credentials validated when an agent is loaded:
BRAVE_SEARCH_API_KEY or GOOGLE_API_KEY - For web search
SLACK_BOT_TOKEN - For Slack tools
GITHUB_TOKEN - For GitHub tools
- etc.
result = web_search(query="test")
if "error" in result:
# Credentials not configured
print(result["help"]) # Instructions for getting API key
The server starts even if tool credentials are missing. Agents will receive clear error messages when trying to use tools without credentials.
Endpoints
Health Check
curl http://localhost:4001/health
# Response: OK
Use for container health checks and monitoring.
Root Endpoint
curl http://localhost:4001/
# Response: Welcome to the Hive MCP Server
MCP Protocol
MCP clients connect to the root endpoint via Server-Sent Events (SSE):
import mcp
client = mcp.ClientSession(
server_url="http://localhost:4001",
timeout=30.0
)
# List available tools
tools = await client.list_tools()
# Call a tool
result = await client.call_tool(
name="web_search",
arguments={"query": "fastmcp", "num_results": 5}
)
Docker Deployment
Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY tools/pyproject.toml tools/
RUN pip install -e tools/
# Copy source code
COPY tools/ tools/
# Expose port
EXPOSE 4001
# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD curl -f http://localhost:4001/health || exit 1
# Run server
CMD ["python", "tools/mcp_server.py", "--host", "0.0.0.0", "--port", "4001"]
Docker Compose
version: '3.8'
services:
mcp-server:
build: .
ports:
- "4001:4001"
environment:
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
- BRAVE_SEARCH_API_KEY=${BRAVE_SEARCH_API_KEY}
- SLACK_BOT_TOKEN=${SLACK_BOT_TOKEN}
- MCP_PORT=4001
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:4001/health"]
interval: 30s
timeout: 3s
retries: 3
Running with Docker
Monitoring
Logs
The server logs important events:
[MCP] Startup credentials validated
[MCP] Registered 150 tools: ['web_search', 'view_file', ...]
[MCP] Starting HTTP server on 0.0.0.0:4001
Health Monitoring
#!/bin/bash
HEALTH_URL="http://localhost:4001/health"
RESPONSE=$(curl -s -o /dev/null -w "%{http_code}" $HEALTH_URL)
if [ $RESPONSE -eq 200 ]; then
echo "Server is healthy"
exit 0
else
echo "Server is down (HTTP $RESPONSE)"
exit 1
fi
Metrics
For production monitoring, integrate with:
- Prometheus - Metric collection
- Grafana - Visualization
- Sentry - Error tracking
Scaling
Horizontal Scaling
Run multiple server instances behind a load balancer:
upstream mcp_servers {
server mcp-server-1:4001;
server mcp-server-2:4001;
server mcp-server-3:4001;
}
server {
listen 80;
location / {
proxy_pass http://mcp_servers;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
location /health {
proxy_pass http://mcp_servers/health;
}
}
Resource Limits
Set appropriate resource limits in docker-compose.yml:
services:
mcp-server:
# ...
deploy:
resources:
limits:
cpus: '2'
memory: 4G
reservations:
cpus: '1'
memory: 2G
Security
API Key Management
Never commit API keys to version control. Use environment variables or secrets management.
# Use Docker secrets
docker secret create anthropic_key anthropic_api_key.txt
# Or use cloud provider secrets
# AWS Secrets Manager
# Google Secret Manager
# Azure Key Vault
Network Security
services:
mcp-server:
networks:
- internal
# Only expose via reverse proxy
# ports:
# - "4001:4001"
networks:
internal:
driver: bridge
Rate Limiting
Implement rate limiting at the reverse proxy level:
limit_req_zone $binary_remote_addr zone=mcp:10m rate=10r/s;
server {
location / {
limit_req zone=mcp burst=20;
proxy_pass http://mcp_servers;
}
}
Troubleshooting
Check:
ANTHROPIC_API_KEY is set
- Port 4001 is not already in use
- Python version is 3.11+
# Check port usage
lsof -i :4001
# Check environment
echo $ANTHROPIC_API_KEY
# Check Python version
python --version
- Limit concurrent requests
- Set resource limits in Docker
- Monitor for memory leaks
- Restart server periodically
# Monitor memory
docker stats mcp-server
- Increase client timeout
- Check network connectivity
- Verify firewall rules
- Check server logs
# Increase timeout
client = mcp.ClientSession(
server_url="http://localhost:4001",
timeout=60.0 # 60 seconds
)
Connection Pooling
import httpx
# Use connection pooling for HTTP requests
limits = httpx.Limits(
max_keepalive_connections=20,
max_connections=100
)
client = httpx.AsyncClient(limits=limits)
Caching
from functools import lru_cache
import time
@lru_cache(maxsize=100)
def cached_search(query: str, timestamp: int) -> dict:
"""Cache search results for 5 minutes."""
return web_search(query=query)
# Use with current time bucket
result = cached_search(
query="fastmcp",
timestamp=int(time.time() / 300) # 5-minute buckets
)
Next Steps
Creating Tools
Build custom tools for the MCP server
MCP Builder Tools
Tools for building MCP servers