Financial Data MCP Server

#!/usr/bin/env python3 """ Python HTTP Bridge for MCP Financial Server This creates HTTP endpoints that call your MCP server tools """ from flask import Flask, request, jsonify from flask_cors import CORS import subprocess import json import os import sys from typing import Dict, Any app = Flask(__name__) CORS(app) # Enable CORS for web interface # Configuration - Update these paths MCP_SERVER_SCRIPT = "/Users/jacobg/projects/finance_mcp_server/financial_mcp_server.py" # Update this path PYTHON_EXECUTABLE = sys.executable # Uses current Python environment class MCPBridge: def __init__(self, server_script: str): self.server_script = server_script def make_json_serializable(self, obj): """Convert non-JSON serializable objects to serializable format""" if isinstance(obj, dict): return {k: self.make_json_serializable(v) for k, v in obj.items()} elif isinstance(obj, list): return [self.make_json_serializable(item) for item in obj] elif isinstance(obj, (bool, int, float, str)): return obj elif obj is None: return None else: # Convert other types to string return str(obj) def call_mcp_tool(self, tool_name: str, parameters: Dict[str, Any] = None) -> Dict[str, Any]: """Call an MCP tool and return the result""" if parameters is None: parameters = {} try: # Prepare the command to run your MCP server cmd = [PYTHON_EXECUTABLE, self.server_script] # MCP requires proper initialization sequence # 1. Initialize request init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "clientInfo": { "name": "mcp-http-bridge", "version": "1.0.0" } } } # 2. Initialized notification (required after initialize) initialized_notification = { "jsonrpc": "2.0", "method": "notifications/initialized", "params": {} } # 3. Tool call request tool_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": tool_name, "arguments": parameters } } # 4. Prepare input with all three messages input_data = (json.dumps(init_request) + "\n" + json.dumps(initialized_notification) + "\n" + json.dumps(tool_request) + "\n") # Run the MCP server result = subprocess.run( cmd, input=input_data, capture_output=True, text=True, timeout=30 # 30 second timeout ) if result.returncode != 0: raise Exception(f"MCP server error: {result.stderr}") # Parse the JSON-RPC responses # We expect two responses: one for initialize, one for tools/call output_lines = result.stdout.strip().split('\n') tool_response = None for line in output_lines: line = line.strip() if line and line.startswith('{'): try: response_data = json.loads(line) # Check if it's the tool call response (id=2) if response_data.get('id') == 2: if 'error' in response_data: raise Exception(f"MCP server returned error: {response_data['error']}") elif 'result' in response_data: tool_response = response_data['result'] break except json.JSONDecodeError: continue if tool_response is None: # If no tool response found, return debug info raise Exception(f"No valid tool response found. Raw output: {result.stdout}") # Clean the response to ensure JSON serializability tool_response = self.make_json_serializable(tool_response) return tool_response except subprocess.TimeoutExpired: raise Exception("MCP server call timed out") except Exception as e: raise Exception(f"MCP server call failed: {str(e)}") # Initialize the bridge mcp_bridge = MCPBridge(MCP_SERVER_SCRIPT) # HTTP endpoints for each MCP tool @app.route('/tools/get_market_overview', methods=['POST']) def get_market_overview(): try: result = mcp_bridge.call_mcp_tool('get_market_overview', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/get_stock_info', methods=['POST']) def get_stock_info(): try: result = mcp_bridge.call_mcp_tool('get_stock_info', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/load_portfolio', methods=['POST']) def load_portfolio(): try: result = mcp_bridge.call_mcp_tool('load_portfolio', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/analyze_portfolio', methods=['POST']) def analyze_portfolio(): try: result = mcp_bridge.call_mcp_tool('analyze_portfolio', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/portfolio_performance', methods=['POST']) def portfolio_performance(): try: result = mcp_bridge.call_mcp_tool('portfolio_performance', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/get_earnings_calendar', methods=['POST']) def get_earnings_calendar(): try: result = mcp_bridge.call_mcp_tool('get_earnings_calendar', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/get_analyst_changes', methods=['POST']) def get_analyst_changes(): try: result = mcp_bridge.call_mcp_tool('get_analyst_changes', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/tools/generate_macd_chart', methods=['POST']) def generate_macd_chart(): try: result = mcp_bridge.call_mcp_tool('generate_macd_chart', request.json or {}) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 # Health check endpoint @app.route('/health', methods=['GET']) def health_check(): return jsonify({ 'status': 'healthy', 'server': 'MCP HTTP Bridge', 'mcp_server': MCP_SERVER_SCRIPT }) # Debug endpoint to list available tools @app.route('/debug/list_tools', methods=['GET']) def debug_list_tools(): try: # Send initialize + initialized + tools/list requests cmd = [PYTHON_EXECUTABLE, MCP_SERVER_SCRIPT] init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "clientInfo": {"name": "debug", "version": "1.0.0"} } } initialized_notification = { "jsonrpc": "2.0", "method": "notifications/initialized", "params": {} } list_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} } input_data = (json.dumps(init_request) + "\n" + json.dumps(initialized_notification) + "\n" + json.dumps(list_request) + "\n") result = subprocess.run(cmd, input=input_data, capture_output=True, text=True, timeout=30) return jsonify({ 'stdout': result.stdout, 'stderr': result.stderr, 'returncode': result.returncode, 'input_sent': input_data }) except Exception as e: return jsonify({'error': str(e)}), 500 # Root endpoint with API info @app.route('/', methods=['GET']) def root(): return jsonify({ 'message': 'MCP HTTP Bridge Server', 'available_endpoints': [ '/tools/get_market_overview', '/tools/get_stock_info', '/tools/load_portfolio', '/tools/analyze_portfolio', '/tools/portfolio_performance', '/tools/get_earnings_calendar', '/tools/get_analyst_changes', '/tools/generate_macd_chart', '/health', '/debug/list_tools' ], 'usage': 'Send POST requests with JSON data to tool endpoints' }) if __name__ == '__main__': print("🚀 Starting MCP HTTP Bridge Server...") print(f"📁 MCP Server Script: {MCP_SERVER_SCRIPT}") print(f"🐍 Python Executable: {PYTHON_EXECUTABLE}") print("🌐 Server will be available at: http://localhost:3001") print("💡 Update your web interface MCP_SERVER_URL to: http://localhost:3001") # Check if MCP server script exists if not os.path.exists(MCP_SERVER_SCRIPT): print(f"⚠️ Warning: MCP server script not found at {MCP_SERVER_SCRIPT}") print(" Please update the MCP_SERVER_SCRIPT variable with the correct path") app.run(host='0.0.0.0', port=3001, debug=True)