http_server.py•13.9 kB
"""
HTTP wrapper for docs-mcp MCP server enabling ChatGPT integration via MCP protocol.
Simplified version that works without importing server.py (MCP server).
Provides /health and /mcp endpoints. /tools endpoint disabled until we solve the server.py import issue.
"""
print("=" * 80)
print("HTTP_SERVER STARTING (Simplified Version)")
print("=" * 80)
import json
import logging
import os
import sys
from datetime import datetime
from typing import Any, Dict, Optional, Tuple
print("Standard library imports complete")
from flask import Flask, jsonify, request
print("Flask imported successfully")
# Import dependencies with graceful fallbacks
print("Attempting to import TOOL_HANDLERS...")
try:
from tool_handlers import TOOL_HANDLERS
print(f"SUCCESS: TOOL_HANDLERS imported ({len(TOOL_HANDLERS)} tools)")
except ImportError as e:
print(f"ERROR: Could not import TOOL_HANDLERS: {e}")
TOOL_HANDLERS = {}
print("Attempting to import logger...")
try:
from logger_config import logger
print("SUCCESS: logger imported")
except ImportError as e:
print(f"ERROR: Could not import logger: {e}")
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('http_server')
print("All imports complete")
# ============================================================================
# MCP PROTOCOL HELPER FUNCTIONS
# ============================================================================
def _build_mcp_tools_list() -> list:
"""Build MCP-format tools list from TOOL_HANDLERS."""
tools = []
# Tool metadata (manually defined for now)
tool_descriptions = {
'list_templates': 'List all available POWER framework documentation templates',
'get_template': 'Retrieve the content of a specific documentation template',
'generate_foundation_docs': 'Generate all foundation documentation for a project',
'generate_individual_doc': 'Generate a single documentation file',
'get_changelog': 'Query project changelog with optional filters',
'add_changelog_entry': 'Add a new entry to the project changelog',
'update_changelog': 'Agentic workflow to update changelog based on recent changes',
'generate_quickref_interactive': 'Generate universal quickref guide via interactive interview',
'establish_standards': 'Extract UI/UX/behavior standards from codebase',
'audit_codebase': 'Audit codebase for standards compliance',
'check_consistency': 'Quick consistency check on modified files',
'get_planning_template': 'Get implementation planning template',
'analyze_project_for_planning': 'Analyze project for implementation planning',
'create_plan': 'Create implementation plan',
'validate_implementation_plan': 'Validate implementation plan quality',
'generate_plan_review_report': 'Generate markdown review report',
'inventory_manifest': 'Generate comprehensive file inventory',
'dependency_inventory': 'Analyze project dependencies',
'api_inventory': 'Discover API endpoints',
'database_inventory': 'Discover database schemas',
'config_inventory': 'Discover configuration files',
'test_inventory': 'Discover test files and coverage',
'documentation_inventory': 'Discover documentation files'
}
for tool_name in TOOL_HANDLERS.keys():
tools.append({
'name': tool_name,
'description': tool_descriptions.get(tool_name, f'{tool_name} tool'),
'inputSchema': {
'type': 'object',
'properties': {
'project_path': {
'type': 'string',
'description': 'Absolute path to project directory'
}
},
'required': ['project_path']
}
})
return tools
def _handle_search(query: str) -> dict:
"""Handle search requests from ChatGPT."""
# Simple search implementation - returns available tools matching query
results = []
query_lower = query.lower()
tool_descriptions = {
'list_templates': 'List all available POWER framework documentation templates',
'generate_foundation_docs': 'Generate all foundation documentation',
'establish_standards': 'Extract coding standards from codebase',
'audit_codebase': 'Audit codebase for compliance',
'analyze_project_for_planning': 'Analyze project for planning',
'inventory_manifest': 'Generate file inventory'
}
for tool_name, description in tool_descriptions.items():
if query_lower in tool_name.lower() or query_lower in description.lower():
results.append({
'title': tool_name,
'description': description,
'uri': f'tool://{tool_name}'
})
return {'results': results}
def _handle_fetch(uri: str) -> dict:
"""Handle fetch requests from ChatGPT."""
# Extract tool name from URI (e.g., "tool://list_templates")
if uri.startswith('tool://'):
tool_name = uri[7:] # Remove "tool://" prefix
if tool_name == 'list_templates':
# Return list of available templates
return {
'content': 'Available templates: readme, architecture, api, components, schema, user-guide',
'mimeType': 'text/plain'
}
elif tool_name in TOOL_HANDLERS:
return {
'content': f'Tool {tool_name} is available. Use tools/call to execute it.',
'mimeType': 'text/plain'
}
return {
'content': f'Resource not found: {uri}',
'mimeType': 'text/plain'
}
# ============================================================================
# APPLICATION FACTORY
# ============================================================================
def create_app() -> Flask:
"""Create and configure Flask application."""
app = Flask(__name__)
app.config['JSON_SORT_KEYS'] = False
app.config['JSONIFY_PRETTYPRINT_REGULAR'] = True
@app.route('/health', methods=['GET'])
def health() -> Tuple[Dict[str, Any], int]:
"""Health check endpoint."""
return jsonify({
'status': 'operational',
'timestamp': datetime.utcnow().isoformat() + 'Z',
'version': '2.0.0',
'tools_available': len(TOOL_HANDLERS)
}), 200
@app.route('/mcp', methods=['POST'])
def mcp_endpoint() -> Tuple[Dict[str, Any], int]:
"""Main MCP endpoint accepting JSON-RPC 2.0 requests."""
try:
# Parse JSON
if not request.is_json:
return jsonify({
'jsonrpc': '2.0',
'id': None,
'error': {
'code': -32700,
'message': 'Parse error: Content-Type must be application/json'
}
}), 400
data = request.get_json(force=True)
# Validate JSON-RPC structure
if not isinstance(data, dict):
return jsonify({
'jsonrpc': '2.0',
'id': None,
'error': {'code': -32600, 'message': 'Invalid Request'}
}), 200
request_id = data.get('id')
method = data.get('method')
params = data.get('params', {})
if not method:
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'error': {'code': -32600, 'message': 'Missing method'}
}), 200
# ================================================================
# MCP PROTOCOL METHODS (Required by ChatGPT)
# ================================================================
# Handle initialize method
if method == 'initialize':
logger.info("MCP initialize request received")
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'result': {
'protocolVersion': '2025-03-26',
'capabilities': {
'tools': {'listChanged': True},
'resources': {}
},
'serverInfo': {
'name': 'docs-mcp',
'version': '2.0.0'
},
'instructions': 'docs-mcp provides 23 tools for documentation generation, changelog management, standards auditing, implementation planning, and project inventory analysis.'
}
}), 200
# Handle notifications/initialized (client ready signal)
if method == 'notifications/initialized':
logger.info("Client initialized notification received")
return '', 204 # No content response for notifications
# Handle tools/list method
if method == 'tools/list':
logger.info("MCP tools/list request received")
tools_list = _build_mcp_tools_list()
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'result': {'tools': tools_list}
}), 200
# Handle search method (required by ChatGPT connector)
if method == 'search':
logger.info(f"MCP search request: {params}")
query = params.get('query', '')
search_results = _handle_search(query)
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'result': search_results
}), 200
# Handle fetch method (required by ChatGPT connector)
if method == 'fetch':
logger.info(f"MCP fetch request: {params}")
uri = params.get('uri', '')
fetch_result = _handle_fetch(uri)
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'result': fetch_result
}), 200
# ================================================================
# TOOL EXECUTION
# ================================================================
# Check if method exists in tool handlers
if method not in TOOL_HANDLERS:
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'error': {
'code': -32601,
'message': f'Method not found: {method}'
}
}), 200
# Execute tool handler
import asyncio
handler = TOOL_HANDLERS[method]
if asyncio.iscoroutinefunction(handler):
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(handler(params))
else:
result = handler(params)
# Format response
response_data = _format_tool_response(result)
return jsonify({
'jsonrpc': '2.0',
'id': request_id,
'result': response_data
}), 200
except Exception as e:
logger.error(f"MCP endpoint error: {str(e)}")
return jsonify({
'jsonrpc': '2.0',
'id': data.get('id') if 'data' in locals() else None,
'error': {
'code': -32603,
'message': 'Internal error',
'data': {'details': str(e)}
}
}), 500
@app.after_request
def add_cors_headers(response):
"""Add CORS headers for development."""
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS'
response.headers['Access-Control-Allow-Headers'] = 'Content-Type'
return response
return app
def _format_tool_response(result: Any) -> Any:
"""Format tool response for JSON serialization."""
if isinstance(result, list):
formatted = []
for item in result:
if hasattr(item, 'type') and hasattr(item, 'text'):
formatted.append({'type': item.type, 'text': item.text})
elif isinstance(item, dict):
formatted.append(item)
else:
formatted.append(str(item))
return formatted
return result
# ============================================================================
# CREATE APP INSTANCE FOR GUNICORN
# ============================================================================
print("=" * 80)
print("Creating Flask app...")
try:
app = create_app()
print(f"SUCCESS: Flask app created: {app}")
print(f"Tools available: {len(TOOL_HANDLERS)}")
print("=" * 80)
print("HTTP_SERVER READY")
print("=" * 80)
except Exception as e:
print("!" * 80)
print(f"CRITICAL ERROR: {e}")
import traceback
traceback.print_exc()
print("!" * 80)
# Fallback
app = Flask(__name__)
@app.route('/health')
def health():
return jsonify({'status': 'error', 'message': str(e)}), 503
if __name__ == '__main__':
port = int(os.environ.get('PORT', 5000))
logger.info(f"Starting HTTP server on port {port}")
app.run(host='0.0.0.0', port=port)