consent elicitation WIP

Author: nemozak1

.env.example

@@ -28,6 +28,15 @@ AUTH_PROVIDER=obp-oidc
 # For bearer-only: Set this to accept tokens from OBP-OIDC
 OBP_OIDC_ISSUER_URL=http://localhost:9000/obp-oidc
 
+# This setting controls how we authorize API requests to OBP after authentication.
+# Options:
+#   - "oauth":  Use OAuth the oauth access token. Best for when using an MCP client that cannot perform the 
+#               consent flow itself (e.g. VS Code extension, Claude Desktop).
+#   - "consent": Elicit user consent before each API call. Best for MCP clients like Opey that will support
+#               user consent flows.
+#   - "none":   No authorization method is used. OBP-API calls are made but with limited access (only public endpoints).
+OBP_AUTHORIZATION_VIA="none" # Options: "oauth", "consent", "none"
+
 # Multi-issuer bearer-only mode:
 # Set BOTH KEYCLOAK_REALM_URL and OBP_OIDC_ISSUER_URL with AUTH_PROVIDER=bearer-only
 # to accept tokens from either identity provider. The server will route validation

src/mcp_server_obp/elicitation.py

@@ -1,22 +1,42 @@
+import json
+
 from fastmcp import Context
+from fastmcp.server.elicitation import (
+    AcceptedElicitation,
+    DeclinedElicitation,
+    CancelledElicitation
+)
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, List
 
 from src.tools.endpoint_index import get_endpoint_index
+from src.tools import EndpointSchema, RoleRequirement
 
 @dataclass
 class ApprovalConsent:
-    consent_id: str
-    
-    
-def get_roles_for_endpoint(endpoint_id: str) -> list[str]:
+    consent_jwt: str
+
+
+async def elicit_consent(ctx: Context, endpoint: EndpointSchema) -> AcceptedElicitation[ApprovalConsent] | DeclinedElicitation | CancelledElicitation:
     """
-    Gets the required entitlements for a given OBP endpoint from its ID.
+    Elicit user consent for accessing a given OBP endpoint.
     """
 
-    index = get_endpoint_index()
-    endpoint = index.get_endpoint_schema(endpoint_id)
-    if not endpoint:
-        raise ValueError(f"Endpoint with ID '{endpoint_id}' not found in index.")
+    prompt = f"Endpoint with operation ID '{endpoint.operation_id}' needs your permission to run."
+    
+    # This message should be json-serializable so we can extract information at the frontend
+    message_json = {
+        "operation_id": endpoint.operation_id,
+        "message_prompt": prompt,
+        "required_roles": [role.model_dump() for role in endpoint.roles] # we will use these to construct our narrowly-scoped consent
+    }
+    
+    response = await ctx.elicit(
+        message=json.dumps(message_json),
+        response_type=ApprovalConsent,
+    )
+    
+    return response
+    
 
-    return endpoint.roles
+    

src/mcp_server_obp/headers.py

@@ -0,0 +1,31 @@
+import os
+from fastmcp import Context
+from fastmcp.server.dependencies import get_access_token
+from dotenv import load_dotenv
+from mcp_server_obp.elicitation import elicit_consent
+from src.tools import EndpointSchema
+
+load_dotenv()
+
+async def construct_obp_headers(ctx: Context, headers: dict, endpoint: EndpointSchema) -> dict:
+    """
+    Construct headers for OBP API requests, eliciting user consent if necessary.
+    """
+    authorization_method = os.getenv("OBP_AUTHORIZATION_VIA", "none").lower()
+    if not authorization_method in ["oauth", "consent", "none"]:
+        raise ValueError(f"Invalid OBP_AUTHORIZATION_VIA value: {authorization_method}")
+    
+    if authorization_method == "oauth":
+        # Use OAuth access token from context
+        access_token = get_access_token()
+        if not access_token:
+            raise ValueError("No access token available in context for OAuth authorization.")
+        
+        obp_headers = {
+            **headers,
+            "Authorization": f"Bearer {access_token}"
+        }
+        return obp_headers
+    elif authorization_method == "consent":
+        # Elicit user consent for the endpoint
+        consent_response = await elicit_consent(ctx, endpo)

src/mcp_server_obp/server.py

@@ -24,22 +24,25 @@
 if str(_project_root) not in sys.path:
     sys.path.insert(0, str(_project_root))
 
-from fastmcp import FastMCP
+from fastmcp import FastMCP, Context
 from fastmcp.server.dependencies import get_access_token
+from fastmcp.server.elicitation import AcceptedElicitation, DeclinedElicitation, CancelledElicitation
 
 from src.tools.endpoint_index import get_endpoint_index
 from src.tools.glossary_index import get_glossary_index
 
 from mcp_server_obp.lifespan import lifespan
 from mcp_server_obp.auth import get_auth_provider
+from mcp_server_obp.elicitation import elicit_consent, ApprovalConsent
 
 logger = logging.getLogger(__name__)
 
 # Initialize authentication provider based on environment configuration
 # Supports: keycloak, obp-oidc, or none (disabled)
 # See auth.py for configuration details
 auth = get_auth_provider()
-    
+
+
 
 mcp = FastMCP(
     "Open Bank Project",
@@ -142,12 +145,13 @@ def get_endpoint_schema(endpoint_id: str) -> str:
 
 
 @mcp.tool()
-def call_obp_api(
+async def call_obp_api(
+    ctx: Context,
     endpoint_id: str,
     path_params: dict = None,
     query_params: dict = None,
     body: dict = None,
-    headers: dict = None
+    headers: dict = None,
 ) -> str:
     """
     Execute an OBP API request to a specific endpoint.
@@ -173,7 +177,7 @@ def call_obp_api(
         index = get_endpoint_index()
 
         # Get endpoint info from lightweight index
-        endpoint = index.get_endpoint_by_id(endpoint_id)
+        endpoint = index.get_endpoint_schema(endpoint_id)
         if not endpoint:
             return json.dumps({
                 "error": f"Endpoint '{endpoint_id}' not found",
@@ -205,6 +209,47 @@ def call_obp_api(
         method = endpoint.method.value  # HttpMethod enum value
         request_headers = headers or {}
 
+        auth_method = os.getenv("OBP_AUTHORIZATION_VIA", "none").lower()
+        match auth_method:
+            case "oauth":
+                logger.info("Using OAuth authorization method for OBP API call")
+                access_token = get_access_token()
+                if not access_token:
+                    logger.error("No access token available in context for OAuth authorization.")
+                else:
+                    # TODO: Elicit a simple approval here to confirm tool use?
+                    request_headers["Authorization"] = f"Bearer {access_token}"
+            case "consent":
+                # Elicit user consent
+                logger.info("Eliciting user consent for OBP API call")
+                from mcp_server_obp.elicitation import elicit_consent
+                
+                response = await elicit_consent(ctx, endpoint)
+                match response:
+                    case AcceptedElicitation(data=consent):
+                        # For now just assume the consent is valid, but we might want to decode the JWT in the future
+                        logger.info("User consent accepted for OBP API call")
+                        request_headers["Consent-JWT"] = consent.consent_jwt
+                    case DeclinedElicitation():
+                        logger.info("User declined consent for OBP API call")
+                        return json.dumps({
+                            "error": "User declined consent for this endpoint"
+                        }, indent=2)
+                    case CancelledElicitation():
+                        logger.info("User cancelled consent elicitation for OBP API call")
+                        return json.dumps({
+                            "error": "User cancelled operation."
+                        }, indent=2)
+            case "none":
+                # No authorization
+                logger.info("No authorization method used for OBP API call")
+                pass
+            case _:
+                logger.error(f"Invalid OBP_AUTHORIZATION_VIA value: {auth_method}")
+                return json.dumps({
+                    "error": f"Internal server error."
+                }, indent=2)
+                
         # Make the request
         logger.info(f"Calling OBP API: {method} {url}")