authentication addeed

app.py

@@ -96,8 +96,15 @@ if __name__ == "__main__":
     try:
         # Add web routes for landing page and API key generation
         from starlette.responses import HTMLResponse, JSONResponse
+        from starlette.requests import Request
         from database.api_keys import generate_api_key as db_generate_api_key
 
+        # NOTE: Custom middleware for FastMCP has known limitations (GitHub issue #817)
+        # Headers/state set in middleware are NOT accessible inside FastMCP tools.
+        # For local development, use SKIP_AUTH=true with ENV=development in .env
+        # For production, API key validation happens via extract_api_key_from_request() in server.py
+        logger.info("[Auth] Using FastMCP built-in request handling for authentication")
+
         @mcp.custom_route("/", methods=["GET"])
         async def landing_page(request):
             """Landing page with MCP connection information"""

server.py

@@ -45,6 +45,25 @@ logger = logging.getLogger(__name__)
 # Store API key per request using context variable
 _current_api_key: ContextVar[str] = ContextVar('api_key', default=None)
 
+# Session-to-API-key store: maps session_id -> api_key
+# This allows tool calls to authenticate using the API key from the SSE connection
+_session_auth_store: dict[str, str] = {}
+
+def store_session_api_key(session_id: str, api_key: str):
+    """Store API key for a session (called when SSE connection is made)"""
+    _session_auth_store[session_id] = api_key
+    logger.debug(f"Stored API key for session {session_id[:16]}...")
+
+def get_api_key_from_session(session_id: str) -> str:
+    """Get API key from session store"""
+    return _session_auth_store.get(session_id)
+
+def cleanup_session(session_id: str):
+    """Remove session from store when disconnected"""
+    if session_id in _session_auth_store:
+        del _session_auth_store[session_id]
+        logger.debug(f"Cleaned up session {session_id[:16]}...")
+
 def get_api_key_from_context() -> str:
     """
     Get API key from the current request context.
@@ -56,15 +75,33 @@ def extract_api_key_from_request():
     """
     Extract API key from HTTP request and store in context variable.
     Called at the start of each tool execution.
+
+    Checks multiple sources:
+    1. api_key query parameter in current request
+    2. session_id query parameter -> lookup in session store
     """
     try:
         from fastmcp.server.dependencies import get_http_request
         request = get_http_request()
+
+        # METHOD 1: Direct api_key in query params
         api_key = request.query_params.get('api_key')
         if api_key:
             _current_api_key.set(api_key)
             logger.debug(f"API key extracted from request: {api_key[:10]}...")
             return api_key
+
+        # METHOD 2: Look up by session_id (for MCP tool calls)
+        session_id = request.query_params.get('session_id')
+        if session_id:
+            api_key = get_api_key_from_session(session_id)
+            if api_key:
+                _current_api_key.set(api_key)
+                logger.debug(f"API key found via session {session_id[:16]}...")
+                return api_key
+            else:
+                logger.debug(f"No API key found for session {session_id[:16]}...")
+
     except RuntimeError:
         # No HTTP request available (e.g., stdio transport)
         logger.debug("No HTTP request available for API key extraction")
@@ -124,7 +161,9 @@ def get_authenticated_user():
 
         # METHOD 3: Development bypass mode (local testing only)
         # SECURITY: Only allow SKIP_AUTH in development environments
-        env = os.getenv("ENV", "production").lower()
+        # Check both ENV and ENVIRONMENT variables for compatibility
+        env = os.getenv("ENV") or os.getenv("ENVIRONMENT", "production")
+        env = env.lower()
         skip_auth = os.getenv("SKIP_AUTH", "false").lower() == "true"
 
         if skip_auth: