import os import logging from dotenv import load_dotenv from flask import Flask, jsonify, request, send_from_directory from oauth import oauth_bp, validate_token load_dotenv() # ---------- Logging ---------- os.makedirs('logs', exist_ok=True) logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', handlers=[ logging.FileHandler('logs/app.log'), logging.StreamHandler() ] ) logger = logging.getLogger('experiment-analyzer-mcp') PROTOCOL_VERSION = "2024-11-05" SERVER_NAME = "experiment-analyzer" SERVER_VERSION = "0.2.0" app = Flask(__name__) app.register_blueprint(oauth_bp) # ---------- Tool registry ---------- TOOLS = {} def register_tool(name, description, input_schema): def decorator(func): TOOLS[name] = { "func": func, "description": description, "inputSchema": input_schema, } return func return decorator # ---------- Tool definitions ---------- @register_tool( name="analyze_experiment", description=( "Generate a post-experiment analysis report for a HomeBuddy A/B test. " "Pulls exposure data from Snowflake, prelead/lead/appointment metrics, " "runs SRM check, computes deltas vs control with confidence intervals, " "and produces a PDF + CSV. Use when the user asks to 'generate " "post-analysis report', 'analyze experiment', 'run post-mortem on', " "or similar phrasing followed by an experiment ID." ), input_schema={ "type": "object", "properties": { "experiment_id": { "type": "string", "description": ( "Statsig experiment name, e.g. " "'08_05_2026_sbg_old_wrapper_test_post_fix'." ), }, "exclusions": { "type": "array", "items": {"type": "string", "format": "date"}, "description": ( "Optional list of YYYY-MM-DD dates to exclude " "(known outages, etc.)." ), }, "window_start": { "type": "string", "format": "date", "description": "Optional override for analysis start (YYYY-MM-DD).", }, "window_end": { "type": "string", "format": "date", "description": "Optional override for analysis end (YYYY-MM-DD).", }, }, "required": ["experiment_id"], }, ) def analyze_experiment(experiment_id, exclusions=None, window_start=None, window_end=None): logger.info( f"analyze_experiment called: experiment_id={experiment_id} " f"window={window_start}..{window_end} exclusions={exclusions}" ) return ( f"Stub response from analizaweb.pl MCP server.\n\n" f"Parsed input:\n" f"- experiment_id: {experiment_id}\n" f"- window_start: {window_start or '(experiment default)'}\n" f"- window_end: {window_end or '(experiment default)'}\n" f"- exclusions: {exclusions if exclusions else 'none'}\n\n" f"Next steps: implement Statsig fetch, Snowflake queries, stats, " f"PDF/CSV generation, Teams notification." ) # ---------- Plain routes ---------- @app.route("/healthz") def healthz(): return jsonify({"status": "ok", "service": "experiment-analyzer-mcp"}) @app.route("/") def index(): return jsonify({"service": "experiment-analyzer-mcp", "version": SERVER_VERSION}) @app.route("/output/") def output_file(filename): return send_from_directory( os.path.join(os.path.dirname(__file__), "output"), filename, ) # ---------- MCP endpoint ---------- def _jsonrpc_response(req_id, result=None, error=None): payload = {"jsonrpc": "2.0", "id": req_id} if error is not None: payload["error"] = error else: payload["result"] = result return jsonify(payload) @app.route("/mcp", methods=["POST"]) @app.route("/mcp/", methods=["POST"]) def mcp_endpoint(): # ---- Bearer auth (OAuth-issued tokens) ---- auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer "): resp = jsonify({ "error": "unauthorized", "error_description": "Bearer token required", }) resp.status_code = 401 resp.headers["WWW-Authenticate"] = ( 'Bearer realm="mcp", ' 'resource_metadata="https://mcp.analizaweb.pl/.well-known/oauth-protected-resource"' ) return resp token_str = auth[7:] if not validate_token(token_str): resp = jsonify({"error": "invalid_token"}) resp.status_code = 401 resp.headers["WWW-Authenticate"] = ( 'Bearer realm="mcp", error="invalid_token", ' 'resource_metadata="https://mcp.analizaweb.pl/.well-known/oauth-protected-resource"' ) return resp # ---- Parse JSON-RPC ---- try: req = request.get_json(force=True) except Exception as e: logger.warning(f"Bad JSON: {e}") return _jsonrpc_response(None, error={"code": -32700, "message": "Parse error"}) method = req.get("method") req_id = req.get("id") params = req.get("params", {}) or {} logger.info(f"MCP request: method={method} id={req_id}") # Notifications (no id) get 204 with empty body if req_id is None and method and method.startswith("notifications/"): return "", 204 # ---- Method dispatch ---- if method == "initialize": return _jsonrpc_response(req_id, result={ "protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, }) if method == "tools/list": tools_list = [ { "name": name, "description": info["description"], "inputSchema": info["inputSchema"], } for name, info in TOOLS.items() ] return _jsonrpc_response(req_id, result={"tools": tools_list}) if method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) or {} if tool_name not in TOOLS: return _jsonrpc_response(req_id, error={ "code": -32601, "message": f"Unknown tool: {tool_name}", }) try: result_text = TOOLS[tool_name]["func"](**arguments) return _jsonrpc_response(req_id, result={ "content": [{"type": "text", "text": str(result_text)}], "isError": False, }) except TypeError as e: return _jsonrpc_response(req_id, error={ "code": -32602, "message": f"Invalid params: {e}", }) except Exception as e: logger.exception(f"Tool {tool_name} crashed") return _jsonrpc_response(req_id, result={ "content": [{"type": "text", "text": f"Tool error: {e}"}], "isError": True, }) # Unknown method return _jsonrpc_response(req_id, error={ "code": -32601, "message": f"Method not found: {method}", }) if __name__ == "__main__": app.run(debug=True)