BluestoneApps MCP Remote Server
by lallen30
#!/usr/bin/env python3
"""
Simple MCP HTTP Server for BluestoneApps Coding Standards and Examples
This server implements the Model Context Protocol (MCP) over HTTP,
allowing remote access to coding standards and code examples.
"""
import os
import json
import glob
import logging
import asyncio
from fastapi import FastAPI, HTTPException, Depends, Request, Response, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.middleware.cors import CORSMiddleware
from sse_starlette.sse import EventSourceResponse
import uvicorn
from dotenv import load_dotenv
import secrets
# Load environment variables
load_dotenv()
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
# Initialize FastAPI
app = FastAPI(title="BluestoneApps MCP Server")
security = HTTPBasic()
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Base directory for resources
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
RESOURCES_DIR = os.path.join(BASE_DIR, "resources")
CODE_EXAMPLES_DIR = os.path.join(RESOURCES_DIR, "code-examples")
# Authentication settings
AUTH_ENABLED = os.environ.get("MCP_AUTH_ENABLED", "false").lower() == "true"
AUTH_USERNAME = os.environ.get("MCP_AUTH_USERNAME", "")
AUTH_PASSWORD = os.environ.get("MCP_AUTH_PASSWORD", "")
# MCP Server Info
SERVER_NAME = "bluestoneapps"
SERVER_VERSION = "0.2.1"
SERVER_DESCRIPTION = "BluestoneApps Coding Standards and Examples"
# Available tools
TOOLS = {
"get_project_structure": {
"description": "Get project structure standards for React Native development",
"parameters": {}
},
"get_api_communication": {
"description": "Get API communication standards for React Native development",
"parameters": {}
},
"get_component_design": {
"description": "Get component design standards for React Native development",
"parameters": {}
},
"get_state_management": {
"description": "Get state management standards for React Native development",
"parameters": {}
},
"get_component_example": {
"description": "Get a React Native component example",
"parameters": {
"component_name": {
"title": "Component Name",
"type": "string"
}
}
},
"get_hook_example": {
"description": "Get a React Native hook example",
"parameters": {
"hook_name": {
"title": "Hook Name",
"type": "string"
}
}
},
"get_service_example": {
"description": "Get a React Native service example",
"parameters": {
"service_name": {
"title": "Service Name",
"type": "string"
}
}
},
"get_screen_example": {
"description": "Get a React Native screen example",
"parameters": {
"screen_name": {
"title": "Screen Name",
"type": "string"
}
}
},
"get_theme_example": {
"description": "Get a React Native theme example",
"parameters": {
"theme_name": {
"title": "Theme Name",
"type": "string"
}
}
},
"list_available_examples": {
"description": "List all available code examples by category",
"parameters": {}
}
}
def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)):
"""Verify HTTP Basic Auth credentials"""
if not AUTH_ENABLED:
return True
correct_username = secrets.compare_digest(credentials.username, AUTH_USERNAME)
correct_password = secrets.compare_digest(credentials.password, AUTH_PASSWORD)
if not (correct_username and correct_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
headers={"WWW-Authenticate": "Basic"},
)
return True
@app.get("/")
def read_root(authenticated: bool = Depends(verify_credentials)):
"""Root endpoint"""
return {
"name": SERVER_NAME,
"version": SERVER_VERSION,
"description": SERVER_DESCRIPTION
}
@app.post("/jsonrpc")
async def jsonrpc_endpoint(request: Request, authenticated: bool = Depends(verify_credentials)):
"""JSON-RPC endpoint for MCP"""
try:
data = await request.json()
# Validate JSON-RPC request
if not isinstance(data, dict):
return {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": None}
jsonrpc = data.get("jsonrpc")
if jsonrpc != "2.0":
return {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid JSON-RPC version"}, "id": data.get("id")}
method = data.get("method")
if not method:
return {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Method not specified"}, "id": data.get("id")}
params = data.get("params", {})
request_id = data.get("id")
# Handle different methods
if method == "tools/list":
result = handle_tools_list()
elif method == "tools/call":
result = await handle_tools_call(params)
else:
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32601,
"message": f"Method {method} not found"
}
}
return {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
except Exception as e:
logging.error(f"Error processing JSON-RPC request: {e}")
return {
"jsonrpc": "2.0",
"id": None,
"error": {
"code": -32603,
"message": "Internal error"
}
}
@app.get("/sse")
async def sse_endpoint(request: Request, authenticated: bool = Depends(verify_credentials)):
"""Server-Sent Events endpoint for MCP"""
async def event_generator():
while True:
# Keep connection alive with empty messages
yield {
"data": json.dumps({"type": "ping"})
}
await asyncio.sleep(30)
return EventSourceResponse(event_generator())
def handle_tools_list():
"""Handle tools/list method"""
tool_list = []
for name, tool_info in TOOLS.items():
tool_list.append({
"name": name,
"description": tool_info["description"],
"parameters": {
"type": "object",
"properties": tool_info["parameters"]
}
})
return {
"tools": tool_list
}
async def handle_tools_call(params):
"""Handle tools/call method"""
tool_name = params.get("name")
tool_params = params.get("parameters", {})
if tool_name not in TOOLS:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tool {tool_name} not found"
)
# Call the appropriate tool function
if tool_name == "get_project_structure":
return get_project_structure()
elif tool_name == "get_api_communication":
return get_api_communication()
elif tool_name == "get_component_design":
return get_component_design()
elif tool_name == "get_state_management":
return get_state_management()
elif tool_name == "get_component_example":
return get_component_example(tool_params.get("component_name", ""))
elif tool_name == "get_hook_example":
return get_hook_example(tool_params.get("hook_name", ""))
elif tool_name == "get_service_example":
return get_service_example(tool_params.get("service_name", ""))
elif tool_name == "get_screen_example":
return get_screen_example(tool_params.get("screen_name", ""))
elif tool_name == "get_theme_example":
return get_theme_example(tool_params.get("theme_name", ""))
elif tool_name == "list_available_examples":
return list_available_examples()
else:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Tool {tool_name} implementation not found"
)
def get_standard_content(category, standard_id):
"""Get the content of a coding standard"""
standard_path = os.path.join(RESOURCES_DIR, "coding-standards", category, f"{standard_id}.md")
if not os.path.exists(standard_path):
raise HTTPException(status_code=404, detail="Standard not found")
with open(standard_path, 'r') as f:
content = f.read()
return content
def get_project_structure():
"""Get project structure standards"""
return get_standard_content("react-native", "project-structure")
def get_api_communication():
"""Get API communication standards"""
return get_standard_content("react-native", "api-communication")
def get_component_design():
"""Get component design standards"""
return get_standard_content("react-native", "component-design")
def get_state_management():
"""Get state management standards"""
return get_standard_content("react-native", "state-management")
def find_file_in_subdirectories(base_dir, file_name, extensions):
"""Find a file in subdirectories based on name and possible extensions"""
# Look for exact matches first (with provided extensions)
for ext in extensions:
for root, dirs, files in os.walk(base_dir):
for file in files:
if file.lower() == f"{file_name.lower()}{ext}":
return os.path.join(root, file)
# Look for partial matches (name in filename)
for ext in extensions:
for root, dirs, files in os.walk(base_dir):
for file in files:
if file.lower().endswith(ext) and file_name.lower() in file.lower():
return os.path.join(root, file)
return None
def get_example_content(subcategory, filename):
"""Get the content of a code example"""
filepath = os.path.join(CODE_EXAMPLES_DIR, "react-native", subcategory, filename)
if not os.path.exists(filepath) or not os.path.isfile(filepath):
raise HTTPException(status_code=404, detail="Example not found")
try:
with open(filepath, 'r') as f:
content = f.read()
except Exception as e:
logging.error(f"Error reading code example {filepath}: {e}")
raise HTTPException(status_code=500, detail=f"Error reading code example: {str(e)}")
return content
def get_component_example(component_name):
"""Get a component example"""
# First try direct match in the root components directory
try:
return get_example_content("components", f"{component_name}.tsx")
except HTTPException as e:
if e.status_code != 404:
raise e
# If not found, search in subdirectories
components_dir = os.path.join(CODE_EXAMPLES_DIR, "react-native", "components")
component_path = find_file_in_subdirectories(components_dir, component_name, [".tsx", ".jsx"])
if component_path:
try:
with open(component_path, 'r') as f:
return f.read()
except Exception as e:
logging.error(f"Error reading component example {component_path}: {e}")
raise HTTPException(status_code=500, detail=f"Error reading component example: {str(e)}")
else:
raise HTTPException(status_code=404, detail=f"Component example '{component_name}' not found")
def get_hook_example(hook_name):
"""Get a hook example"""
# First try direct match in the root hooks directory
try:
return get_example_content("hooks", f"{hook_name}.ts")
except HTTPException as e:
if e.status_code != 404:
raise e
# If not found, search in subdirectories
hooks_dir = os.path.join(CODE_EXAMPLES_DIR, "react-native", "hooks")
hook_path = find_file_in_subdirectories(hooks_dir, hook_name, [".ts", ".tsx", ".js"])
if hook_path:
try:
with open(hook_path, 'r') as f:
return f.read()
except Exception as e:
logging.error(f"Error reading hook example {hook_path}: {e}")
raise HTTPException(status_code=500, detail=f"Error reading hook example: {str(e)}")
else:
raise HTTPException(status_code=404, detail=f"Hook example '{hook_name}' not found")
def get_service_example(service_name):
"""Get a service example"""
# First try direct match in the root services directory
try:
return get_example_content("services", f"{service_name}.ts")
except HTTPException as e:
if e.status_code != 404:
raise e
# If not found, search in subdirectories
services_dir = os.path.join(CODE_EXAMPLES_DIR, "react-native", "services")
service_path = find_file_in_subdirectories(services_dir, service_name, [".ts", ".js"])
if service_path:
try:
with open(service_path, 'r') as f:
return f.read()
except Exception as e:
logging.error(f"Error reading service example {service_path}: {e}")
raise HTTPException(status_code=500, detail=f"Error reading service example: {str(e)}")
else:
raise HTTPException(status_code=404, detail=f"Service example '{service_name}' not found")
def get_screen_example(screen_name):
"""Get a screen example"""
# First try direct match in the root screens directory
try:
return get_example_content("screens", f"{screen_name}.tsx")
except HTTPException as e:
if e.status_code != 404:
raise e
# If not found, search in subdirectories
screens_dir = os.path.join(CODE_EXAMPLES_DIR, "react-native", "screens")
screen_path = find_file_in_subdirectories(screens_dir, screen_name, [".tsx", ".jsx"])
if screen_path:
try:
with open(screen_path, 'r') as f:
return f.read()
except Exception as e:
logging.error(f"Error reading screen example {screen_path}: {e}")
raise HTTPException(status_code=500, detail=f"Error reading screen example: {str(e)}")
else:
raise HTTPException(status_code=404, detail=f"Screen example '{screen_name}' not found")
def get_theme_example(theme_name):
"""Get a theme example"""
# First try direct match in the root theme directory
try:
return get_example_content("theme", f"{theme_name}.ts")
except HTTPException as e:
if e.status_code != 404:
raise e
# If not found, search in subdirectories
theme_dir = os.path.join(CODE_EXAMPLES_DIR, "react-native", "theme")
theme_path = find_file_in_subdirectories(theme_dir, theme_name, [".ts", ".js"])
if theme_path:
try:
with open(theme_path, 'r') as f:
return f.read()
except Exception as e:
logging.error(f"Error reading theme example {theme_path}: {e}")
raise HTTPException(status_code=500, detail=f"Error reading theme example: {str(e)}")
else:
raise HTTPException(status_code=404, detail=f"Theme example '{theme_name}' not found")
def list_available_examples():
"""List all available code examples by category"""
examples = {}
# Base path for examples
react_native_dir = os.path.join(CODE_EXAMPLES_DIR, "react-native")
try:
for category in os.listdir(react_native_dir):
category_path = os.path.join(react_native_dir, category)
if os.path.isdir(category_path):
examples[category] = []
# Handle screens directory with nested structure
if category == "screens":
# Walk through all subdirectories
for root, dirs, files in os.walk(category_path):
for file in files:
if file.endswith(".tsx") or file.endswith(".jsx"):
# Get relative path from the screens directory
rel_path = os.path.relpath(os.path.join(root, file), category_path)
# Add to examples list
examples[category].append(rel_path)
else:
# For other categories, just list files in the directory
for file in os.listdir(category_path):
if os.path.isfile(os.path.join(category_path, file)):
examples[category].append(file)
except Exception as e:
logging.error(f"Error listing code examples: {e}")
return examples
if __name__ == "__main__":
# Get transport configuration from environment variables or use defaults
transport = os.environ.get("MCP_TRANSPORT", "http")
host = os.environ.get("MCP_HOST", "0.0.0.0")
port = int(os.environ.get("MCP_PORT", "8000"))
# Print authentication status
if AUTH_ENABLED:
logging.info(f"Authentication enabled for user: {AUTH_USERNAME}")
else:
logging.info("Authentication disabled")
# Run the server
logging.info(f"Starting MCP server with {transport} transport on {host}:{port}")
uvicorn.run(app, host=host, port=port)