bump: version 0.25.0 → 0.26.0

fix: Consolidate OAuth callbacks and implement PKCE for all flows

.github/workflows/docker-build-publish.yml

@@ -16,7 +16,7 @@ jobs:
 
       - name: Docker meta
         id: meta
-        uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f # v5
+        uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5
         with:
           # list of Docker images to use as base name for tags
           images: |

.github/workflows/test.yml

@@ -18,6 +18,9 @@ jobs:
       - name: Linting
         run: |
           uv run --frozen ruff check
+      - name: Linting
+        run: |
+          uv run --frozen ty check -- nextcloud_mcp_server
 
 
   integration-test:

.pre-commit-config.yaml

@@ -18,3 +18,9 @@ repos:
       entry: uv run ruff format
       language: system
       types: [python]
+    - id: ty-check
+      name: ty-check
+      language: system
+      types: [python]
+      exclude: tests/.*
+      entry: uv run ty check

CHANGELOG.md

@@ -1,3 +1,81 @@
+## v0.26.0 (2025-11-08)
+
+### Feat
+
+- add real elicitation integration test with python-sdk MCP client
+- unify session architecture and enhance login status visibility
+
+### Fix
+
+- Consolidate OAuth callbacks and implement PKCE for all flows
+
+## v0.25.0 (2025-11-05)
+
+### BREAKING CHANGE
+
+- All OAuth deployments must be reconfigured to specify
+resource URIs (NEXTCLOUD_MCP_SERVER_URL and NEXTCLOUD_RESOURCE_URI) and
+choose between multi-audience or token exchange mode.
+
+### Feat
+
+- Implement ADR-005 unified token verifier to eliminate token passthrough vulnerability
+
+### Fix
+
+- Implement proper OAuth resource parameters and PRM-based discovery
+- Simplify token verifier to be RFC 7519 compliant
+- Use Keycloak client ID for NEXTCLOUD_RESOURCE_URI in token exchange
+- Correct OAuth token audience validation for multi-audience mode
+
+### Refactor
+
+- Eliminate duplicate validation logic in UnifiedTokenVerifier
+
+## v0.24.1 (2025-11-04)
+
+### Fix
+
+- **deps**: update dependency mcp to >=1.20,<1.21
+
+## v0.24.0 (2025-11-04)
+
+### Feat
+
+- add scope protection to OAuth provisioning tools
+- enable authorization services for token exchange in Keycloak
+- implement scope-based audience mapping and RFC 9728 support
+- integrate token exchange into MCP server application
+- implement RFC 8693 Standard Token Exchange for Keycloak
+- Add userinfo route/page
+- add browser-based user info page with separate OAuth flow
+- Implement ADR-004 Progressive Consent foundation (partial)
+- Complete ADR-004 Progressive Consent OAuth flows implementation
+- Implement ADR-004 Progressive Consent foundation components
+- Implement ADR-004 Hybrid Flow with comprehensive integration tests
+
+### Fix
+
+- add missing await for get_nextcloud_client in capabilities resource
+- use valid Fernet encryption keys in token exchange tests
+- accept resource URL in token audience for Nextcloud JWT tokens
+- remove token-exchange-nextcloud scope and accept tokens without audience
+- move audience mapper from scope to nextcloud-mcp-server client
+- move token-exchange-nextcloud from default to optional scopes
+- restructure routes to prevent SessionAuthBackend from interfering with FastMCP OAuth
+- allow OAuth Bearer tokens on /mcp endpoint by excluding from session auth
+- correct OAuth token audience validation using RFC 8707 resource parameter
+- remove remaining references to deleted oauth_callback and oauth_token
+- remove Hybrid Flow, make Progressive Consent default (ADR-004)
+- browser OAuth userinfo endpoint and refresh token rotation
+- make ENABLE_PROGRESSIVE_CONSENT consistently opt-in (default false)
+- make provisioning checks opt-in (default false)
+- Disable Progressive Consent for mcp-oauth to enable Hybrid Flow tests
+
+### Refactor
+
+- integrate token exchange into unified get_client() pattern
+
 ## v0.23.0 (2025-11-03)
 
 ### Feat

CLAUDE.md

@@ -167,23 +167,35 @@ docker compose exec db mariadb -u root -ppassword nextcloud -e \
 
 ### Progressive Consent Architecture (ADR-004)
 
-**Status**: Always enabled in OAuth mode (default)
+**Important**: Progressive consent is a *mechanism* for granting access, not a feature flag. The architecture is always present in OAuth mode. Whether provisioning tools are available is controlled by `ENABLE_OFFLINE_ACCESS`.
 
 **What is Progressive Consent?**
 - Dual OAuth flow architecture that separates client authentication (Flow 1) from resource provisioning (Flow 2)
 - Flow 1: MCP client authenticates directly to IdP with resource scopes (notes:*, calendar:*, etc.)
   - Token audience: "mcp-server"
   - Client receives resource-scoped token for MCP session
-- Flow 2: Server explicitly provisions Nextcloud access via separate login
+- Flow 2: Server explicitly provisions Nextcloud access via separate login (only when `ENABLE_OFFLINE_ACCESS=true`)
   - Server requests: openid, profile, email, offline_access
   - Token audience: "nextcloud"
   - Server receives refresh token for offline access
   - Client never sees this token
 - Provides clear separation between session tokens and offline access tokens
 
+**Modes:**
+- **Pass-through mode** (`ENABLE_OFFLINE_ACCESS=false`, default):
+  - No Flow 2 provisioning
+  - Server uses client's token to access Nextcloud (pass-through)
+  - No provisioning tools available
+  - Suitable for stateless, client-driven operations
+- **Offline access mode** (`ENABLE_OFFLINE_ACCESS=true`):
+  - Flow 2 provisioning available
+  - Server stores refresh tokens for background operations
+  - Provisioning tools available: `provision_nextcloud_access`, `check_logged_in`
+  - Suitable for background jobs and server-initiated operations
+
 **When to use OAuth mode:**
 - Multi-user deployments
-- Background jobs requiring offline access
+- Background jobs requiring offline access (with `ENABLE_OFFLINE_ACCESS=true`)
 - Enhanced security with separate authorization contexts
 - Explicit user control over resource access
 

Dockerfile

@@ -1,4 +1,4 @@
-FROM ghcr.io/astral-sh/uv:0.9.7-python3.11-alpine@sha256:0006b77df7ebf46e68959fdc8d3af9d19f1adfae8c2e7e77907ad257e5d05be4
+FROM ghcr.io/astral-sh/uv:0.9.8-python3.11-alpine@sha256:6c842c49ad032f46b62f32a7e7779f45f12671a8e0d82ea24c766ab62d58b396
 
 # Install dependencies
 # 1. git (required for caldav dependency from git)

app-hooks/post-installation/10-install-oidc-app.sh

@@ -31,8 +31,10 @@ else
 fi
 
 # Configure OIDC Identity Provider with dynamic client registration enabled
-php /var/www/html/occ config:app:set oidc dynamic_client_registration --value='true'
+php /var/www/html/occ config:app:set oidc dynamic_client_registration --value='true' # NOTE: String
 php /var/www/html/occ config:app:set oidc proof_key_for_code_exchange --value=true --type=boolean
+php /var/www/html/occ config:app:set oidc allow_user_settings --value='enabled'
 php /var/www/html/occ config:app:set oidc default_token_type --value='jwt'
+php /var/www/html/occ config:app:set oidc default_resource_identifier --value='http://localhost:8080'
 
 echo "OIDC app installed and configured successfully"

app-hooks/post-installation/30-disable-welcome-wizard.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+php /var/www/html/occ config:app:set --value false firstrunwizard wizard_enabled

charts/nextcloud-mcp-server/Chart.yaml

@@ -2,8 +2,8 @@ apiVersion: v2
 name: nextcloud-mcp-server
 description: A Helm chart for Nextcloud MCP Server - enables AI assistants to interact with Nextcloud
 type: application
-version: 0.23.0
-appVersion: "0.23.0"
+version: 0.26.0
+appVersion: "0.26.0"
 keywords:
   - nextcloud
   - mcp

docker-compose.yml

@@ -17,17 +17,18 @@ services:
   # Note: Redis is an external service. You can find more information about the configuration here:
   # https://hub.docker.com/_/redis
   redis:
-    image: docker.io/library/redis:alpine@sha256:59b6e694653476de2c992937ebe1c64182af4728e54bb49e9b7a6c26614d8933
+    image: docker.io/library/redis:alpine@sha256:28c9c4d7596949a24b183eaaab6455f8e5d55ecbf72d02ff5e2c17fe72671d31
     restart: always
 
   app:
-    image: docker.io/library/nextcloud:32.0.1@sha256:1e4eae55eebe094cae6f9e7b6e0b4bccf4a4fe7b7e6f6f8f57010994b3b2ee42
+    image: docker.io/library/nextcloud:32.0.1@sha256:5b043f7ea2f609d5ff5635f475c30d303bec17775a5c3f7fa435e3818e669120
     restart: always
     ports:
       - 0.0.0.0:8080:80
     depends_on:
       - redis
       - db
+      - keycloak
     volumes:
       - nextcloud:/var/www/html
       - ./app-hooks:/docker-entrypoint-hooks.d:ro
@@ -95,6 +96,7 @@ services:
       # OIDC_CLIENT_ID not set - uses Dynamic Client Registration (DCR)
       - NEXTCLOUD_HOST=http://app:80
       - NEXTCLOUD_MCP_SERVER_URL=http://localhost:8001
+      - NEXTCLOUD_RESOURCE_URI=http://localhost:8080  # ADR-005: Nextcloud resource identifier for audience validation
       - NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8080
       - NEXTCLOUD_OIDC_SCOPES=openid profile email notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write todo:read todo:write
 
@@ -103,8 +105,9 @@ services:
       - TOKEN_ENCRYPTION_KEY=ESF1BvEQdGYsCluwMx9Cxvw3uh5pFowPH7Rg_nIliyo=
       - TOKEN_STORAGE_DB=/app/data/tokens.db
 
-      # ADR-004: Use Hybrid Flow (server intercepts OAuth callback)
-      # Set to false to enable Hybrid Flow tests - server stores refresh token and issues MCP codes
+      # ADR-005: Multi-audience mode (default - ENABLE_TOKEN_EXCHANGE=false)
+      # Tokens must contain BOTH MCP and Nextcloud audiences
+      # No token exchange needed - tokens work for both MCP auth and Nextcloud APIs
 
       # NO admin credentials - using OAuth with Dynamic Client Registration (DCR)
       # Client credentials registered via RFC 7591 and stored in volume
@@ -114,7 +117,7 @@ services:
       - oauth-tokens:/app/data
 
   keycloak:
-    image: quay.io/keycloak/keycloak:26.4.2
+    image: quay.io/keycloak/keycloak:26.4.4@sha256:c6459d5fae1b759f5d667ebdc6237ab3121379c3494e213898569014ede1846d
     command:
       - "start-dev"
       - "--import-realm"
@@ -158,6 +161,7 @@ services:
       # Nextcloud API endpoint (for accessing APIs with validated token)
       - NEXTCLOUD_HOST=http://app:80
       - NEXTCLOUD_MCP_SERVER_URL=http://localhost:8002
+      - NEXTCLOUD_RESOURCE_URI=nextcloud  # ADR-005: Keycloak uses client IDs as audiences, not URLs
       - NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8888/realms/nextcloud-mcp
 
       # Refresh token storage (ADR-002 Tier 1 & 2)
@@ -165,8 +169,11 @@ services:
       - TOKEN_ENCRYPTION_KEY=ESF1BvEQdGYsCluwMx9Cxvw3uh5pFowPH7Rg_nIliyo=
       - TOKEN_STORAGE_DB=/app/data/tokens.db
 
-      # Token exchange (RFC 8693) - convert aud:nextcloud-mcp-server → aud:nextcloud
+      # ADR-005: Token exchange mode (RFC 8693)
+      # Exchange MCP tokens (aud: nextcloud-mcp-server) for Nextcloud tokens (aud: http://localhost:8080)
+      # Provides strict audience separation between MCP session and Nextcloud API access
       - ENABLE_TOKEN_EXCHANGE=true
+      - TOKEN_EXCHANGE_CACHE_TTL=300  # Cache exchanged tokens for 5 minutes (default)
 
       # OAuth scopes (optional - uses defaults if not specified)
       - NEXTCLOUD_OIDC_SCOPES=openid profile email offline_access notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write todo:read todo:write

docs/ADR-006-progressive-consent-elicitation.md

@@ -0,0 +1,865 @@
+# ADR-006: Progressive Consent via URL Elicitation (SEP-1036)
+
+**Status**: Partially Implemented (Interim Workaround)
+**Date**: 2025-01-05 (Updated: 2025-01-07)
+**Related**: [SEP-1036](https://github.com/modelcontextprotocol/specification/pull/887), ADR-004
+**Depends On**: ADR-005 (token validation)
+
+## Context
+
+### What is Progressive Consent?
+
+**Progressive consent is a mechanism, not a feature**. It describes HOW users grant the MCP server access to Nextcloud resources through OAuth elicitation. The server can operate in two modes:
+
+1. **Pass-through mode (ENABLE_OFFLINE_ACCESS=false)**:
+   - No refresh tokens requested or stored
+   - Server passes through client's access token to Nextcloud
+   - No provisioning tools available
+   - Suitable for stateless, client-driven operations
+
+2. **Offline access mode (ENABLE_OFFLINE_ACCESS=true)**:
+   - Server requests `offline_access` scope and stores refresh tokens
+   - Enables background operations and server-initiated API calls
+   - Provisioning tools available (`provision_nextcloud_access`, `check_logged_in`)
+   - Requires explicit user consent via OAuth Flow 2
+
+**Single-user mode (BasicAuth)** doesn't use progressive consent at all - credentials are directly available.
+
+### Current User Experience Issues
+
+The current offline access provisioning flow (ADR-004) requires users to manually visit OAuth URLs returned by MCP tools. This creates a poor user experience:
+
+1. User calls `provision_nextcloud_access` tool
+2. Tool returns a URL as text in the response
+3. User must manually copy URL and open in browser
+4. No indication when provisioning is complete
+5. User must retry the original operation manually
+
+### SEP-1036: URL Mode Elicitation
+
+The MCP specification now supports **URL mode elicitation** ([SEP-1036](https://github.com/modelcontextprotocol/specification/pull/887)), which enables servers to:
+
+- Request out-of-band user interactions via secure URLs
+- Handle sensitive operations like OAuth flows without exposing credentials to the client
+- Provide progress tracking for async operations
+- Return errors that automatically trigger elicitation flows
+
+**Key benefits for progressive consent**:
+- **Automatic URL Opening**: Client opens URL in browser automatically (with user consent)
+- **Progress Tracking**: Server can notify client when provisioning is complete
+- **Error-Triggered Flows**: Server can return `ElicitationRequired` error to trigger provisioning
+- **Better UX**: User doesn't manually copy/paste URLs
+
+### Current Implementation Limitations
+
+The current progressive consent flow in `nextcloud_mcp_server/server/oauth_tools.py`:
+
+```python
+@mcp.tool(name="provision_nextcloud_access")
+async def tool_provision_access(ctx: Context) -> ProvisioningResult:
+    """Returns OAuth URL as text - user must manually open it."""
+    return ProvisioningResult(
+        success=True,
+        authorization_url=auth_url,  # User must copy this
+        message="Please visit the authorization URL..."
+    )
+```
+
+**Problems**:
+1. Manual URL handling (copy/paste)
+2. No progress tracking
+3. No automatic retry after provisioning
+4. Tool call required just to get URL
+5. No client integration (URL just displayed as text)
+
+## Decision
+
+We will **migrate progressive consent from manual tools to URL mode elicitation**, leveraging SEP-1036 for better user experience and OAuth security.
+
+### New Architecture: Elicitation-Driven Consent
+
+Instead of explicit tools, use **automatic elicitation** triggered by authorization errors:
+
+```
+User → Calls Nextcloud Tool → Server Checks Provisioning
+                                     ↓ Not Provisioned
+                                Error: ElicitationRequired
+                                     ↓
+                          Client Shows Consent UI
+                                     ↓ User Accepts
+                          Client Opens OAuth URL
+                                     ↓
+                          User Completes OAuth
+                                     ↓
+                          Server Sends Progress Update
+                                     ↓
+                      Original Tool Call Auto-Retries
+```
+
+### Mode 1: Elicitation-Required Error (Primary)
+
+When a tool requires provisioning, return an **ElicitationRequired error** (-32000):
+
+```python
+# In any Nextcloud tool decorated with @require_provisioning
+@mcp.tool()
+@require_provisioning  # New decorator
+async def nc_notes_list_notes(ctx: Context):
+    """List notes - auto-triggers provisioning if needed."""
+    # If not provisioned, decorator returns ElicitationRequired error
+    # If provisioned, continues normally
+    client = await get_client(ctx)
+    return await client.notes.list_notes()
+```
+
+**Error response structure**:
+```json
+{
+  "jsonrpc": "2.0",
+  "id": 1,
+  "error": {
+    "code": -32000,
+    "message": "Nextcloud access provisioning required",
+    "data": {
+      "elicitations": [
+        {
+          "mode": "url",
+          "elicitationId": "550e8400-e29b-41d4-a716-446655440000",
+          "url": "https://mcp.example.com/oauth/provision?id=550e8400...",
+          "message": "Grant the MCP server access to your Nextcloud account to continue."
+        }
+      ]
+    }
+  }
+}
+```
+
+**Client behavior**:
+1. Receives error with elicitation
+2. Shows consent UI: "App wants to access Nextcloud. Open authorization page?"
+3. On user acceptance, opens URL in browser
+4. Optionally tracks progress via `elicitation/track`
+5. Auto-retries original tool call when complete
+
+### Mode 2: Explicit Elicitation Request (Fallback)
+
+For clients that don't support error-triggered elicitation, provide explicit tool:
+
+```python
+@mcp.tool(name="request_nextcloud_access")
+async def request_access(ctx: Context) -> ElicitationResponse:
+    """Explicitly request provisioning via elicitation."""
+    # Send elicitation/create request
+    return await create_elicitation(
+        mode="url",
+        url=generate_oauth_url(),
+        message="Grant access to Nextcloud",
+        elicitation_id=generate_id()
+    )
+```
+
+**Note**: This is a fallback for compatibility. Primary flow uses error-triggered elicitation.
+
+## Implementation
+
+### 1. New Decorator: `@require_provisioning`
+
+Replace explicit provisioning checks with a decorator that returns `ElicitationRequired`:
+
+```python
+# nextcloud_mcp_server/auth/provisioning_decorator.py
+
+def require_provisioning(func):
+    """
+    Decorator that ensures user has provisioned Nextcloud access.
+
+    If not provisioned, returns ElicitationRequired error with OAuth URL.
+    Otherwise, proceeds with normal tool execution.
+    """
+    @functools.wraps(func)
+    async def wrapper(ctx: Context, *args, **kwargs):
+        # Extract user ID from token
+        user_id = get_user_id_from_context(ctx)
+
+        # Check if provisioned
+        storage = RefreshTokenStorage.from_env()
+        await storage.initialize()
+
+        if not await storage.has_refresh_token(user_id):
+            # Not provisioned - return ElicitationRequired error
+            elicitation_id = str(uuid.uuid4())
+            oauth_url = await generate_oauth_url_for_provisioning(
+                user_id=user_id,
+                elicitation_id=elicitation_id,
+                ctx=ctx
+            )
+
+            # Store elicitation for tracking
+            await storage.store_elicitation(
+                elicitation_id=elicitation_id,
+                user_id=user_id,
+                status="pending",
+                created_at=datetime.now(timezone.utc)
+            )
+
+            raise McpError(
+                code=ErrorCode.ELICITATION_REQUIRED,  # -32000
+                message="Nextcloud access provisioning required",
+                data={
+                    "elicitations": [
+                        {
+                            "mode": "url",
+                            "elicitationId": elicitation_id,
+                            "url": oauth_url,
+                            "message": (
+                                "Grant the MCP server access to your Nextcloud "
+                                "account to continue. This is a one-time setup."
+                            )
+                        }
+                    ]
+                }
+            )
+
+        # Already provisioned - proceed normally
+        return await func(ctx, *args, **kwargs)
+
+    return wrapper
+```
+
+### 2. Elicitation Tracking Endpoint
+
+Implement `elicitation/track` to provide progress updates:
+
+```python
+# nextcloud_mcp_server/server/elicitation.py
+
+@mcp.request_handler("elicitation/track")
+async def track_elicitation(
+    elicitation_id: str,
+    _meta: dict = None
+) -> dict:
+    """
+    Track progress of an elicitation request.
+
+    Returns when elicitation is complete or times out.
+    """
+    progress_token = _meta.get("progressToken") if _meta else None
+
+    storage = RefreshTokenStorage.from_env()
+    await storage.initialize()
+
+    # Poll for completion (with timeout)
+    timeout = 300  # 5 minutes
+    start_time = datetime.now(timezone.utc)
+
+    while (datetime.now(timezone.utc) - start_time).seconds < timeout:
+        elicitation = await storage.get_elicitation(elicitation_id)
+
+        if not elicitation:
+            raise McpError(
+                code=-32602,  # Invalid params
+                message=f"Unknown elicitation ID: {elicitation_id}"
+            )
+
+        # Send progress notification if token provided
+        if progress_token and elicitation["status"] == "pending":
+            await send_progress_notification(
+                progress_token=progress_token,
+                progress=50,
+                message="Waiting for OAuth authorization..."
+            )
+
+        # Check if complete
+        if elicitation["status"] == "complete":
+            return {"status": "complete"}
+
+        # Check if failed
+        if elicitation["status"] == "failed":
+            return {
+                "status": "failed",
+                "error": elicitation.get("error_message")
+            }
+
+        # Wait before polling again
+        await asyncio.sleep(2)
+
+    # Timeout
+    raise McpError(
+        code=-32000,
+        message="Elicitation timed out - user did not complete authorization"
+    )
+```
+
+### 3. OAuth Callback Updates
+
+Update the OAuth callback to mark elicitations as complete:
+
+```python
+# nextcloud_mcp_server/auth/oauth_routes.py
+
+async def oauth_callback(request: Request) -> Response:
+    """Handle OAuth callback and mark elicitation complete."""
+    code = request.query_params.get("code")
+    state = request.query_params.get("state")
+
+    # Validate and exchange code for tokens
+    tokens = await exchange_authorization_code(code)
+
+    # Store refresh token
+    await storage.store_refresh_token(
+        user_id=user_id,
+        refresh_token=tokens["refresh_token"]
+    )
+
+    # Mark elicitation as complete
+    elicitation_id = request.query_params.get("elicitation_id")
+    if elicitation_id:
+        await storage.update_elicitation(
+            elicitation_id=elicitation_id,
+            status="complete",
+            completed_at=datetime.now(timezone.utc)
+        )
+
+    return Response(
+        content="<h1>Authorization Complete!</h1>"
+        "<p>You can close this window and return to the application.</p>",
+        media_type="text/html"
+    )
+```
+
+### 4. Update All Nextcloud Tools
+
+Add `@require_provisioning` decorator to all Nextcloud tools:
+
+```python
+# nextcloud_mcp_server/server/notes.py
+
+@mcp.tool()
+@require_scopes("notes:read")
+@require_provisioning  # NEW: Auto-triggers provisioning
+async def nc_notes_list_notes(
+    ctx: Context,
+    category: Optional[str] = None
+) -> NotesListResponse:
+    """List all notes - automatically handles provisioning."""
+    client = await get_client(ctx)
+    # Tool logic proceeds only if provisioned
+    notes = await client.notes.list_notes(category=category)
+    return NotesListResponse(results=notes)
+```
+
+### 5. Capability Declaration
+
+Declare URL elicitation support during initialization:
+
+```python
+# nextcloud_mcp_server/app.py
+
+capabilities = {
+    "elicitation": {
+        "url": {}  # Declare URL mode support
+        # Note: We don't support "form" mode (in-band data collection)
+    },
+    # ... other capabilities
+}
+```
+
+### 6. Environment Variables
+
+**Primary control**:
+```bash
+# ENABLE_OFFLINE_ACCESS: Controls whether server requests refresh tokens and enables provisioning tools
+# Default: false (pass-through mode)
+# Set to true to enable offline access mode with Flow 2 provisioning
+ENABLE_OFFLINE_ACCESS=true
+```
+
+**Future variables** (when URL elicitation is implemented):
+```bash
+# ELICITATION_CALLBACK_URL: Base URL for OAuth callbacks with elicitation tracking
+# Default: NEXTCLOUD_MCP_SERVER_URL + /oauth/callback
+ELICITATION_CALLBACK_URL=http://localhost:8000/oauth/callback
+
+# ELICITATION_TIMEOUT_SECONDS: How long to wait for user to complete OAuth
+# Default: 300 (5 minutes)
+ELICITATION_TIMEOUT_SECONDS=300
+```
+
+**Removed variables**:
+```bash
+# ENABLE_PROGRESSIVE_CONSENT - Removed. Progressive consent is a mechanism, not a feature toggle.
+#                               Use ENABLE_OFFLINE_ACCESS to control whether provisioning tools are available.
+# MCP_SERVER_CLIENT_ID - merged into OIDC_CLIENT_ID
+```
+
+## User Experience Comparison
+
+### Before (ADR-004 Manual Tools)
+
+```
+User: "List my notes"
+Assistant: *calls nc_notes_list_notes*
+Server: Error - not provisioned
+Assistant: "You need to provision access first. Let me do that."
+Assistant: *calls provision_nextcloud_access*
+Server: {authorization_url: "https://..."}
+Assistant: "Please visit this URL: https://..."
+User: *copies URL, opens browser, completes OAuth*
+User: "OK, I'm done"
+Assistant: *calls nc_notes_list_notes again*
+Server: Success! [notes...]
+```
+
+**Issues**: 4 interactions, manual URL handling, no automation
+
+### After (ADR-006 Elicitation)
+
+```
+User: "List my notes"
+Assistant: *calls nc_notes_list_notes*
+Server: ElicitationRequired error
+Client: Shows dialog: "Grant access to Nextcloud? [Yes] [No]"
+User: *clicks Yes*
+Client: Opens OAuth URL in browser automatically
+User: *completes OAuth*
+Server: Sends progress notification "Complete!"
+Client: Auto-retries nc_notes_list_notes
+Server: Success! [notes...]
+Assistant: "Here are your notes: ..."
+```
+
+**Benefits**: 1 interaction, automatic URL opening, seamless retry
+
+## Migration Path
+
+### Phase 1: Add Elicitation Support (v0.26.0)
+
+- Implement `@require_provisioning` decorator
+- Add `elicitation/track` endpoint
+- Keep existing tools (`provision_nextcloud_access`) for compatibility
+- Update OAuth callback to track elicitations
+- Add capability declaration
+
+**Breaking changes**: None (additive)
+
+### Phase 2: Update Documentation (v0.27.0)
+
+- Document elicitation-based flow as primary
+- Mark manual tools as deprecated
+- Update examples and guides
+
+**Breaking changes**: None (documentation only)
+
+### Phase 3: Remove Manual Tools (v0.28.0)
+
+- Remove `provision_nextcloud_access` tool
+- Remove `check_provisioning_status` tool (status in error message)
+- Remove `revoke_nextcloud_access` (or keep for explicit revocation?)
+
+**Breaking changes**: Yes (removed tools)
+
+### Phase 4: Optimize (v0.29.0+)
+
+- Add elicitation result caching
+- Implement retry strategies
+- Add metrics and monitoring
+
+## Testing
+
+### Test Cases
+
+1. **First-Time User Flow**
+   ```python
+   @pytest.mark.oauth
+   async def test_elicitation_first_time_user(nc_mcp_oauth_client):
+       """Test that first tool call triggers elicitation."""
+       # User has no provisioning
+       with pytest.raises(McpError) as exc:
+           await nc_mcp_oauth_client.call_tool("nc_notes_list_notes")
+
+       # Should get ElicitationRequired error
+       assert exc.value.code == -32000
+       assert "elicitations" in exc.value.data
+       assert exc.value.data["elicitations"][0]["mode"] == "url"
+
+       # Verify URL is valid OAuth URL
+       url = exc.value.data["elicitations"][0]["url"]
+       assert "oauth" in url
+       assert "elicitationId" in url
+   ```
+
+2. **Progress Tracking**
+   ```python
+   @pytest.mark.oauth
+   async def test_elicitation_progress_tracking(nc_mcp_oauth_client):
+       """Test progress tracking during OAuth flow."""
+       # Trigger elicitation
+       elicitation_id = trigger_elicitation()
+
+       # Start tracking
+       track_task = asyncio.create_task(
+           nc_mcp_oauth_client.track_elicitation(
+               elicitation_id=elicitation_id,
+               progress_token="test-token"
+           )
+       )
+
+       # Simulate OAuth completion
+       await asyncio.sleep(1)
+       await complete_oauth_flow(elicitation_id)
+
+       # Track should complete
+       result = await track_task
+       assert result["status"] == "complete"
+   ```
+
+3. **Auto-Retry After Provisioning**
+   ```python
+   @pytest.mark.oauth
+   async def test_auto_retry_after_provisioning(nc_mcp_oauth_client):
+       """Test that client auto-retries after elicitation."""
+       # Mock client that auto-retries on ElicitationRequired
+       client = AutoRetryMcpClient(nc_mcp_oauth_client)
+
+       # First call triggers elicitation, client handles it, retries
+       result = await client.call_tool_with_elicitation("nc_notes_list_notes")
+
+       # Should succeed after provisioning
+       assert result.success
+       assert "notes" in result.data
+   ```
+
+4. **Timeout Handling**
+   ```python
+   @pytest.mark.oauth
+   async def test_elicitation_timeout(nc_mcp_oauth_client):
+       """Test timeout if user doesn't complete OAuth."""
+       elicitation_id = trigger_elicitation()
+
+       # Track with short timeout
+       with pytest.raises(McpError, match="timed out"):
+           await nc_mcp_oauth_client.track_elicitation(
+               elicitation_id=elicitation_id,
+               timeout=5  # 5 seconds
+           )
+   ```
+
+## Security Considerations
+
+### Out-of-Band OAuth Flow
+
+**Benefit**: OAuth credentials never pass through MCP client
+- User enters credentials directly on IdP page
+- MCP server receives only authorization code
+- Client never sees passwords or refresh tokens
+
+**Threat mitigation**:
+- **Credential theft**: Client can't intercept credentials (out-of-band)
+- **Token exposure**: Client never receives Nextcloud refresh tokens
+- **CSRF**: State parameter validates OAuth callback
+- **URL tampering**: Elicitation ID ties OAuth flow to user session
+
+### Elicitation ID as Security Token
+
+The `elicitationId` serves as a capability token:
+- Cryptographically random (UUID v4)
+- Single-use (invalidated after completion)
+- Time-limited (expires after timeout)
+- User-scoped (tied to user session)
+
+**Validation**:
+```python
+async def validate_elicitation_id(elicitation_id: str, user_id: str) -> bool:
+    """Validate that elicitation belongs to user and is still valid."""
+    elicitation = await storage.get_elicitation(elicitation_id)
+
+    if not elicitation:
+        return False
+
+    # Check ownership
+    if elicitation["user_id"] != user_id:
+        logger.warning(f"Elicitation ID mismatch: {elicitation_id}")
+        return False
+
+    # Check expiry
+    if elicitation["expires_at"] < datetime.now(timezone.utc):
+        return False
+
+    # Check not already used
+    if elicitation["status"] != "pending":
+        return False
+
+    return True
+```
+
+### Progress Tracking Security
+
+**Risk**: Progress token reuse across users
+
+**Mitigation**:
+- Progress tokens tied to elicitation ID
+- Elicitation ID tied to user session
+- Server validates ownership before sending updates
+
+## Consequences
+
+### Positive
+
+1. **Better UX**: Automatic URL opening, no manual copy/paste
+2. **Seamless Flow**: Auto-retry after provisioning
+3. **Progress Feedback**: User knows when OAuth is complete
+4. **Spec Compliance**: Implements SEP-1036 correctly
+5. **Secure by Design**: Out-of-band OAuth prevents credential exposure
+6. **Simpler API**: No explicit provisioning tools needed
+
+### Negative
+
+1. **Client Dependency**: Requires client support for URL elicitation
+2. **Complexity**: More moving parts (elicitation tracking, callbacks)
+3. **Polling**: Progress tracking uses polling (not ideal)
+4. **Breaking Change**: Removes manual provisioning tools (in v0.28.0)
+
+### Neutral
+
+1. **Storage Requirements**: Need to store elicitation state
+2. **Timeout Management**: Must handle long-running OAuth flows
+3. **Fallback Support**: Still need compatibility for older clients
+
+## Alternatives Considered
+
+### 1. Keep Manual Tools Only (Rejected)
+
+**Pros**: Simple, no client changes needed
+**Cons**: Poor UX, doesn't leverage SEP-1036
+
+**Rejection reason**: SEP-1036 provides better UX and security
+
+### 2. Form Mode Elicitation (Rejected)
+
+**Pros**: No browser redirect needed
+**Cons**: Would expose OAuth credentials to client (security violation)
+
+**Rejection reason**: Form mode only for non-sensitive data per SEP-1036
+
+### 3. Hybrid: Both Tools and Elicitation (Considered)
+
+**Pros**: Maximum compatibility, gradual migration
+**Cons**: API duplication, maintenance burden, confusing for users
+
+**Decision**: Support during migration (v0.26-0.27), remove in v0.28
+
+### 4. WebSocket for Progress (Rejected)
+
+**Pros**: Real-time updates instead of polling
+**Cons**: MCP spec uses polling pattern, adds complexity
+
+**Rejection reason**: Follow spec pattern (polling via elicitation/track)
+
+## Interim Implementation: Inline Form Elicitation (Pre-SEP-1036)
+
+**Note**: SEP-1036 (URL mode elicitation) is not yet available in the stable MCP Python SDK. As a temporary workaround, we've implemented a simplified version using the current **inline form elicitation** API.
+
+### What Changed
+
+Instead of waiting for URL mode elicitation, we implemented a `check_logged_in` tool that:
+
+1. Checks if the user has completed Flow 2 (resource provisioning)
+2. If logged in, returns `"yes"`
+3. If not logged in, uses **inline form elicitation** to prompt the user
+
+### Implementation Details
+
+**New Tool**: `check_logged_in`
+
+```python
+# nextcloud_mcp_server/server/oauth_tools.py
+
+class LoginConfirmation(BaseModel):
+    """Schema for login confirmation elicitation."""
+    acknowledged: bool = Field(
+        default=False,
+        description="Check this box after completing login at the provided URL",
+    )
+
+@mcp.tool(name="check_logged_in")
+@require_scopes("openid")
+async def tool_check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str:
+    """Check if user is logged in and elicit login if needed."""
+    # Check if already logged in
+    status = await get_provisioning_status(ctx, user_id)
+    if status.is_provisioned:
+        return "yes"
+
+    # Generate OAuth URL for Flow 2
+    auth_url = generate_oauth_url_for_flow2(...)
+
+    # Use inline form elicitation (current MCP API)
+    result = await ctx.elicit(
+        message=f"Please log in to Nextcloud at the following URL:\n\n{auth_url}\n\nAfter completing the login, check the box below and click OK.",
+        schema=LoginConfirmation,
+    )
+
+    if result.action == "accept":
+        # Verify login succeeded
+        status = await get_provisioning_status(ctx, user_id)
+        return "yes" if status.is_provisioned else "Login not detected"
+    elif result.action == "decline":
+        return "Login declined by user."
+    else:
+        return "Login cancelled by user."
+```
+
+**OAuth Routes** (added to `app.py`):
+
+```python
+# Flow 2 routes for resource provisioning
+routes.append(
+    Route("/oauth/authorize-nextcloud", oauth_authorize_nextcloud, methods=["GET"])
+)
+routes.append(
+    Route("/oauth/callback-nextcloud", oauth_callback_nextcloud, methods=["GET"])
+)
+```
+
+### User Experience
+
+```
+User: *calls check_logged_in tool*
+
+MCP Client: Displays form elicitation
+┌─────────────────────────────────────────────────────────┐
+│ Please log in to Nextcloud at the following URL:      │
+│                                                         │
+│ http://localhost:8000/oauth/authorize-nextcloud?...    │
+│                                                         │
+│ After completing the login, check the box below and    │
+│ click OK.                                               │
+│                                                         │
+│ ☐ Check this box after completing login                │
+│                                                         │
+│ [Accept] [Decline] [Cancel]                            │
+└─────────────────────────────────────────────────────────┘
+
+User: *copies URL, opens in browser, completes OAuth*
+User: *checks box and clicks Accept*
+
+MCP Server: Verifies login and returns "yes"
+```
+
+### Limitations of Interim Approach
+
+1. **Manual URL Handling**: User must manually copy and paste the URL (not clickable)
+2. **No Automatic Browser Opening**: Client doesn't automatically open the URL
+3. **No Progress Tracking**: Can't track OAuth completion status in real-time
+4. **URL in Message Text**: Login URL embedded in plain text message (not as structured field)
+5. **Client-Side Confirmation**: Relies on user clicking "OK" after OAuth (honor system)
+
+### Why Not Use URL Mode Now?
+
+The current stable MCP Python SDK (`main` branch) only supports **inline form elicitation**:
+
+```python
+# Current API (no 'mode' parameter)
+class ElicitRequestParams(RequestParams):
+    message: str
+    requestedSchema: ElicitRequestedSchema
+    # No 'mode', 'url', or 'elicitationId' fields
+```
+
+URL mode elicitation (`mode: "url"`) is only available in the SEP-1036 branch, which has not been merged to `main` yet.
+
+### Migration to URL Mode (When SEP-1036 Lands)
+
+Once SEP-1036 is merged and available in the stable SDK, we will migrate to URL mode elicitation:
+
+**Before (Current Workaround)**:
+```python
+result = await ctx.elicit(
+    message=f"Please log in at: {auth_url}\n\nClick OK after login.",
+    schema=LoginConfirmation,
+)
+```
+
+**After (URL Mode)**:
+```python
+result = await ctx.session.elicit_url(
+    message="Please log in to Nextcloud to authorize this MCP server.",
+    url=auth_url,
+    elicitation_id=elicitation_id,
+)
+```
+
+**Benefits of migration**:
+- Automatic URL opening (with user consent)
+- Clickable URLs in client UI
+- Progress tracking via `elicitation/track`
+- Better security (URL not in message text)
+- Auto-retry support
+
+### Testing
+
+Integration tests validate the current inline form elicitation:
+
+```python
+# tests/server/oauth/test_login_elicitation.py
+
+async def test_check_logged_in_already_authenticated(nc_mcp_oauth_client):
+    """Test immediate 'yes' for authenticated users."""
+    result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={})
+    assert "yes" in result.content[0].text.lower()
+
+async def test_check_logged_in_url_format(nc_mcp_oauth_client):
+    """Test that login URL (when needed) contains correct OAuth parameters."""
+    result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={})
+    response_text = result.content[0].text
+
+    # If URL present, validate OAuth parameters
+    if "http" in response_text:
+        assert "response_type=code" in response_text
+        assert "client_id=" in response_text
+        assert "redirect_uri=" in response_text
+        assert "openid" in response_text
+```
+
+### Future Work
+
+- **Monitor SEP-1036**: Watch for merge to MCP Python SDK `main` branch
+- **Implement URL Mode**: Once available, migrate `check_logged_in` to use `ctx.session.elicit_url()`
+- **Add Progress Tracking**: Implement `elicitation/track` endpoint for OAuth completion status
+- **Implement Error-Triggered Elicitation**: Use `@require_provisioning` decorator to return `ElicitationRequired` errors
+- **Remove Manual Workaround**: Deprecate inline form approach once URL mode is stable
+
+## References
+
+- [SEP-1036: URL Mode Elicitation](https://github.com/modelcontextprotocol/specification/pull/887)
+- [MCP Elicitation Specification](https://modelcontextprotocol.io/specification/draft/client/elicitation)
+- [ADR-004: Federated Authentication Architecture](./ADR-004-mcp-application-oauth.md)
+- [ADR-005: Token Audience Validation](./ADR-005-token-audience-validation.md)
+- [RFC 8252: OAuth 2.0 for Native Apps](https://datatracker.ietf.org/doc/html/rfc8252)
+
+## Implementation Checklist
+
+### Interim Implementation (Inline Form Elicitation)
+
+- [x] Create `check_logged_in` tool with inline form elicitation
+- [x] Register Flow 2 OAuth routes (`/oauth/authorize-nextcloud`, `/oauth/callback-nextcloud`)
+- [x] Write integration tests for login elicitation flow
+- [x] Update ADR-006 with interim implementation documentation
+- [x] Add `LoginConfirmation` schema for elicitation
+- [ ] Run tests to validate implementation
+
+### Future Work (URL Mode Elicitation - Post SEP-1036)
+
+- [ ] Implement `@require_provisioning` decorator with ElicitationRequired error
+- [ ] Add `elicitation/track` request handler
+- [ ] Update OAuth callback to mark elicitations complete
+- [ ] Add elicitation storage (ID, user, status, timestamps)
+- [ ] Update all Nextcloud tools with `@require_provisioning`
+- [ ] Add URL elicitation capability declaration
+- [ ] Write tests for progress tracking
+- [ ] Update documentation with URL mode examples
+- [ ] Add migration guide for manual tools → elicitation
+- [ ] Migrate `check_logged_in` from inline form to URL mode
+- [ ] Keep manual tools with deprecation warnings (v0.26-0.27)
+- [ ] Remove manual tools (v0.28.0)
+- [ ] Update CHANGELOG.md with migration timeline

keycloak/realm-export.json

@@ -751,6 +751,40 @@
         "display.on.consent.screen": "true",
         "consent.screen.text": "Create, update, and delete tasks"
       }
+    },
+    {
+      "name": "default-audience",
+      "protocol": "openid-connect",
+      "attributes": {
+        "include.in.token.scope": "false",
+        "display.on.consent.screen": "false",
+        "gui.order": "",
+        "consent.screen.text": ""
+      },
+      "protocolMappers": [
+        {
+          "name": "mcp-server-audience",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-audience-mapper",
+          "consentRequired": false,
+          "config": {
+            "included.client.audience": "nextcloud-mcp-server",
+            "access.token.claim": "true",
+            "id.token.claim": "false"
+          }
+        },
+        {
+          "name": "mcp-url-audience",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-audience-mapper",
+          "consentRequired": false,
+          "config": {
+            "included.custom.audience": "http://localhost:8002",
+            "access.token.claim": "true",
+            "id.token.claim": "false"
+          }
+        }
+      ]
     }
   ],
   "components": {
@@ -791,7 +825,8 @@
     "profile",
     "email",
     "roles",
-    "web-origins"
+    "web-origins",
+    "default-audience"
   ],
   "defaultOptionalClientScopes": [
     "offline_access",

nextcloud_mcp_server/app.py

@@ -27,9 +27,7 @@ from nextcloud_mcp_server.auth import (
     has_required_scopes,
     is_jwt_token,
 )
-from nextcloud_mcp_server.auth.progressive_token_verifier import (
-    ProgressiveConsentTokenVerifier,
-)
+from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
 from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.config import (
     LOGGING_CONFIG,
@@ -215,12 +213,13 @@ class OAuthAppContext:
     """Application context for OAuth mode."""
 
     nextcloud_host: str
-    token_verifier: (
-        object  # Can be NextcloudTokenVerifier or ProgressiveConsentTokenVerifier
-    )
+    token_verifier: object  # UnifiedTokenVerifier (ADR-005 compliant)
     refresh_token_storage: Optional["RefreshTokenStorage"] = None
     oauth_client: Optional[object] = None  # NextcloudOAuthClient or KeycloakOAuthClient
     oauth_provider: str = "nextcloud"  # "nextcloud" or "keycloak"
+    server_client_id: Optional[str] = (
+        None  # MCP server's OAuth client ID (static or DCR)
+    )
 
 
 def is_oauth_mode() -> bool:
@@ -296,8 +295,7 @@ async def load_oauth_client_credentials(
         logger.info("Dynamic client registration available")
         mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
         redirect_uris = [
-            f"{mcp_server_url}/oauth/callback",  # MCP OAuth flow
-            f"{mcp_server_url}/oauth/login-callback",  # Browser OAuth flow for /user/page
+            f"{mcp_server_url}/oauth/callback",  # Unified callback (flow determined by query param)
         ]
 
         # MCP server DCR: Register with ALL supported scopes
@@ -555,46 +553,80 @@ async def setup_oauth_config():
     else:
         client_issuer = issuer
 
-    # Progressive Consent mode (always enabled) - dual OAuth flows with audience separation
-    logger.info("✓ Progressive Consent mode enabled - dual OAuth flows active")
+    # ADR-005: Unified Token Verifier with proper audience validation
+    # Get MCP server URL for audience validation
+    mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
+    nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host)
 
-    # Get encryption key for token broker
-    encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY")
-    if not encryption_key:
+    # Warn if resource URIs are not configured (required for ADR-005 compliance)
+    if not os.getenv("NEXTCLOUD_MCP_SERVER_URL"):
         logger.warning(
-            "TOKEN_ENCRYPTION_KEY not set - token broker will not be available"
+            f"NEXTCLOUD_MCP_SERVER_URL not set, defaulting to: {mcp_server_url}. "
+            "This should be set explicitly for proper audience validation."
+        )
+    if not os.getenv("NEXTCLOUD_RESOURCE_URI"):
+        logger.warning(
+            f"NEXTCLOUD_RESOURCE_URI not set, defaulting to: {nextcloud_resource_uri}. "
+            "This should be set explicitly for proper audience validation."
         )
 
-    # Create token broker service
-    from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
+    # Create settings for UnifiedTokenVerifier
+    from nextcloud_mcp_server.config import get_settings
 
-    token_broker = None
-    if encryption_key and refresh_token_storage:
-        token_broker = TokenBrokerService(
-            storage=refresh_token_storage,
-            oidc_discovery_url=discovery_url,
-            nextcloud_host=nextcloud_host,
-            encryption_key=encryption_key,
+    settings = get_settings()
+    # Override with discovered values if not set in environment
+    if not settings.oidc_client_id:
+        settings.oidc_client_id = client_id
+    if not settings.oidc_client_secret:
+        settings.oidc_client_secret = client_secret
+    if not settings.jwks_uri:
+        settings.jwks_uri = jwks_uri
+    if not settings.introspection_uri:
+        settings.introspection_uri = introspection_uri
+    if not settings.userinfo_uri:
+        settings.userinfo_uri = userinfo_uri
+    if not settings.oidc_issuer:
+        # Use client_issuer which handles public URL override
+        settings.oidc_issuer = client_issuer
+    if not settings.nextcloud_mcp_server_url:
+        settings.nextcloud_mcp_server_url = mcp_server_url
+    if not settings.nextcloud_resource_uri:
+        settings.nextcloud_resource_uri = nextcloud_resource_uri
+
+    # Create Unified Token Verifier (ADR-005 compliant)
+    token_verifier = UnifiedTokenVerifier(settings)
+
+    # Log the mode
+    enable_token_exchange = (
+        os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true"
+    )
+    if enable_token_exchange:
+        logger.info(
+            "✓ Token Exchange mode enabled (ADR-005) - exchanging MCP tokens for Nextcloud tokens via RFC 8693"
         )
-        logger.info("✓ Token Broker service initialized for audience-specific tokens")
+        logger.info(f"  MCP audience: {client_id} or {mcp_server_url}")
+        logger.info(f"  Nextcloud audience: {nextcloud_resource_uri}")
+    else:
+        logger.info(
+            "✓ Multi-audience mode enabled (ADR-005) - tokens must contain both MCP and Nextcloud audiences"
+        )
+        logger.info(f"  Required MCP audience: {client_id} or {mcp_server_url}")
+        logger.info(f"  Required Nextcloud audience: {nextcloud_resource_uri}")
 
-    # Create Progressive Consent token verifier
-    token_verifier = ProgressiveConsentTokenVerifier(
-        token_storage=refresh_token_storage,
-        token_broker=token_broker,
-        oidc_discovery_url=discovery_url,
-        nextcloud_host=nextcloud_host,
-        encryption_key=encryption_key,
-        mcp_client_id=client_id,
-        introspection_uri=introspection_uri,
-        client_secret=client_secret,
-    )
-
-    logger.info(
-        "✓ Progressive Consent verifier configured - enforcing audience separation"
-    )
     if introspection_uri:
         logger.info("✓ Opaque token introspection enabled (RFC 7662)")
+    if jwks_uri:
+        logger.info("✓ JWT signature verification enabled (JWKS)")
+
+    # Progressive Consent mode (for offline access / background jobs)
+    encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY")
+    if enable_offline_access and encryption_key and refresh_token_storage:
+        logger.info("✓ Progressive Consent mode enabled - offline access available")
+
+        # Note: Token Broker service would be initialized here for background job support
+        # Currently not used in ADR-005 implementation as it's specific to offline access patterns
+        # that are separate from the real-time token exchange flow
+        logger.debug("Token broker available for future offline access features")
 
     # Create OAuth client for server-initiated flows (e.g., token exchange, background workers)
     oauth_client = None
@@ -603,6 +635,8 @@ async def setup_oauth_config():
         from nextcloud_mcp_server.auth.keycloak_oauth import KeycloakOAuthClient
 
         mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
+        # Note: This redirect_uri is for OAuth client initialization, not used for actual redirects
+        # since this client is used for backend token operations (exchange, refresh)
         redirect_uri = f"{mcp_server_url}/oauth/callback"
 
         # Extract base URL and realm from discovery URL
@@ -708,6 +742,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
                     refresh_token_storage=refresh_token_storage,
                     oauth_client=oauth_client,
                     oauth_provider=oauth_provider,
+                    server_client_id=client_id,
                 )
             finally:
                 logger.info("Shutting down MCP server")
@@ -763,16 +798,27 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
                 f"Unknown app: {app_name}. Available apps: {list(available_apps.keys())}"
             )
 
-    # Register OAuth provisioning tools (only when offline access/Progressive Consent is used)
+    # Register OAuth provisioning tools (only when offline access is enabled)
     # With token exchange enabled (external IdP), provisioning is not needed for MCP operations
     enable_token_exchange = (
         os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true"
     )
-    if oauth_enabled and not enable_token_exchange:
-        logger.info("Registering OAuth provisioning tools for Progressive Consent")
+    enable_offline_access_for_tools = os.getenv(
+        "ENABLE_OFFLINE_ACCESS", "false"
+    ).lower() in (
+        "true",
+        "1",
+        "yes",
+    )
+    if oauth_enabled and enable_offline_access_for_tools and not enable_token_exchange:
+        logger.info("Registering OAuth provisioning tools for offline access")
         register_oauth_tools(mcp)
     elif oauth_enabled and enable_token_exchange:
         logger.info("Skipping provisioning tools registration (token exchange enabled)")
+    elif oauth_enabled and not enable_offline_access_for_tools:
+        logger.info(
+            "Skipping provisioning tools registration (offline access not enabled)"
+        )
 
     # Override list_tools to filter based on user's token scopes (OAuth mode only)
     if oauth_enabled:
@@ -818,7 +864,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
             return allowed_tools
 
         # Replace the tool manager's list_tools method
-        mcp._tool_manager.list_tools = list_tools_filtered
+        mcp._tool_manager.list_tools = list_tools_filtered  # type: ignore[method-assign]
         logger.info(
             "Dynamic tool filtering enabled for OAuth mode (JWT and Bearer tokens)"
         )
@@ -837,13 +883,16 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
                 mcp_server_url = os.getenv(
                     "NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000"
                 )
+                nextcloud_resource_uri = os.getenv(
+                    "NEXTCLOUD_RESOURCE_URI", nextcloud_host
+                )
                 discovery_url = os.getenv(
                     "OIDC_DISCOVERY_URL",
                     f"{nextcloud_host}/.well-known/openid-configuration",
                 )
                 scopes = os.getenv("NEXTCLOUD_OIDC_SCOPES", "")
 
-                app.state.oauth_context = {
+                oauth_context_dict = {
                     "storage": refresh_token_storage,
                     "oauth_client": oauth_client,
                     "token_verifier": token_verifier,  # For querying IdP userinfo endpoint
@@ -854,9 +903,23 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
                         "client_secret": client_secret,  # From setup_oauth_config (DCR or static)
                         "scopes": scopes,
                         "nextcloud_host": nextcloud_host,
+                        "nextcloud_resource_uri": nextcloud_resource_uri,
                         "oauth_provider": oauth_provider,
                     },
                 }
+                app.state.oauth_context = oauth_context_dict
+
+                # Also set oauth_context on browser_app for session authentication
+                # browser_app is in the same function scope (defined later in create_app)
+                # We need to find it in the mounted routes
+                for route in app.routes:
+                    if isinstance(route, Mount) and route.path == "/user":
+                        route.app.state.oauth_context = oauth_context_dict
+                        logger.info(
+                            "OAuth context shared with browser_app for session auth"
+                        )
+                        break
+
                 logger.info(
                     f"OAuth context initialized for login routes (client_id={client_id[:16]}...)"
                 )
@@ -997,6 +1060,38 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
         routes.append(Route("/oauth/authorize", oauth_authorize, methods=["GET"]))
         logger.info("OAuth login routes enabled: /oauth/authorize (Flow 1)")
 
+        # Add unified OAuth callback endpoint supporting both flows
+        from nextcloud_mcp_server.auth.oauth_routes import (
+            oauth_authorize_nextcloud,
+            oauth_callback,
+            oauth_callback_nextcloud,
+        )
+
+        routes.append(Route("/oauth/callback", oauth_callback, methods=["GET"]))
+        logger.info(
+            "OAuth unified callback enabled: /oauth/callback?flow={browser|provisioning}"
+        )
+
+        # Add OAuth resource provisioning routes (ADR-004 Progressive Consent Flow 2)
+        routes.append(
+            Route(
+                "/oauth/authorize-nextcloud",
+                oauth_authorize_nextcloud,
+                methods=["GET"],
+            )
+        )
+        # Keep old callback endpoint as backwards-compatible alias
+        routes.append(
+            Route(
+                "/oauth/callback-nextcloud",
+                oauth_callback_nextcloud,
+                methods=["GET"],
+            )
+        )
+        logger.info(
+            "OAuth resource provisioning routes enabled: /oauth/authorize-nextcloud, /oauth/callback-nextcloud (Flow 2, legacy)"
+        )
+
     # Add browser OAuth login routes (OAuth mode only)
     if oauth_enabled:
         from nextcloud_mcp_server.auth.browser_oauth_routes import (
@@ -1008,6 +1103,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
         routes.append(
             Route("/oauth/login", oauth_login, methods=["GET"], name="oauth_login")
         )
+        # Keep old callback endpoint as backwards-compatible alias
         routes.append(
             Route(
                 "/oauth/login-callback",
@@ -1020,13 +1116,14 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
             Route("/oauth/logout", oauth_logout, methods=["GET"], name="oauth_logout")
         )
         logger.info(
-            "Browser OAuth routes enabled: /oauth/login, /oauth/login-callback, /oauth/logout"
+            "Browser OAuth routes enabled: /oauth/login, /oauth/login-callback (legacy), /oauth/logout"
         )
 
     # Add user info routes (available in both BasicAuth and OAuth modes)
     # These require session authentication, so we wrap them in a separate app
     from nextcloud_mcp_server.auth.session_backend import SessionAuthBackend
     from nextcloud_mcp_server.auth.userinfo_routes import (
+        revoke_session,
         user_info_html,
         user_info_json,
     )
@@ -1036,6 +1133,9 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
     browser_routes = [
         Route("/", user_info_json, methods=["GET"]),  # /user/ → user_info_json
         Route("/page", user_info_html, methods=["GET"]),  # /user/page → user_info_html
+        Route(
+            "/revoke", revoke_session, methods=["POST"], name="revoke_session_endpoint"
+        ),  # /user/revoke → revoke_session
     ]
 
     browser_app = Starlette(routes=browser_routes)

nextcloud_mcp_server/auth/__init__.py

@@ -14,11 +14,11 @@ from .scope_authorization import (
     is_jwt_token,
     require_scopes,
 )
-from .token_verifier import NextcloudTokenVerifier
+from .unified_verifier import UnifiedTokenVerifier
 
 __all__ = [
     "BearerAuth",
-    "NextcloudTokenVerifier",
+    "UnifiedTokenVerifier",
     "register_client",
     "ensure_oauth_client",
     "get_client_from_context",

nextcloud_mcp_server/auth/browser_oauth_routes.py

@@ -4,9 +4,11 @@ Separate from MCP OAuth flow - these routes establish browser sessions
 for accessing admin UI endpoints like /user/page.
 """
 
+import hashlib
 import logging
 import os
 import secrets
+from base64 import urlsafe_b64encode
 from urllib.parse import urlencode
 
 import httpx
@@ -53,39 +55,36 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
 
     # Build OAuth authorization URL
     mcp_server_url = oauth_config["mcp_server_url"]
-    callback_uri = f"{mcp_server_url}/oauth/login-callback"
+    callback_uri = f"{mcp_server_url}/oauth/callback"
 
     # Request only basic OIDC scopes for browser session
     # Note: Nextcloud app scopes (notes:read, etc.) are for MCP client access tokens,
     # not for the MCP server's own browser authentication
     scopes = "openid profile email offline_access"
 
-    code_challenge = ""
-    code_verifier = ""
+    # Generate PKCE values for ALL modes (both external and integrated IdP require PKCE)
+    code_verifier = secrets.token_urlsafe(32)
+    digest = hashlib.sha256(code_verifier.encode()).digest()
+    code_challenge = urlsafe_b64encode(digest).decode().rstrip("=")
+
+    # Store code_verifier in session for retrieval during callback (using state as key)
+    await storage.store_oauth_session(
+        session_id=state,  # Use state as session ID
+        client_id="browser-ui",
+        client_redirect_uri="/user/page",
+        state=state,
+        code_challenge=code_challenge,
+        code_challenge_method="S256",
+        mcp_authorization_code=code_verifier,  # Store code_verifier here temporarily
+        flow_type="browser",
+        ttl_seconds=600,  # 10 minutes
+    )
 
     if oauth_client:
         # External IdP mode (Keycloak)
-        # Keycloak requires PKCE, so generate code_verifier and code_challenge
         if not oauth_client.authorization_endpoint:
             await oauth_client.discover()
 
-        # Generate PKCE values
-        code_verifier, code_challenge = oauth_client.generate_pkce_challenge()
-
-        # Store code_verifier temporarily (using state as key)
-        # We'll retrieve it in the callback using the state parameter
-        await storage.store_oauth_session(
-            session_id=state,  # Use state as session ID
-            client_id="browser-ui",
-            client_redirect_uri="/user/page",
-            state=state,
-            code_challenge=code_challenge,
-            code_challenge_method="S256",
-            mcp_authorization_code=code_verifier,  # Store code_verifier here temporarily
-            flow_type="browser",
-            ttl_seconds=600,  # 10 minutes
-        )
-
         idp_params = {
             "client_id": oauth_client.client_id,
             "redirect_uri": callback_uri,
@@ -138,6 +137,8 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
             "response_type": "code",
             "scope": scopes,
             "state": state,
+            "code_challenge": code_challenge,
+            "code_challenge_method": "S256",
             "prompt": "consent",  # Ensure refresh token
         }
 
@@ -213,20 +214,18 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
     oauth_client = oauth_ctx["oauth_client"]
     oauth_config = oauth_ctx["config"]
 
-    # Retrieve code_verifier from session storage (if using PKCE)
+    # Retrieve code_verifier from session storage (PKCE required for all modes)
     code_verifier = ""
-    if oauth_client:
-        # For Keycloak (external IdP), we stored the code_verifier in the session
-        oauth_session = await storage.get_oauth_session(state)
-        if oauth_session:
-            # code_verifier was stored in mcp_authorization_code field
-            code_verifier = oauth_session.get("mcp_authorization_code", "")
-            # Clean up the temporary session
-            # Note: We don't have delete_oauth_session method, but it will expire after TTL
+    oauth_session = await storage.get_oauth_session(state)
+    if oauth_session:
+        # code_verifier was stored in mcp_authorization_code field
+        code_verifier = oauth_session.get("mcp_authorization_code", "")
+        # Clean up the temporary session
+        # Note: We don't have delete_oauth_session method, but it will expire after TTL
 
     # Exchange authorization code for tokens
     mcp_server_url = oauth_config["mcp_server_url"]
-    callback_uri = f"{mcp_server_url}/oauth/login-callback"
+    callback_uri = f"{mcp_server_url}/oauth/callback"
 
     try:
         if oauth_client:
@@ -263,16 +262,22 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
                 discovery = response.json()
                 token_endpoint = discovery["token_endpoint"]
 
+            token_params = {
+                "grant_type": "authorization_code",
+                "code": code,
+                "redirect_uri": callback_uri,
+                "client_id": oauth_config["client_id"],
+                "client_secret": oauth_config["client_secret"],
+            }
+
+            # Add code_verifier for PKCE (required by Nextcloud OIDC)
+            if code_verifier:
+                token_params["code_verifier"] = code_verifier
+
             async with httpx.AsyncClient() as http_client:
                 response = await http_client.post(
                     token_endpoint,
-                    data={
-                        "grant_type": "authorization_code",
-                        "code": code,
-                        "redirect_uri": callback_uri,
-                        "client_id": oauth_config["client_id"],
-                        "client_secret": oauth_config["client_secret"],
-                    },
+                    data=token_params,
                 )
                 response.raise_for_status()
                 token_data = response.json()
@@ -336,13 +341,18 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
     # Store refresh token (for background jobs ONLY)
     if refresh_token:
         logger.info(f"Storing refresh token for user_id: {user_id}")
+        logger.info(f"  State parameter (provisioning_client_id): {state[:16]}...")
         await storage.store_refresh_token(
             user_id=user_id,
             refresh_token=refresh_token,
             expires_at=None,
             flow_type="browser",  # Browser-based login flow
+            provisioning_client_id=state,  # Store state for unified session lookup
         )
         logger.info(f"✓ Refresh token stored successfully for user_id: {user_id}")
+        logger.info(
+            f"  Token can now be found via provisioning_client_id={state[:16]}..."
+        )
     else:
         logger.warning("No refresh token in token response - cannot store session")
 

nextcloud_mcp_server/auth/context_helper.py

@@ -1,6 +1,11 @@
-"""Helper functions for extracting OAuth context from MCP requests."""
+"""Helper functions for extracting OAuth context from MCP requests.
 
+ADR-005 compliant implementation with token exchange caching.
+"""
+
+import hashlib
 import logging
+import time
 
 from mcp.server.auth.provider import AccessToken
 from mcp.server.fastmcp import Context
@@ -11,35 +16,36 @@ from .token_exchange import exchange_token_for_audience
 
 logger = logging.getLogger(__name__)
 
+# Token exchange cache: token_hash -> (exchanged_token, expiry_timestamp)
+_exchange_cache: dict[str, tuple[str, float]] = {}
+
 
 def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
     """
-    Extract authenticated user context from MCP request and create NextcloudClient.
+    Create NextcloudClient for multi-audience mode (no exchange needed).
 
-    This function retrieves the OAuth access token from the MCP context,
-    extracts the username from the token's resource field (where we stored it
-    during token verification), and creates a NextcloudClient with bearer auth.
+    ADR-005 Mode 1: Use multi-audience tokens directly.
+    The UnifiedTokenVerifier validated MCP audience per RFC 7519.
+    Nextcloud will independently validate its own audience.
 
     Args:
         ctx: MCP request context containing session info
         base_url: Nextcloud base URL
 
     Returns:
-        NextcloudClient configured with bearer token auth
+        NextcloudClient configured with multi-audience token
 
     Raises:
         AttributeError: If context doesn't contain expected OAuth session data
         ValueError: If username cannot be extracted from token
     """
     try:
-        # In Starlette with FastMCP OAuth, the authenticated user info is stored in request.user
-        # The FastMCP auth middleware sets request.user to an AuthenticatedUser object
-        # which contains the access_token
+        # Extract validated access token from MCP context
         if hasattr(ctx.request_context.request, "user") and hasattr(
             ctx.request_context.request.user, "access_token"
         ):
             access_token: AccessToken = ctx.request_context.request.user.access_token
-            logger.debug("Retrieved access token from request.user for OAuth request")
+            logger.debug("Retrieved multi-audience token from request.user")
         else:
             logger.error(
                 "OAuth authentication failed: No access token found in request"
@@ -47,16 +53,20 @@ def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
             raise AttributeError("No access token found in OAuth request context")
 
         # Extract username from resource field (RFC 8707)
-        # We stored the username here during token verification
+        # UnifiedTokenVerifier stored the username here during validation
         username = access_token.resource
 
         if not username:
             logger.error("No username found in access token resource field")
             raise ValueError("Username not available in OAuth token context")
 
-        logger.debug(f"Creating OAuth NextcloudClient for user: {username}")
+        logger.debug(
+            f"Creating NextcloudClient for user {username} with multi-audience token "
+            f"(no exchange needed)"
+        )
 
-        # Create client with bearer token
+        # Token was validated to have MCP audience
+        # Nextcloud will validate its own audience independently
         return NextcloudClient.from_token(
             base_url=base_url, token=access_token.token, username=username
         )
@@ -71,12 +81,19 @@ async def get_session_client_from_context(
     ctx: Context, base_url: str
 ) -> NextcloudClient:
     """
-    Create NextcloudClient using RFC 8693 token exchange for session operations.
+    Create NextcloudClient using RFC 8693 token exchange with caching.
+
+    ADR-005 Mode 2: Exchange MCP token for Nextcloud token via RFC 8693.
 
     This implements the token exchange pattern where:
-    1. Extract Flow 1 token from context (aud: "mcp-server")
-    2. Exchange it for ephemeral Nextcloud token via RFC 8693
-    3. Create client with delegated token (NOT stored)
+    1. Extract MCP token from context (validated by UnifiedTokenVerifier)
+    2. Check cache for existing exchanged token
+    3. If not cached or expired, exchange via RFC 8693
+    4. Cache the exchanged token to minimize exchange frequency
+    5. Create client with exchanged token
+
+    CRITICAL: This is where token exchange happens, NOT in the verifier.
+    The verifier already validated the MCP audience; now we exchange for Nextcloud.
 
     Note: Nextcloud doesn't support OAuth scopes natively. Scopes are enforced
     by the MCP server via @require_scopes decorator, not by the IdP. Therefore,
@@ -88,7 +105,7 @@ async def get_session_client_from_context(
         base_url: Nextcloud base URL
 
     Returns:
-        NextcloudClient configured with ephemeral delegated token
+        NextcloudClient configured with ephemeral exchanged token
 
     Raises:
         AttributeError: If context doesn't contain expected OAuth session data
@@ -96,43 +113,60 @@ async def get_session_client_from_context(
     """
     settings = get_settings()
 
-    # Check if token exchange is enabled
-    if not settings.enable_token_exchange:
-        logger.info("Token exchange disabled, falling back to standard OAuth flow")
-        return get_client_from_context(ctx, base_url)
-
     try:
-        # Extract Flow 1 token from context
+        # Extract MCP token from context
         if hasattr(ctx.request_context.request, "user") and hasattr(
             ctx.request_context.request.user, "access_token"
         ):
             access_token: AccessToken = ctx.request_context.request.user.access_token
-            flow1_token = access_token.token
-            username = access_token.resource  # Username stored during verification
-            logger.debug(f"Retrieved Flow 1 token for user: {username}")
+            mcp_token = access_token.token
+            username = access_token.resource  # Username from UnifiedTokenVerifier
+            logger.debug(f"Retrieved MCP token for user: {username}")
         else:
-            logger.error("No Flow 1 token found in request context")
+            logger.error("No MCP token found in request context")
             raise AttributeError("No access token found in OAuth request context")
 
         if not username:
             logger.error("No username found in access token resource field")
             raise ValueError("Username not available in OAuth token context")
 
-        logger.info("Exchanging client token for Nextcloud API token (pure RFC 8693)")
+        # Check cache for existing exchanged token
+        cache_key = hashlib.sha256(mcp_token.encode()).hexdigest()
+        if cache_key in _exchange_cache:
+            cached_token, expiry = _exchange_cache[cache_key]
+            if time.time() < expiry:
+                logger.debug(
+                    f"Using cached exchanged token (expires in {expiry - time.time():.1f}s)"
+                )
+                return NextcloudClient.from_token(
+                    base_url=base_url, token=cached_token, username=username
+                )
+            else:
+                logger.debug("Cached token expired, removing from cache")
+                del _exchange_cache[cache_key]
 
-        # Perform pure RFC 8693 token exchange (no refresh tokens)
-        # Note: We don't pass scopes since Nextcloud doesn't enforce them.
-        # The MCP server's @require_scopes decorator handles authorization.
+        # Perform RFC 8693 token exchange
+        logger.info(f"Exchanging MCP token for Nextcloud API token (user: {username})")
+
+        # Exchange for Nextcloud resource URI audience
         exchanged_token, expires_in = await exchange_token_for_audience(
-            subject_token=flow1_token,
-            requested_audience="nextcloud",
+            subject_token=mcp_token,
+            requested_audience=settings.nextcloud_resource_uri or "nextcloud",
             requested_scopes=None,  # Nextcloud doesn't support scopes
         )
 
-        logger.info(f"Pure token exchange successful. Token expires in {expires_in}s")
+        logger.info(f"Token exchange successful. Token expires in {expires_in}s")
+
+        # Cache the exchanged token
+        # Use the minimum of exchange TTL and configured cache TTL
+        cache_ttl = min(expires_in, settings.token_exchange_cache_ttl)
+        _exchange_cache[cache_key] = (exchanged_token, time.time() + cache_ttl)
+        logger.debug(f"Cached exchanged token for {cache_ttl}s")
+
+        # Clean up expired cache entries
+        _cleanup_exchange_cache()
 
         # Create client with exchanged token
-        # This token is ephemeral (per-request) and NOT stored
         return NextcloudClient.from_token(
             base_url=base_url, token=exchanged_token, username=username
         )
@@ -143,3 +177,21 @@ async def get_session_client_from_context(
     except Exception as e:
         logger.error(f"Token exchange failed: {e}")
         raise RuntimeError(f"Token exchange required but failed: {e}") from e
+
+
+def _cleanup_exchange_cache():
+    """Remove expired entries from the token exchange cache."""
+    global _exchange_cache
+    now = time.time()
+    expired_keys = [k for k, (_, expiry) in _exchange_cache.items() if expiry <= now]
+    for key in expired_keys:
+        del _exchange_cache[key]
+    if expired_keys:
+        logger.debug(f"Cleaned up {len(expired_keys)} expired cache entries")
+
+
+def clear_exchange_cache():
+    """Clear the entire token exchange cache. Useful for testing."""
+    global _exchange_cache
+    _exchange_cache.clear()
+    logger.debug("Token exchange cache cleared")

nextcloud_mcp_server/auth/keycloak_oauth.py

@@ -90,6 +90,8 @@ class KeycloakOAuthClient:
             )
 
         # Parse server URL to construct redirect URI
+        # Note: This is for OAuth client initialization, not used for actual redirects
+        # since this client is used for backend token operations (exchange, refresh)
         parsed_url = urlparse(server_url)
         redirect_uri = f"{parsed_url.scheme}://{parsed_url.netloc}/oauth/callback"
 

nextcloud_mcp_server/auth/oauth_routes.py

@@ -1,7 +1,7 @@
 """
-OAuth 2.0 Login Routes for ADR-004 Progressive Consent Architecture
+OAuth 2.0 Login Routes for ADR-004 (Offline Access Architecture)
 
-Implements dual OAuth flows with explicit provisioning:
+Implements dual OAuth flows with optional offline access provisioning:
 
 Flow 1: Client Authentication - MCP client authenticates directly to IdP
 - Client requests: Nextcloud MCP resource scopes (notes:*, calendar:*, etc.)
@@ -19,8 +19,11 @@ Flow 2: Resource Provisioning - MCP server gets delegated Nextcloud access
 
 """
 
+import hashlib
 import logging
 import os
+import secrets
+from base64 import urlsafe_b64encode
 from urllib.parse import urlencode
 
 import httpx
@@ -118,7 +121,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
             status_code=400,
         )
 
-    # Validate client_id (required for Progressive Consent Flow 1)
+    # Validate client_id (required for Flow 1)
     if not client_id:
         return JSONResponse(
             {
@@ -168,7 +171,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
     # The MCP server does NOT see the IdP authorization code!
 
     logger.info(
-        f"Starting Progressive Consent Flow 1 - no server session needed, "
+        f"Starting Flow 1 - no server session needed, "
         f"client will handle IdP response directly at {redirect_uri}"
     )
 
@@ -188,7 +191,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
     # Use client's own client_id (client must be pre-registered at IdP)
     idp_client_id = client_id
 
-    logger.info("Flow 1 (Progressive Consent): Direct client auth to IdP")
+    logger.info("Flow 1: Direct client auth to IdP")
     logger.info(f"  Client ID: {client_id}")
     logger.info(f"  Client will receive IdP code directly at: {callback_uri}")
     logger.info(f"  Scopes: {scopes} (resource access for MCP tools)")
@@ -252,6 +255,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
             "scope": scopes,
             "state": idp_state,
             "prompt": "consent",  # Ensure refresh token
+            "resource": f"{oauth_config['mcp_server_url']}/mcp",  # MCP server audience
         }
 
         auth_url = f"{authorization_endpoint}?{urlencode(idp_params)}"
@@ -313,12 +317,31 @@ async def oauth_authorize_nextcloud(
         )
 
     mcp_server_url = oauth_config["mcp_server_url"]
-    callback_uri = f"{mcp_server_url}/oauth/callback-nextcloud"
+    callback_uri = f"{mcp_server_url}/oauth/callback"
 
     # Flow 2: Server only needs identity + offline access (no resource scopes)
     # Resource scopes are requested by client in Flow 1
     scopes = "openid profile email offline_access"
 
+    # Generate PKCE values (required by Nextcloud OIDC)
+    code_verifier = secrets.token_urlsafe(32)
+    digest = hashlib.sha256(code_verifier.encode()).digest()
+    code_challenge = urlsafe_b64encode(digest).decode().rstrip("=")
+
+    # Store code_verifier in session for retrieval during callback
+    storage = oauth_ctx["storage"]
+    await storage.store_oauth_session(
+        session_id=state,
+        client_id=mcp_server_client_id,
+        client_redirect_uri=callback_uri,
+        state=state,
+        code_challenge=code_challenge,
+        code_challenge_method="S256",
+        mcp_authorization_code=code_verifier,  # Store code_verifier here temporarily
+        flow_type="flow2",
+        ttl_seconds=600,  # 10 minutes
+    )
+
     # Get authorization endpoint
     discovery_url = oauth_config.get("discovery_url")
     if not discovery_url:
@@ -357,8 +380,11 @@ async def oauth_authorize_nextcloud(
         "response_type": "code",
         "scope": scopes,
         "state": state,
+        "code_challenge": code_challenge,
+        "code_challenge_method": "S256",
         "prompt": "consent",  # Force consent to show resource access
         "access_type": "offline",  # Request refresh token
+        "resource": oauth_config["nextcloud_resource_uri"],  # Nextcloud audience
     }
 
     auth_url = f"{authorization_endpoint}?{urlencode(idp_params)}"
@@ -414,6 +440,16 @@ async def oauth_callback_nextcloud(request: Request):
     storage: RefreshTokenStorage = oauth_ctx["storage"]
     oauth_config = oauth_ctx["config"]
 
+    # Retrieve code_verifier from session storage (PKCE required by Nextcloud OIDC)
+    code_verifier = ""
+    oauth_session = await storage.get_oauth_session(state)
+    if oauth_session:
+        # code_verifier was stored in mcp_authorization_code field
+        code_verifier = oauth_session.get("mcp_authorization_code", "")
+        logger.info(
+            f"Retrieved code_verifier for Flow 2 callback (state={state[:16]}...)"
+        )
+
     # Exchange code for tokens
     mcp_server_client_id = os.getenv(
         "MCP_SERVER_CLIENT_ID", oauth_config.get("client_id")
@@ -422,7 +458,7 @@ async def oauth_callback_nextcloud(request: Request):
         "MCP_SERVER_CLIENT_SECRET", oauth_config.get("client_secret")
     )
     mcp_server_url = oauth_config["mcp_server_url"]
-    callback_uri = f"{mcp_server_url}/oauth/callback-nextcloud"
+    callback_uri = f"{mcp_server_url}/oauth/callback"
 
     discovery_url = oauth_config.get("discovery_url")
     async with httpx.AsyncClient() as http_client:
@@ -431,17 +467,24 @@ async def oauth_callback_nextcloud(request: Request):
         discovery = response.json()
         token_endpoint = discovery["token_endpoint"]
 
+    # Build token exchange params
+    token_params = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": callback_uri,
+        "client_id": mcp_server_client_id,
+        "client_secret": mcp_server_client_secret,
+    }
+
+    # Add code_verifier for PKCE (required by Nextcloud OIDC)
+    if code_verifier:
+        token_params["code_verifier"] = code_verifier
+
     # Exchange code for tokens
     async with httpx.AsyncClient() as http_client:
         response = await http_client.post(
             token_endpoint,
-            data={
-                "grant_type": "authorization_code",
-                "code": code,
-                "redirect_uri": callback_uri,
-                "client_id": mcp_server_client_id,
-                "client_secret": mcp_server_client_secret,
-            },
+            data=token_params,
         )
         response.raise_for_status()
         token_data = response.json()
@@ -450,14 +493,22 @@ async def oauth_callback_nextcloud(request: Request):
     id_token = token_data.get("id_token")
 
     # Decode ID token to get user info
+    logger.info("=" * 60)
+    logger.info("oauth_callback_nextcloud: Extracting user_id from ID token")
+    logger.info("=" * 60)
     try:
         userinfo = jwt.decode(id_token, options={"verify_signature": False})
         user_id = userinfo.get("sub")
         username = userinfo.get("preferred_username") or userinfo.get("email")
+        logger.info("  ✓ ID token decode SUCCESSFUL")
+        logger.info(f"  Extracted user_id: {user_id}")
+        logger.info(f"  Username: {username}")
+        logger.info(f"  ID token payload keys: {list(userinfo.keys())}")
         logger.info(f"Flow 2: User {username} provisioned resource access")
     except Exception as e:
-        logger.warning(f"Failed to decode ID token: {e}")
+        logger.error(f"  ✗ ID token decode FAILED: {type(e).__name__}: {e}")
         user_id = "unknown"
+        logger.error(f"  Using fallback user_id: {user_id}")
 
     # Store master refresh token for Flow 2
     if refresh_token:
@@ -466,6 +517,13 @@ async def oauth_callback_nextcloud(request: Request):
             token_data.get("scope", "").split() if token_data.get("scope") else None
         )
 
+        logger.info("Storing refresh token:")
+        logger.info(f"  user_id: {user_id}")
+        logger.info("  flow_type: flow2")
+        logger.info("  token_audience: nextcloud")
+        logger.info(f"  provisioning_client_id: {state[:16]}...")
+        logger.info(f"  scopes: {granted_scopes}")
+
         await storage.store_refresh_token(
             user_id=user_id,
             refresh_token=refresh_token,
@@ -475,7 +533,8 @@ async def oauth_callback_nextcloud(request: Request):
             scopes=granted_scopes,
             expires_at=None,  # Refresh tokens typically don't expire
         )
-        logger.info(f"Stored Flow 2 master refresh token for user {user_id}")
+        logger.info(f"✓ Stored Flow 2 master refresh token for user {user_id}")
+        logger.info("=" * 60)
 
     # Return success HTML page
     success_html = """
@@ -500,3 +559,82 @@ async def oauth_callback_nextcloud(request: Request):
     from starlette.responses import HTMLResponse
 
     return HTMLResponse(content=success_html, status_code=200)
+
+
+async def oauth_callback(request: Request):
+    """
+    Unified OAuth callback endpoint supporting multiple flows.
+
+    This endpoint consolidates all OAuth callback handling into a single URL.
+    The flow type is determined by looking up the OAuth session using the
+    state parameter.
+
+    This simplifies IdP configuration by requiring only one callback URL
+    to be registered: /oauth/callback
+
+    Query parameters:
+        code: Authorization code from IdP
+        state: CSRF protection state (also used to lookup flow type)
+        error: Error code (if authorization failed)
+
+    Returns:
+        Response from the appropriate flow handler
+    """
+    # Get state parameter to lookup OAuth session
+    state = request.query_params.get("state")
+    if not state:
+        logger.warning("Unified callback called without state parameter")
+        return JSONResponse(
+            {
+                "error": "invalid_request",
+                "error_description": "state parameter is required",
+            },
+            status_code=400,
+        )
+
+    # Lookup OAuth session to determine flow type
+    oauth_ctx = request.app.state.oauth_context
+    if not oauth_ctx:
+        logger.error("OAuth context not available")
+        return JSONResponse(
+            {
+                "error": "server_error",
+                "error_description": "OAuth not configured on server",
+            },
+            status_code=500,
+        )
+
+    storage = oauth_ctx["storage"]
+    oauth_session = await storage.get_oauth_session(state)
+
+    # Determine flow type from session, default to "browser" for backwards compatibility
+    flow_type = (
+        oauth_session.get("flow_type", "browser") if oauth_session else "browser"
+    )
+
+    logger.info(f"Unified callback: flow_type={flow_type} (from session lookup)")
+
+    if flow_type == "flow2":
+        # Flow 2: Resource Provisioning - MCP server gets delegated Nextcloud access
+        logger.info("Routing to Flow 2 (resource provisioning)")
+        return await oauth_callback_nextcloud(request)
+
+    elif flow_type == "browser":
+        # Browser UI Login - establish browser session for /user/page access
+        logger.info("Routing to browser login flow")
+        from nextcloud_mcp_server.auth.browser_oauth_routes import (
+            oauth_login_callback,
+        )
+
+        return await oauth_login_callback(request)
+
+    else:
+        # Unknown flow type
+        logger.warning(f"Unknown flow_type in OAuth session: {flow_type}")
+        return JSONResponse(
+            {
+                "error": "invalid_request",
+                "error_description": f"Unknown flow type: {flow_type}",
+            },
+            status_code=400,
+        )

nextcloud_mcp_server/auth/progressive_token_verifier.py

@@ -1,361 +0,0 @@
-"""
-Token Verifier for ADR-004 Progressive Consent Architecture.
-
-This module implements token verification with strict audience separation:
-- Flow 1 tokens have aud: <mcp-client-id> for MCP authentication
-- Flow 2 tokens have aud: "nextcloud" for resource access
-- Token Broker manages the exchange between audiences
-"""
-
-import logging
-import os
-from datetime import datetime, timezone
-from typing import Optional
-
-import httpx
-import jwt
-from mcp.server.auth.provider import AccessToken
-
-from nextcloud_mcp_server.auth.refresh_token_storage import RefreshTokenStorage
-from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
-
-logger = logging.getLogger(__name__)
-
-
-class ProgressiveConsentTokenVerifier:
-    """
-    Token verifier for Progressive Consent dual OAuth flows.
-
-    This verifier:
-    1. Validates Flow 1 tokens (aud: <mcp-client-id>) for MCP authentication
-    2. Checks if user has provisioned Nextcloud access (Flow 2)
-    3. Uses Token Broker to obtain aud: "nextcloud" tokens when needed
-    """
-
-    def __init__(
-        self,
-        token_storage: RefreshTokenStorage,
-        token_broker: Optional[TokenBrokerService] = None,
-        oidc_discovery_url: Optional[str] = None,
-        nextcloud_host: Optional[str] = None,
-        encryption_key: Optional[str] = None,
-        mcp_client_id: Optional[str] = None,
-        introspection_uri: Optional[str] = None,
-        client_secret: Optional[str] = None,
-    ):
-        """
-        Initialize the Progressive Consent token verifier.
-
-        Args:
-            token_storage: Storage for refresh tokens
-            token_broker: Token broker service (created if not provided)
-            oidc_discovery_url: OIDC provider discovery URL
-            nextcloud_host: Nextcloud server URL
-            encryption_key: Fernet key for token encryption
-            mcp_client_id: MCP server OAuth client ID for audience validation
-            introspection_uri: OAuth introspection endpoint URL (for opaque tokens)
-            client_secret: OAuth client secret (required for introspection)
-        """
-        self.storage = token_storage
-        self.oidc_discovery_url = oidc_discovery_url or os.getenv(
-            "OIDC_DISCOVERY_URL",
-            f"{os.getenv('NEXTCLOUD_HOST')}/.well-known/openid-configuration",
-        )
-        self.nextcloud_host = nextcloud_host or os.getenv("NEXTCLOUD_HOST")
-        self.encryption_key = encryption_key or os.getenv("TOKEN_ENCRYPTION_KEY")
-        self.mcp_client_id = mcp_client_id or os.getenv("OIDC_CLIENT_ID")
-        self.introspection_uri = introspection_uri
-        self.client_secret = client_secret or os.getenv("OIDC_CLIENT_SECRET")
-
-        # HTTP client for introspection requests
-        self._http_client: Optional[httpx.AsyncClient] = None
-        if self.introspection_uri and self.mcp_client_id and self.client_secret:
-            self._http_client = httpx.AsyncClient(timeout=10.0)
-            logger.info(f"Introspection support enabled: {introspection_uri}")
-        elif self.introspection_uri:
-            logger.warning(
-                "Introspection URI provided but missing client credentials - introspection disabled"
-            )
-
-        # Create token broker if not provided
-        if token_broker:
-            self.token_broker = token_broker
-        elif self.encryption_key:
-            self.token_broker = TokenBrokerService(
-                storage=token_storage,
-                oidc_discovery_url=self.oidc_discovery_url,
-                nextcloud_host=self.nextcloud_host,
-                encryption_key=self.encryption_key,
-            )
-        else:
-            self.token_broker = None
-            logger.warning("Token broker not available - encryption key missing")
-
-    async def verify_token(self, token: str) -> Optional[AccessToken]:
-        """
-        Verify a Flow 1 token (aud: <mcp-client-id>).
-
-        This validates that:
-        1. Token has correct audience for MCP server (matches client ID)
-        2. Token is not expired
-        3. Token has valid signature (if verification enabled)
-
-        Supports both JWT and opaque tokens:
-        - JWT tokens: Decoded directly from payload
-        - Opaque tokens: Validated via introspection endpoint (RFC 7662)
-
-        Args:
-            token: Access token from Flow 1 (JWT or opaque)
-
-        Returns:
-            AccessToken if valid, None otherwise
-        """
-        logger.info("🔐 verify_token called - attempting to validate token")
-        logger.info(f"Token (first 50 chars): {token[:50]}...")
-        logger.info(f"Expected MCP client ID: {self.mcp_client_id}")
-
-        # Check if token is JWT format (has 3 parts separated by dots)
-        is_jwt = "." in token and token.count(".") == 2
-        logger.info(f"Token format: {'JWT' if is_jwt else 'opaque'}")
-
-        if is_jwt:
-            # Try JWT verification
-            return await self._verify_jwt_token(token)
-        else:
-            # Fall back to introspection for opaque tokens
-            return await self._verify_opaque_token(token)
-
-    async def _verify_jwt_token(self, token: str) -> Optional[AccessToken]:
-        """Verify JWT token by decoding payload."""
-        try:
-            # Decode without signature verification (IdP handles that)
-            # In production, would verify signature with IdP public key
-            payload = jwt.decode(token, options={"verify_signature": False})
-            logger.info(f"Token payload decoded: {payload}")
-
-            # CRITICAL: Verify audience is for MCP server (Flow 1)
-            audiences = payload.get("aud", [])
-            if isinstance(audiences, str):
-                audiences = [audiences]
-
-            # Audience validation:
-            # - Accept tokens with no audience (will validate via introspection if needed)
-            # - Accept tokens with MCP client ID in audience (Keycloak multi-audience)
-            # - Accept tokens with resource URL in audience (Nextcloud JWT redirect URI)
-            # - Reject tokens with "nextcloud" audience only (wrong flow)
-            if audiences:
-                # Check if MCP client ID is in the audience (Keycloak multi-audience)
-                if self.mcp_client_id in audiences:
-                    logger.debug(
-                        f"Token has audience {audiences} - MCP client ID present"
-                    )
-                # Check if this is a Nextcloud-only token (wrong flow)
-                elif audiences == ["nextcloud"]:
-                    logger.warning(
-                        f"Token rejected: Nextcloud-only audience {audiences}"
-                    )
-                    logger.error(
-                        "Received Nextcloud token in MCP context - "
-                        "client may be using wrong token"
-                    )
-                    return None
-                # Otherwise accept (likely resource URL audience from Nextcloud JWT)
-                else:
-                    logger.info(
-                        f"Token has audience {audiences} (resource URL or non-standard) - accepting"
-                    )
-            else:
-                logger.info(
-                    "Token has no audience claim - accepting for MCP server validation"
-                )
-
-            # Check expiry
-            exp = payload.get("exp", 0)
-            if exp < datetime.now(timezone.utc).timestamp():
-                logger.warning(
-                    f"❌ Token expired: exp={exp}, now={datetime.now(timezone.utc).timestamp()}"
-                )
-                return None
-
-            # Extract user info
-            user_id = payload.get("sub", "unknown")
-            client_id = payload.get("client_id", "unknown")
-            scopes = payload.get("scope", "").split()
-            exp = payload.get("exp", None)
-
-            logger.info(
-                f"✅ Token validation successful! user={user_id}, scopes={scopes}"
-            )
-
-            # Create AccessToken for MCP framework
-            return AccessToken(
-                token=token,
-                client_id=client_id,
-                scopes=scopes,
-                expires_at=exp,
-                resource=user_id,  # Store user_id in resource field (RFC 8707)
-            )
-
-        except jwt.InvalidTokenError as e:
-            logger.warning(f"❌ Invalid token (JWT decode failed): {e}")
-            return None
-        except Exception as e:
-            logger.error(f"❌ Token verification failed with exception: {e}")
-            return None
-
-    async def _verify_opaque_token(self, token: str) -> Optional[AccessToken]:
-        """
-        Verify opaque token via introspection endpoint (RFC 7662).
-
-        Args:
-            token: Opaque access token
-
-        Returns:
-            AccessToken if active and valid, None otherwise
-        """
-        if not self._http_client or not self.introspection_uri:
-            logger.error(
-                "❌ Cannot verify opaque token - introspection not configured. "
-                "Set introspection_uri and client credentials."
-            )
-            return None
-
-        try:
-            logger.info(f"Introspecting token at {self.introspection_uri}")
-
-            # Call introspection endpoint (requires client authentication)
-            response = await self._http_client.post(
-                self.introspection_uri,
-                data={"token": token},
-                auth=(self.mcp_client_id, self.client_secret),
-            )
-
-            if response.status_code != 200:
-                logger.warning(
-                    f"❌ Introspection failed: HTTP {response.status_code} - {response.text[:200]}"
-                )
-                return None
-
-            introspection_data = response.json()
-            logger.info(f"Introspection response: {introspection_data}")
-
-            # Check if token is active
-            if not introspection_data.get("active", False):
-                logger.warning("❌ Token introspection returned active=false")
-                return None
-
-            # Extract user info
-            user_id = introspection_data.get("sub") or introspection_data.get(
-                "username"
-            )
-            if not user_id:
-                logger.error("❌ No username found in introspection response")
-                return None
-
-            # Extract scopes (space-separated string)
-            scope_string = introspection_data.get("scope", "")
-            scopes = scope_string.split() if scope_string else []
-
-            # Extract client ID and expiration
-            client_id = introspection_data.get("client_id", "unknown")
-            exp = introspection_data.get("exp")
-
-            logger.info(f"✅ Opaque token validated! user={user_id}, scopes={scopes}")
-
-            return AccessToken(
-                token=token,
-                client_id=client_id,
-                scopes=scopes,
-                expires_at=int(exp) if exp else None,
-                resource=user_id,
-            )
-
-        except httpx.TimeoutException:
-            logger.error("❌ Timeout while introspecting token")
-            return None
-        except httpx.RequestError as e:
-            logger.error(f"❌ Network error during introspection: {e}")
-            return None
-        except Exception as e:
-            logger.error(f"❌ Introspection failed with exception: {e}")
-            return None
-
-    async def check_provisioning(self, user_id: str) -> bool:
-        """
-        Check if user has provisioned Nextcloud access (Flow 2).
-
-        Args:
-            user_id: User identifier from Flow 1 token
-
-        Returns:
-            True if user has completed Flow 2, False otherwise
-        """
-        if not self.storage:
-            return False
-
-        refresh_data = await self.storage.get_refresh_token(user_id)
-        return refresh_data is not None
-
-    async def get_nextcloud_token(self, user_id: str) -> Optional[str]:
-        """
-        Get a Nextcloud access token (aud: "nextcloud") for the user.
-
-        This uses the Token Broker to:
-        1. Check for cached Nextcloud token
-        2. If expired, refresh using stored master refresh token
-        3. Return token with aud: "nextcloud" for API access
-
-        Args:
-            user_id: User identifier from Flow 1 token
-
-        Returns:
-            Nextcloud access token if provisioned, None otherwise
-        """
-        if not self.token_broker:
-            logger.error("Token broker not available")
-            return None
-
-        # Check if user has provisioned access
-        if not await self.check_provisioning(user_id):
-            logger.info(f"User {user_id} has not provisioned Nextcloud access")
-            return None
-
-        # Get or refresh Nextcloud token
-        try:
-            nextcloud_token = await self.token_broker.get_nextcloud_token(user_id)
-            if nextcloud_token:
-                logger.debug(f"Obtained Nextcloud token for user {user_id}")
-            return nextcloud_token
-        except Exception as e:
-            logger.error(f"Failed to get Nextcloud token: {e}")
-            return None
-
-    async def validate_scopes(
-        self, token: AccessToken, required_scopes: list[str]
-    ) -> bool:
-        """
-        Validate that token has required scopes.
-
-        Args:
-            token: The access token
-            required_scopes: List of required scopes
-
-        Returns:
-            True if all required scopes present, False otherwise
-        """
-        token_scopes = set(token.scopes) if token.scopes else set()
-        required = set(required_scopes)
-
-        missing = required - token_scopes
-        if missing:
-            logger.debug(f"Token missing required scopes: {missing}")
-            return False
-
-        return True
-
-    async def close(self):
-        """Clean up resources."""
-        if self.token_broker:
-            await self.token_broker.close()
-        if self._http_client:
-            await self._http_client.aclose()

nextcloud_mcp_server/auth/provisioning_decorator.py

@@ -1,8 +1,8 @@
 """
-Provisioning decorator for ADR-004 Progressive Consent Architecture.
+Provisioning decorator for ADR-004 (Offline Access Architecture).
 
 This decorator ensures users have completed Flow 2 (Resource Provisioning)
-before accessing Nextcloud resources.
+before accessing Nextcloud resources when offline access is enabled.
 """
 
 import functools
@@ -73,7 +73,7 @@ def require_provisioning(func: Callable) -> Callable:
             logger.debug("Token exchange mode detected - skipping provisioning check")
             return await func(*args, **kwargs)
 
-        # Progressive Consent mode (offline access) - check if user has completed Flow 2 provisioning
+        # Offline access mode - check if user has completed Flow 2 provisioning
         # Get user_id from authorization token
         user_id = None
         if hasattr(ctx, "authorization") and ctx.authorization:

nextcloud_mcp_server/auth/refresh_token_storage.py

@@ -25,7 +25,7 @@ import logging
 import os
 import time
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 import aiosqlite
 from cryptography.fernet import Fernet
@@ -283,7 +283,7 @@ class RefreshTokenStorage:
         )
 
     async def store_user_profile(
-        self, user_id: str, profile_data: dict[str, any]
+        self, user_id: str, profile_data: dict[str, Any]
     ) -> None:
         """
         Store user profile data (cached from IdP userinfo endpoint).
@@ -314,7 +314,7 @@ class RefreshTokenStorage:
 
         logger.debug(f"Cached user profile for {user_id}")
 
-    async def get_user_profile(self, user_id: str) -> Optional[dict[str, any]]:
+    async def get_user_profile(self, user_id: str) -> Optional[dict[str, Any]]:
         """
         Retrieve cached user profile data.
 
@@ -430,6 +430,84 @@ class RefreshTokenStorage:
             logger.error(f"Failed to decrypt refresh token for user {user_id}: {e}")
             return None
 
+    async def get_refresh_token_by_provisioning_client_id(
+        self, provisioning_client_id: str
+    ) -> Optional[dict]:
+        """
+        Retrieve and decrypt refresh token by provisioning_client_id (state parameter).
+
+        This is used to check if an OAuth Flow 2 login completed successfully
+        by looking up the refresh token using the state parameter that was generated
+        during the authorization request.
+
+        Args:
+            provisioning_client_id: OAuth state parameter from the authorization request
+
+        Returns:
+            Dictionary with token data or None if not found
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with aiosqlite.connect(self.db_path) as db:
+            async with db.execute(
+                """
+                SELECT user_id, encrypted_token, expires_at, flow_type, token_audience,
+                       provisioned_at, provisioning_client_id, scopes
+                FROM refresh_tokens WHERE provisioning_client_id = ?
+                """,
+                (provisioning_client_id,),
+            ) as cursor:
+                row = await cursor.fetchone()
+
+        if not row:
+            logger.debug(
+                f"No refresh token found for provisioning_client_id {provisioning_client_id[:16]}..."
+            )
+            return None
+
+        (
+            user_id,
+            encrypted_token,
+            expires_at,
+            flow_type,
+            token_audience,
+            provisioned_at,
+            prov_client_id,
+            scopes_json,
+        ) = row
+
+        # Check expiration
+        if expires_at is not None and expires_at < time.time():
+            logger.warning(
+                f"Refresh token for provisioning_client_id {provisioning_client_id[:16]}... has expired"
+            )
+            return None
+
+        try:
+            decrypted_token = self.cipher.decrypt(encrypted_token).decode()
+            scopes = json.loads(scopes_json) if scopes_json else None
+
+            logger.debug(
+                f"Retrieved refresh token for provisioning_client_id {provisioning_client_id[:16]}... (user_id: {user_id})"
+            )
+
+            return {
+                "user_id": user_id,
+                "refresh_token": decrypted_token,
+                "expires_at": expires_at,
+                "flow_type": flow_type or "hybrid",
+                "token_audience": token_audience or "nextcloud",
+                "provisioned_at": provisioned_at,
+                "provisioning_client_id": prov_client_id,
+                "scopes": scopes,
+            }
+        except Exception as e:
+            logger.error(
+                f"Failed to decrypt refresh token for provisioning_client_id {provisioning_client_id[:16]}...: {e}"
+            )
+            return None
+
     async def delete_refresh_token(self, user_id: str) -> bool:
         """
         Delete refresh token for user.

nextcloud_mcp_server/auth/scope_authorization.py

@@ -3,7 +3,7 @@
 import logging
 import os
 from functools import wraps
-from typing import Callable
+from typing import Any, Callable
 
 from mcp.server.auth.middleware.auth_context import get_access_token
 from mcp.server.auth.provider import AccessToken
@@ -88,15 +88,18 @@ def require_scopes(*required_scopes: str):
         ScopeAuthorizationError: If required scopes are not present in the access token
     """
 
-    def decorator(func: Callable):
+    def decorator(func: Callable) -> Callable:
         # Store scope requirements as function metadata for dynamic filtering
-        func._required_scopes = list(required_scopes)  # type: ignore
+        func._required_scopes = list(required_scopes)  # type: ignore[attr-defined]
+
+        # Get function name for logging (works for any callable)
+        func_name = getattr(func, "__name__", repr(func))
 
         # Find which parameter receives the Context (FastMCP injects it by name)
         context_param_name = find_context_parameter(func)
 
         @wraps(func)
-        async def wrapper(*args, **kwargs):
+        async def wrapper(*args: Any, **kwargs: Any) -> Any:
             # Extract context from kwargs (where FastMCP injected it)
             ctx: Context | None = (
                 kwargs.get(context_param_name) if context_param_name else None
@@ -106,7 +109,7 @@ def require_scopes(*required_scopes: str):
                 # No context parameter found - likely BasicAuth mode
                 # In BasicAuth mode, all operations are allowed
                 logger.debug(
-                    f"No context parameter for {func.__name__} - allowing (BasicAuth mode)"
+                    f"No context parameter for {func_name} - allowing (BasicAuth mode)"
                 )
                 return await func(*args, **kwargs)
 
@@ -119,7 +122,7 @@ def require_scopes(*required_scopes: str):
                 # Not in OAuth mode (BasicAuth or no auth)
                 # In BasicAuth mode, all operations are allowed
                 logger.debug(
-                    f"No access token present for {func.__name__} - allowing (BasicAuth mode)"
+                    f"No access token present for {func_name} - allowing (BasicAuth mode)"
                 )
                 return await func(*args, **kwargs)
 
@@ -127,13 +130,13 @@ def require_scopes(*required_scopes: str):
             token_scopes = set(access_token.scopes or [])
             required_scopes_set = set(required_scopes)
 
-            # Check if Progressive Consent is enabled
-            enable_progressive = (
-                os.getenv("ENABLE_PROGRESSIVE_CONSENT", "false").lower() == "true"
+            # Check if offline access is enabled
+            enable_offline_access = (
+                os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true"
             )
 
-            # In Progressive Consent mode, check if Nextcloud scopes require provisioning
-            if enable_progressive:
+            # In offline access mode, check if Nextcloud scopes require provisioning
+            if enable_offline_access:
                 # Check if any required scopes are Nextcloud-specific
                 nextcloud_scopes = [
                     s
@@ -172,7 +175,7 @@ def require_scopes(*required_scopes: str):
 
                     if not has_nextcloud_scopes:
                         error_msg = (
-                            f"Access denied to {func.__name__}: "
+                            f"Access denied to {func_name}: "
                             f"Nextcloud resource access not provisioned. "
                             f"Please run the 'provision_nextcloud_access' tool first."
                         )
@@ -183,7 +186,7 @@ def require_scopes(*required_scopes: str):
             missing_scopes = required_scopes_set - token_scopes
             if missing_scopes:
                 error_msg = (
-                    f"Access denied to {func.__name__}: "
+                    f"Access denied to {func_name}: "
                     f"Missing required scopes: {', '.join(sorted(missing_scopes))}. "
                     f"Token has scopes: {', '.join(sorted(token_scopes)) if token_scopes else 'none'}"
                 )
@@ -192,7 +195,7 @@ def require_scopes(*required_scopes: str):
 
             # All required scopes present - allow execution
             logger.debug(
-                f"Scope authorization passed for {func.__name__}: {required_scopes}"
+                f"Scope authorization passed for {func_name}: {required_scopes}"
             )
             return await func(*args, **kwargs)
 

nextcloud_mcp_server/auth/token_broker.py

@@ -68,7 +68,7 @@ class TokenCache:
             logger.debug(f"Using cached token for user {user_id}")
             return token
 
-    async def set(self, user_id: str, token: str, expires_in: int = None):
+    async def set(self, user_id: str, token: str, expires_in: int | None = None):
         """Store token in cache."""
         async with self._lock:
             # Use provided expiry or default TTL

nextcloud_mcp_server/auth/token_exchange.py

@@ -114,7 +114,8 @@ class TokenExchangeService:
         if not self.oidc_discovery_url:
             # Fallback to Nextcloud OIDC if no discovery URL
             self.oidc_discovery_url = urljoin(
-                self.nextcloud_host, "/.well-known/openid-configuration"
+                self.nextcloud_host,  # type: ignore[arg-type]
+                "/.well-known/openid-configuration",
             )
 
         try:
@@ -363,6 +364,7 @@ class TokenExchangeService:
             True if provisioned, False otherwise
         """
         await self._ensure_storage()
+        assert self.storage is not None  # _ensure_storage() ensures this
         token_data = await self.storage.get_refresh_token(user_id)
         return token_data is not None
 
@@ -376,6 +378,7 @@ class TokenExchangeService:
             Refresh token if found, None otherwise
         """
         await self._ensure_storage()
+        assert self.storage is not None  # _ensure_storage() ensures this
         token_data = await self.storage.get_refresh_token(user_id)
         if token_data:
             return token_data.get("refresh_token")

nextcloud_mcp_server/auth/token_verifier.py

@@ -1,491 +0,0 @@
-"""Token verification using Nextcloud OIDC userinfo endpoint."""
-
-import logging
-import time
-from typing import Any
-
-import httpx
-import jwt
-from jwt import PyJWKClient
-from mcp.server.auth.provider import AccessToken, TokenVerifier
-
-logger = logging.getLogger(__name__)
-
-
-class NextcloudTokenVerifier(TokenVerifier):
-    """
-    Validates access tokens using JWT verification with JWKS or userinfo endpoint fallback.
-
-    This verifier supports both JWT and opaque tokens:
-    1. For JWT tokens: Verifies signature with JWKS and extracts scopes from payload
-    2. For opaque tokens: Falls back to userinfo endpoint validation
-    3. Caches successful responses to avoid repeated API calls/verifications
-
-    JWT validation provides:
-    - Faster validation (no HTTP call needed)
-    - Direct scope extraction from token payload
-    - Signature verification using JWKS
-
-    Userinfo fallback provides:
-    - Support for opaque tokens
-    - Backward compatibility
-    - Additional validation layer
-    """
-
-    def __init__(
-        self,
-        nextcloud_host: str,
-        userinfo_uri: str,
-        jwks_uri: str | None = None,
-        issuer: str | None = None,
-        introspection_uri: str | None = None,
-        client_id: str | None = None,
-        client_secret: str | None = None,
-        cache_ttl: int = 3600,
-    ):
-        """
-        Initialize the token verifier.
-
-        Args:
-            nextcloud_host: Base URL of the Nextcloud instance (e.g., https://cloud.example.com)
-            userinfo_uri: Full URL to the userinfo endpoint
-            jwks_uri: Full URL to the JWKS endpoint (for JWT verification)
-            issuer: Expected issuer claim value (for JWT verification)
-            introspection_uri: Full URL to the introspection endpoint (for opaque tokens)
-            client_id: OAuth client ID (required for introspection)
-            client_secret: OAuth client secret (required for introspection)
-            cache_ttl: Time-to-live for cached tokens in seconds (default: 3600)
-        """
-        self.nextcloud_host = nextcloud_host.rstrip("/")
-        self.userinfo_uri = userinfo_uri
-        self.jwks_uri = jwks_uri
-        self.issuer = issuer
-        self.introspection_uri = introspection_uri
-        self.client_id = client_id
-        self.client_secret = client_secret
-        self.cache_ttl = cache_ttl
-
-        # Cache: token -> (userinfo, expiry_timestamp)
-        self._token_cache: dict[str, tuple[dict[str, Any], float]] = {}
-
-        # HTTP client for userinfo/introspection requests
-        self._client = httpx.AsyncClient(timeout=10.0)
-
-        # PyJWKClient for JWT verification (lazy initialization)
-        self._jwks_client: PyJWKClient | None = None
-        if jwks_uri:
-            logger.info(f"JWT verification enabled with JWKS URI: {jwks_uri}")
-            self._jwks_client = PyJWKClient(jwks_uri, cache_keys=True)
-
-        # Introspection support
-        if introspection_uri and client_id and client_secret:
-            logger.info(f"Token introspection enabled: {introspection_uri}")
-        elif introspection_uri:
-            logger.warning(
-                "Introspection URI provided but missing client credentials - introspection disabled"
-            )
-
-    async def verify_token(self, token: str) -> AccessToken | None:
-        """
-        Verify a bearer token using JWT verification, introspection, or userinfo endpoint.
-
-        This method:
-        1. Checks the cache first for recent validations
-        2. Attempts JWT verification if JWKS is configured and token looks like JWT
-        3. Falls back to introspection for opaque tokens (if configured)
-        4. Falls back to userinfo endpoint as last resort
-        5. Returns AccessToken with username and scopes
-
-        Args:
-            token: The bearer token to verify
-
-        Returns:
-            AccessToken if valid, None if invalid or expired
-        """
-        # Check cache first
-        cached = self._get_cached_token(token)
-        if cached:
-            logger.debug("Token found in cache")
-            return cached
-
-        # Try JWT verification first if enabled and token looks like JWT
-        is_jwt_format = self._is_jwt_format(token)
-        logger.debug(
-            f"Token format check: is_jwt_format={is_jwt_format}, _jwks_client={self._jwks_client is not None}"
-        )
-        if self._jwks_client and is_jwt_format:
-            logger.debug("Attempting JWT verification...")
-            jwt_result = self._verify_jwt(token)
-            if jwt_result:
-                logger.info("Token validated via JWT verification")
-                return jwt_result
-            else:
-                logger.warning("JWT verification failed, will try other methods")
-
-        # For opaque tokens, try introspection if available
-        if self.introspection_uri and self.client_id and self.client_secret:
-            logger.debug("Attempting token introspection...")
-            try:
-                introspection_result = await self._verify_via_introspection(token)
-                if introspection_result:
-                    logger.info("Token validated via introspection")
-                    return introspection_result
-            except Exception as e:
-                logger.warning(f"Introspection failed: {e}")
-
-        # Fall back to userinfo endpoint validation (last resort)
-        logger.debug("Attempting userinfo endpoint validation...")
-        try:
-            return await self._verify_via_userinfo(token)
-        except Exception as e:
-            logger.warning(f"Token verification failed: {e}")
-            return None
-
-    def _is_jwt_format(self, token: str) -> bool:
-        """
-        Check if token looks like a JWT (has 3 parts separated by dots).
-
-        Args:
-            token: The token to check
-
-        Returns:
-            True if token appears to be JWT format
-        """
-        return "." in token and token.count(".") == 2
-
-    def _verify_jwt(self, token: str) -> AccessToken | None:
-        """
-        Verify JWT token with signature validation using JWKS.
-
-        Args:
-            token: The JWT token to verify
-
-        Returns:
-            AccessToken if valid, None if invalid
-        """
-        try:
-            # Get signing key from JWKS
-            signing_key = self._jwks_client.get_signing_key_from_jwt(token)
-
-            # Verify and decode JWT
-            # Accept tokens with audience: "mcp-server" or ["mcp-server", "nextcloud"]
-            # This allows:
-            # 1. Tokens from MCP clients (aud: "mcp-server")
-            # 2. Tokens for Nextcloud APIs (aud: "nextcloud")
-            # 3. Tokens for both (aud: ["mcp-server", "nextcloud"])
-            payload = jwt.decode(
-                token,
-                signing_key.key,
-                algorithms=["RS256"],
-                issuer=self.issuer,
-                audience=["mcp-server", "nextcloud"],  # Accept either audience
-                options={
-                    "verify_signature": True,
-                    "verify_exp": True,
-                    "verify_iat": True,
-                    "verify_iss": True if self.issuer else False,
-                    "verify_aud": True,  # Enable audience validation
-                },
-            )
-
-            logger.debug(f"JWT verified successfully for user: {payload.get('sub')}")
-            logger.debug(f"Full JWT payload: {payload}")
-
-            # Extract username (sub claim, with fallback to preferred_username)
-            # Some OIDC providers (like Keycloak) may not include sub in access tokens
-            username = payload.get("sub") or payload.get("preferred_username")
-            if not username:
-                logger.error(
-                    "No 'sub' or 'preferred_username' claim found in JWT payload"
-                )
-                return None
-
-            # Extract scopes from scope claim (space-separated string)
-            scope_string = payload.get("scope", "")
-            scopes = scope_string.split() if scope_string else []
-            logger.debug(
-                f"Extracted scopes from JWT - scope claim: '{scope_string}' -> scopes list: {scopes}"
-            )
-
-            # Extract expiration
-            exp = payload.get("exp")
-            if not exp:
-                logger.warning("No 'exp' claim in JWT, using default TTL")
-                exp = int(time.time() + self.cache_ttl)
-
-            # Cache the result
-            userinfo = {
-                "sub": username,
-                "scope": scope_string,
-                **{k: v for k, v in payload.items() if k not in ["sub", "scope"]},
-            }
-            self._token_cache[token] = (userinfo, exp)
-
-            return AccessToken(
-                token=token,
-                client_id=payload.get("client_id", ""),
-                scopes=scopes,
-                expires_at=exp,
-                resource=username,  # Store username in resource field (RFC 8707)
-            )
-
-        except jwt.ExpiredSignatureError:
-            logger.info("JWT token has expired")
-            return None
-        except jwt.InvalidIssuerError as e:
-            logger.warning(f"JWT issuer validation failed: {e}")
-            return None
-        except jwt.InvalidTokenError as e:
-            logger.warning(f"JWT validation failed: {e}")
-            return None
-        except Exception as e:
-            logger.error(f"Unexpected error during JWT verification: {e}")
-            return None
-
-    async def _verify_via_introspection(self, token: str) -> AccessToken | None:
-        """
-        Validate token by calling the introspection endpoint (RFC 7662).
-
-        This method validates opaque tokens and retrieves their scopes.
-
-        Args:
-            token: The bearer token to introspect
-
-        Returns:
-            AccessToken if active, None if inactive or invalid
-        """
-        try:
-            # Introspection requires client authentication
-            response = await self._client.post(
-                self.introspection_uri,
-                data={"token": token},
-                auth=(self.client_id, self.client_secret),
-            )
-
-            if response.status_code == 200:
-                introspection_data = response.json()
-
-                # Check if token is active
-                if not introspection_data.get("active", False):
-                    logger.info("Token introspection returned inactive=false")
-                    return None
-
-                logger.debug(
-                    f"Token introspected successfully for user: {introspection_data.get('sub')}"
-                )
-
-                # Extract username
-                username = introspection_data.get("sub") or introspection_data.get(
-                    "username"
-                )
-                if not username:
-                    logger.error("No username found in introspection response")
-                    return None
-
-                # Extract scopes (space-separated string)
-                scope_string = introspection_data.get("scope", "")
-                scopes = scope_string.split() if scope_string else []
-                logger.debug(f"Extracted scopes from introspection: {scopes}")
-
-                # Extract expiration
-                exp = introspection_data.get("exp")
-                if exp:
-                    expiry = float(exp)
-                else:
-                    logger.warning(
-                        "No 'exp' in introspection response, using default TTL"
-                    )
-                    expiry = time.time() + self.cache_ttl
-
-                # Cache the result
-                cache_data = {
-                    "sub": username,
-                    "scope": scope_string,
-                    **{
-                        k: v
-                        for k, v in introspection_data.items()
-                        if k not in ["sub", "scope", "active"]
-                    },
-                }
-                self._token_cache[token] = (cache_data, expiry)
-
-                return AccessToken(
-                    token=token,
-                    client_id=introspection_data.get("client_id", ""),
-                    scopes=scopes,
-                    expires_at=int(expiry),
-                    resource=username,
-                )
-
-            elif response.status_code in (400, 401, 403):
-                logger.warning(
-                    f"Token introspection failed: HTTP {response.status_code}. "
-                    f"This may indicate: (1) Client credentials mismatch - trying to introspect "
-                    f"token issued to different OAuth client, (2) Expired client credentials, "
-                    f"(3) Invalid token. Will fall back to userinfo endpoint. "
-                    f"Response: {response.text[:200] if response.text else 'empty'}"
-                )
-                return None
-            else:
-                logger.warning(
-                    f"Unexpected response from introspection: {response.status_code}. "
-                    f"Response: {response.text[:200] if response.text else 'empty'}"
-                )
-                return None
-
-        except httpx.TimeoutException:
-            logger.error("Timeout while introspecting token")
-            return None
-        except httpx.RequestError as e:
-            logger.error(f"Network error while introspecting token: {e}")
-            return None
-        except Exception as e:
-            logger.error(f"Unexpected error during token introspection: {e}")
-            return None
-
-    async def _verify_via_userinfo(self, token: str) -> AccessToken | None:
-        """
-        Validate token by calling the userinfo endpoint.
-
-        Args:
-            token: The bearer token to verify
-
-        Returns:
-            AccessToken if valid, None otherwise
-        """
-        try:
-            response = await self._client.get(
-                self.userinfo_uri, headers={"Authorization": f"Bearer {token}"}
-            )
-
-            if response.status_code == 200:
-                userinfo = response.json()
-                logger.debug(
-                    f"Token validated successfully for user: {userinfo.get('sub')}"
-                )
-
-                # Cache the result
-                expiry = time.time() + self.cache_ttl
-                self._token_cache[token] = (userinfo, expiry)
-
-                # Create AccessToken with username in resource field (workaround for MCP SDK)
-                username = userinfo.get("sub") or userinfo.get("preferred_username")
-                if not username:
-                    logger.error("No username found in userinfo response")
-                    return None
-
-                return AccessToken(
-                    token=token,
-                    client_id="",  # Not available from userinfo
-                    scopes=self._extract_scopes(userinfo),
-                    expires_at=int(expiry),
-                    resource=username,  # Store username in resource field (RFC 8707)
-                )
-
-            elif response.status_code in (400, 401, 403):
-                logger.info(f"Token validation failed: HTTP {response.status_code}")
-                return None
-            else:
-                logger.warning(
-                    f"Unexpected response from userinfo: {response.status_code}"
-                )
-                return None
-
-        except httpx.TimeoutException:
-            logger.error("Timeout while validating token via userinfo endpoint")
-            return None
-        except httpx.RequestError as e:
-            logger.error(f"Network error while validating token: {e}")
-            return None
-        except Exception as e:
-            logger.error(f"Unexpected error during token validation: {e}")
-            return None
-
-    def _get_cached_token(self, token: str) -> AccessToken | None:
-        """
-        Retrieve a token from cache if not expired.
-
-        Args:
-            token: The bearer token to look up
-
-        Returns:
-            AccessToken if cached and valid, None otherwise
-        """
-        if token not in self._token_cache:
-            return None
-
-        userinfo, expiry = self._token_cache[token]
-
-        # Check if expired
-        if time.time() >= expiry:
-            logger.debug("Cached token expired, removing from cache")
-            del self._token_cache[token]
-            return None
-
-        # Return cached AccessToken
-        username = userinfo.get("sub") or userinfo.get("preferred_username")
-        return AccessToken(
-            token=token,
-            client_id="",
-            scopes=self._extract_scopes(userinfo),
-            expires_at=int(expiry),
-            resource=username,
-        )
-
-    def _extract_scopes(self, userinfo: dict[str, Any]) -> list[str]:
-        """
-        Extract scopes from userinfo response.
-
-        First attempts to read actual scopes from the 'scope' field (RFC 8693).
-        If not present, infers scopes from the claims present in the response.
-
-        Args:
-            userinfo: The userinfo response dictionary
-
-        Returns:
-            List of scopes (actual or inferred)
-        """
-        # Try to get actual scopes from userinfo response (if OIDC provider includes it)
-        scope_string = userinfo.get("scope")
-        if scope_string:
-            scopes = scope_string.split() if isinstance(scope_string, str) else []
-            if scopes:
-                logger.debug(
-                    f"Using actual scopes from userinfo: {scopes} (scope field present)"
-                )
-                return scopes
-
-        # Fallback: Infer scopes from claims present in response
-        # This maintains backward compatibility with OIDC providers that don't
-        # include the scope field in userinfo responses
-        logger.debug(
-            "No scope field in userinfo response, inferring scopes from claims"
-        )
-        scopes = ["openid"]  # Always present
-
-        if "email" in userinfo:
-            scopes.append("email")
-
-        if any(
-            key in userinfo for key in ["name", "given_name", "family_name", "picture"]
-        ):
-            scopes.append("profile")
-
-        if "roles" in userinfo:
-            scopes.append("roles")
-
-        if "groups" in userinfo:
-            scopes.append("groups")
-
-        logger.debug(f"Inferred scopes from userinfo claims: {scopes}")
-        return scopes
-
-    def clear_cache(self):
-        """Clear the token cache."""
-        self._token_cache.clear()
-        logger.debug("Token cache cleared")
-
-    async def close(self):
-        """Cleanup resources."""
-        await self._client.aclose()
-        logger.debug("Token verifier closed")

nextcloud_mcp_server/auth/unified_verifier.py

@@ -0,0 +1,417 @@
+"""
+Unified Token Verifier for ADR-005 Token Audience Validation.
+
+This module replaces both NextcloudTokenVerifier and ProgressiveConsentTokenVerifier
+with a single implementation that supports two compliant OAuth modes:
+
+1. Multi-audience mode (default): Validates MCP audience per RFC 7519 (resource servers
+   validate only their own audience). Nextcloud independently validates its own audience.
+2. Token exchange mode (opt-in): Tokens have MCP audience only, exchanged for Nextcloud tokens
+
+Key Design Principles:
+- Token verification happens HERE (validates MCP audience per OAuth spec)
+- Token exchange happens in context_helper.py (when creating NextcloudClient)
+- No token passthrough allowed (complies with MCP Security Specification)
+- Token reuse IS allowed for multi-audience tokens (RFC 8707)
+"""
+
+import hashlib
+import logging
+import time
+from typing import Any
+
+import httpx
+import jwt
+from jwt import PyJWKClient
+from mcp.server.auth.provider import AccessToken, TokenVerifier
+
+from nextcloud_mcp_server.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+class UnifiedTokenVerifier(TokenVerifier):
+    """
+    Unified token verifier supporting both multi-audience and token exchange modes.
+    Compliant with MCP security specification - no token pass-through.
+
+    This verifier:
+    1. Validates tokens using JWT verification with JWKS or introspection fallback
+    2. Enforces proper audience validation based on configured mode
+    3. Caches successful validations to avoid repeated API calls
+
+    Mode Selection (via ENABLE_TOKEN_EXCHANGE setting):
+    - False/omit (default): Multi-audience mode - validates MCP audience only (per RFC 7519).
+      Nextcloud independently validates its own audience when receiving API calls.
+    - True: Exchange mode - requires MCP audience only, then exchanges for Nextcloud token
+    """
+
+    def __init__(self, settings: Settings):
+        """
+        Initialize the unified token verifier.
+
+        Args:
+            settings: Application settings containing OAuth configuration
+        """
+        self.settings = settings
+        self.mode = "exchange" if settings.enable_token_exchange else "multi-audience"
+
+        # Common components for all modes
+        self.http_client = httpx.AsyncClient(timeout=10.0)
+
+        # JWT verification support
+        self.jwks_client: PyJWKClient | None = None
+        if hasattr(settings, "jwks_uri") and settings.jwks_uri:
+            logger.info(f"JWT verification enabled with JWKS URI: {settings.jwks_uri}")
+            self.jwks_client = PyJWKClient(settings.jwks_uri, cache_keys=True)
+
+        # Introspection support (for opaque tokens)
+        self.introspection_uri: str | None = None
+        if (
+            hasattr(settings, "introspection_uri")
+            and settings.introspection_uri
+            and settings.oidc_client_id
+            and settings.oidc_client_secret
+        ):
+            self.introspection_uri = settings.introspection_uri
+            logger.info(f"Token introspection enabled: {self.introspection_uri}")
+
+        # Token cache: token_hash -> (userinfo, expiry_timestamp)
+        self._token_cache: dict[str, tuple[dict[str, Any], float]] = {}
+        self.cache_ttl = 3600  # 1 hour default
+
+        logger.info(
+            f"UnifiedTokenVerifier initialized in {self.mode} mode. "
+            f"MCP audience: {settings.oidc_client_id} or {settings.nextcloud_mcp_server_url}, "
+            f"Nextcloud resource URI: {settings.nextcloud_resource_uri}"
+        )
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """
+        Verify token according to MCP TokenVerifier protocol.
+
+        Per RFC 7519, we validate only MCP audience. The mode determines what
+        happens AFTER verification in context_helper.py:
+        - Multi-audience mode: Use token directly (Nextcloud validates its own audience)
+        - Exchange mode: Exchange for Nextcloud-audience token via RFC 8693
+
+        Args:
+            token: Bearer token to verify
+
+        Returns:
+            AccessToken if valid with MCP audience, None otherwise
+        """
+        # Check cache first
+        cached = self._get_cached_token(token)
+        if cached:
+            logger.debug("Token found in cache")
+            return cached
+
+        # Both modes do the same validation (MCP audience only)
+        return await self._verify_mcp_audience(token)
+
+    async def _verify_mcp_audience(self, token: str) -> AccessToken | None:
+        """
+        Validate token has MCP audience.
+
+        Per RFC 7519 Section 4.1.3, resource servers validate only their own
+        presence in the audience claim. We don't validate Nextcloud's audience -
+        that's Nextcloud's responsibility when it receives the token.
+
+        Args:
+            token: Bearer token to verify
+
+        Returns:
+            AccessToken if valid with MCP audience, None otherwise
+        """
+        try:
+            # Attempt JWT verification first
+            if self._is_jwt_format(token) and self.jwks_client:
+                payload = await self._verify_jwt_signature(token)
+            else:
+                # Fall back to introspection for opaque tokens
+                payload = await self._introspect_token(token)
+                if not payload:
+                    return None
+
+            # Check payload is valid
+            if not payload:
+                return None
+
+            # Validate MCP audience is present
+            if not self._has_mcp_audience(payload):
+                audiences = payload.get("aud", [])
+                logger.error(
+                    f"Token rejected: Missing MCP audience. "
+                    f"Got {audiences}, need MCP ({self.settings.oidc_client_id} or "
+                    f"{self.settings.nextcloud_mcp_server_url})"
+                )
+                return None
+
+            # Log based on mode for clarity
+            if self.mode == "multi-audience":
+                logger.info(
+                    "MCP audience validated - token can be used directly "
+                    "(Nextcloud will validate its own audience)"
+                )
+            else:
+                logger.info(
+                    "MCP audience validated - token will be exchanged for Nextcloud access"
+                )
+
+            return self._create_access_token(token, payload)
+
+        except Exception as e:
+            logger.error(f"Token verification failed: {e}")
+            return None
+
+    def _has_mcp_audience(self, payload: dict[str, Any]) -> bool:
+        """
+        Check if token has MCP audience.
+
+        Per RFC 7519 Section 4.1.3, resource servers should only validate their own
+        presence in the audience claim. We don't validate Nextcloud's audience - that's
+        Nextcloud's responsibility when it receives the token.
+
+        Args:
+            payload: Decoded token payload
+
+        Returns:
+            True if MCP audience present, False otherwise
+        """
+        audiences = payload.get("aud", [])
+        if isinstance(audiences, str):
+            audiences = [audiences]
+
+        audiences_set = set(audiences)
+
+        # MCP must have at least one: client_id OR server_url OR server_url/mcp
+        return bool(
+            self.settings.oidc_client_id in audiences_set
+            or (
+                self.settings.nextcloud_mcp_server_url
+                and (
+                    self.settings.nextcloud_mcp_server_url in audiences_set
+                    or f"{self.settings.nextcloud_mcp_server_url}/mcp" in audiences_set
+                )
+            )
+        )
+
+    def _is_jwt_format(self, token: str) -> bool:
+        """
+        Check if token looks like a JWT (has 3 parts separated by dots).
+
+        Args:
+            token: The token to check
+
+        Returns:
+            True if token appears to be JWT format
+        """
+        return "." in token and token.count(".") == 2
+
+    async def _verify_jwt_signature(self, token: str) -> dict[str, Any] | None:
+        """
+        Verify JWT token with signature validation using JWKS.
+
+        Args:
+            token: JWT token to verify
+
+        Returns:
+            Decoded payload if valid, None if invalid
+        """
+        try:
+            assert self.jwks_client is not None  # Caller should check before calling
+
+            # Get signing key from JWKS
+            signing_key = self.jwks_client.get_signing_key_from_jwt(token)
+
+            # Verify and decode JWT
+            # Note: We don't validate audience here - that's done separately based on mode
+            payload = jwt.decode(
+                token,
+                signing_key.key,
+                algorithms=["RS256"],
+                issuer=self.settings.oidc_issuer
+                if hasattr(self.settings, "oidc_issuer")
+                else None,
+                options={
+                    "verify_signature": True,
+                    "verify_exp": True,
+                    "verify_iat": True,
+                    "verify_iss": True
+                    if hasattr(self.settings, "oidc_issuer")
+                    and self.settings.oidc_issuer
+                    else False,
+                    "verify_aud": False,  # We handle audience validation separately
+                },
+            )
+
+            logger.debug(f"JWT signature verified for user: {payload.get('sub')}")
+            return payload
+
+        except jwt.ExpiredSignatureError:
+            logger.info("JWT token has expired")
+            return None
+        except jwt.InvalidIssuerError as e:
+            logger.warning(f"JWT issuer validation failed: {e}")
+            return None
+        except jwt.InvalidTokenError as e:
+            logger.warning(f"JWT validation failed: {e}")
+            return None
+        except Exception as e:
+            logger.error(f"Unexpected error during JWT verification: {e}")
+            return None
+
+    async def _introspect_token(self, token: str) -> dict[str, Any] | None:
+        """
+        Validate token by calling the introspection endpoint (RFC 7662).
+
+        Args:
+            token: Bearer token to introspect
+
+        Returns:
+            Token payload if active, None if inactive or invalid
+        """
+        if not self.introspection_uri:
+            logger.debug("No introspection endpoint configured")
+            return None
+
+        try:
+            # Introspection requires client authentication
+            response = await self.http_client.post(
+                self.introspection_uri,
+                data={"token": token},
+                auth=(self.settings.oidc_client_id, self.settings.oidc_client_secret),
+            )
+
+            if response.status_code == 200:
+                introspection_data = response.json()
+
+                # Check if token is active
+                if not introspection_data.get("active", False):
+                    logger.info("Token introspection returned inactive=false")
+                    return None
+
+                logger.debug(
+                    f"Token introspected successfully for user: {introspection_data.get('sub')}"
+                )
+                return introspection_data
+
+            elif response.status_code in (400, 401, 403):
+                logger.warning(
+                    f"Token introspection failed: HTTP {response.status_code}. "
+                    f"Response: {response.text[:200] if response.text else 'empty'}"
+                )
+                return None
+            else:
+                logger.warning(
+                    f"Unexpected response from introspection: {response.status_code}. "
+                    f"Response: {response.text[:200] if response.text else 'empty'}"
+                )
+                return None
+
+        except httpx.TimeoutException:
+            logger.error("Timeout while introspecting token")
+            return None
+        except httpx.RequestError as e:
+            logger.error(f"Network error while introspecting token: {e}")
+            return None
+        except Exception as e:
+            logger.error(f"Unexpected error during token introspection: {e}")
+            return None
+
+    def _create_access_token(
+        self, token: str, payload: dict[str, Any]
+    ) -> AccessToken | None:
+        """
+        Create AccessToken object from validated token payload.
+
+        Args:
+            token: The bearer token
+            payload: Validated token payload
+
+        Returns:
+            AccessToken object or None if required fields missing
+        """
+        # Extract username (sub claim, with fallback to preferred_username)
+        username = payload.get("sub") or payload.get("preferred_username")
+        if not username:
+            logger.error(
+                "No 'sub' or 'preferred_username' claim found in token payload"
+            )
+            return None
+
+        # Extract scopes from scope claim (space-separated string)
+        scope_string = payload.get("scope", "")
+        scopes = scope_string.split() if scope_string else []
+        logger.debug(
+            f"Extracted scopes from token - scope claim: '{scope_string}' -> scopes list: {scopes}"
+        )
+
+        # Extract expiration
+        exp = payload.get("exp")
+        if not exp:
+            logger.warning("No 'exp' claim in token, using default TTL")
+            exp = int(time.time() + self.cache_ttl)
+
+        # Cache the result
+        token_hash = hashlib.sha256(token.encode()).hexdigest()
+        userinfo = {
+            "sub": username,
+            "scope": scope_string,
+            **{k: v for k, v in payload.items() if k not in ["sub", "scope"]},
+        }
+        self._token_cache[token_hash] = (userinfo, exp)
+
+        return AccessToken(
+            token=token,
+            client_id=payload.get("client_id", ""),
+            scopes=scopes,
+            expires_at=exp,
+            resource=username,  # Store username in resource field (RFC 8707)
+        )
+
+    def _get_cached_token(self, token: str) -> AccessToken | None:
+        """
+        Retrieve a token from cache if not expired.
+
+        Args:
+            token: The bearer token to look up
+
+        Returns:
+            AccessToken if cached and valid, None otherwise
+        """
+        token_hash = hashlib.sha256(token.encode()).hexdigest()
+        if token_hash not in self._token_cache:
+            return None
+
+        userinfo, expiry = self._token_cache[token_hash]
+
+        # Check if expired
+        if time.time() >= expiry:
+            logger.debug("Cached token expired, removing from cache")
+            del self._token_cache[token_hash]
+            return None
+
+        # Return cached AccessToken
+        username = userinfo.get("sub") or userinfo.get("preferred_username")
+        scope_string = userinfo.get("scope", "")
+        scopes = scope_string.split() if scope_string else []
+
+        return AccessToken(
+            token=token,
+            client_id=userinfo.get("client_id", ""),
+            scopes=scopes,
+            expires_at=int(expiry),
+            resource=username,
+        )
+
+    def clear_cache(self):
+        """Clear the token cache."""
+        self._token_cache.clear()
+        logger.debug("Token cache cleared")
+
+    async def close(self):
+        """Cleanup resources."""
+        await self.http_client.aclose()
+        logger.debug("Unified token verifier closed")

nextcloud_mcp_server/auth/userinfo_routes.py

@@ -141,9 +141,23 @@ async def _get_user_info(request: Request) -> dict[str, Any]:
 
     try:
         # Check if background access was granted (refresh token exists)
+        # This works for both Flow 2 (elicitation) and browser login
         token_data = await storage.get_refresh_token(session_id)
         background_access_granted = token_data is not None
 
+        # Build background access details
+        background_access_details = None
+        if token_data:
+            background_access_details = {
+                "flow_type": token_data.get("flow_type", "unknown"),
+                "provisioned_at": token_data.get("provisioned_at", "unknown"),
+                "provisioning_client_id": token_data.get(
+                    "provisioning_client_id", "N/A"
+                ),
+                "scopes": token_data.get("scopes", "N/A"),
+                "token_audience": token_data.get("token_audience", "unknown"),
+            }
+
         # Retrieve cached user profile (no token operations!)
         profile_data = await storage.get_user_profile(session_id)
 
@@ -153,6 +167,7 @@ async def _get_user_info(request: Request) -> dict[str, Any]:
             "auth_mode": "oauth",
             "session_id": session_id[:16] + "...",  # Truncated for security
             "background_access_granted": background_access_granted,
+            "background_access_details": background_access_details,
         }
 
         # Include cached profile if available
@@ -291,6 +306,47 @@ async def user_info_html(request: Request) -> HTMLResponse:
     session_info_html = ""
     if auth_mode == "oauth" and "session_id" in user_context:
         session_id = user_context.get("session_id", "unknown")
+        background_access_granted = user_context.get("background_access_granted", False)
+        background_details = user_context.get("background_access_details")
+
+        # Build background access section
+        background_html = ""
+        if background_access_granted and background_details:
+            flow_type = background_details.get("flow_type", "unknown")
+            provisioned_at = background_details.get("provisioned_at", "unknown")
+            scopes = background_details.get("scopes", "N/A")
+            token_audience = background_details.get("token_audience", "unknown")
+
+            background_html = f"""
+            <tr>
+                <td><strong>Background Access</strong></td>
+                <td><span style="color: #4caf50; font-weight: bold;">✓ Granted</span></td>
+            </tr>
+            <tr>
+                <td><strong>Flow Type</strong></td>
+                <td>{flow_type}</td>
+            </tr>
+            <tr>
+                <td><strong>Provisioned At</strong></td>
+                <td>{provisioned_at}</td>
+            </tr>
+            <tr>
+                <td><strong>Token Audience</strong></td>
+                <td>{token_audience}</td>
+            </tr>
+            <tr>
+                <td><strong>Scopes</strong></td>
+                <td><code style="font-size: 11px;">{scopes}</code></td>
+            </tr>
+            """
+        else:
+            background_html = """
+            <tr>
+                <td><strong>Background Access</strong></td>
+                <td><span style="color: #999;">Not Granted</span></td>
+            </tr>
+            """
+
         session_info_html = f"""
         <h2>Session Information</h2>
         <table>
@@ -298,9 +354,23 @@ async def user_info_html(request: Request) -> HTMLResponse:
                 <td><strong>Session ID</strong></td>
                 <td><code>{session_id}</code></td>
             </tr>
+            {background_html}
         </table>
         """
 
+        # Add revoke button if background access is granted
+        if background_access_granted:
+            revoke_url = str(request.url_for("revoke_session_endpoint"))
+            session_info_html += f"""
+            <div style="margin-top: 15px;">
+                <form method="post" action="{revoke_url}" onsubmit="return confirm('Are you sure you want to revoke background access? This will delete the refresh token.');">
+                    <button type="submit" style="padding: 8px 16px; background-color: #ff9800; color: white; border: none; border-radius: 4px; cursor: pointer; font-size: 14px;">
+                        Revoke Background Access
+                    </button>
+                </form>
+            </div>
+            """
+
     # Build IdP profile HTML
     idp_profile_html = ""
     if "idp_profile" in user_context:
@@ -446,3 +516,117 @@ async def user_info_html(request: Request) -> HTMLResponse:
     """
 
     return HTMLResponse(content=html_content)
+
+
+@requires("authenticated", redirect="oauth_login")
+async def revoke_session(request: Request) -> HTMLResponse:
+    """Revoke background access (delete refresh token).
+
+    This endpoint allows users to revoke the refresh token that grants
+    background access to Nextcloud resources. The session cookie remains
+    valid for browser UI access, but background jobs will no longer work.
+
+    Args:
+        request: Starlette request object
+
+    Returns:
+        HTML response confirming revocation or showing error
+    """
+    oauth_ctx = getattr(request.app.state, "oauth_context", None)
+
+    if not oauth_ctx:
+        return HTMLResponse(
+            """
+            <!DOCTYPE html>
+            <html>
+            <head><title>Error</title></head>
+            <body>
+                <h1>Error</h1>
+                <p>OAuth mode not enabled</p>
+            </body>
+            </html>
+            """,
+            status_code=400,
+        )
+
+    storage = oauth_ctx.get("storage")
+    session_id = request.cookies.get("mcp_session")
+
+    if not storage or not session_id:
+        return HTMLResponse(
+            """
+            <!DOCTYPE html>
+            <html>
+            <head><title>Error</title></head>
+            <body>
+                <h1>Error</h1>
+                <p>Session not found</p>
+            </body>
+            </html>
+            """,
+            status_code=400,
+        )
+
+    try:
+        # Delete the refresh token
+        logger.info(f"Revoking background access for session {session_id[:16]}...")
+        await storage.delete_refresh_token(session_id)
+        logger.info(f"✓ Background access revoked for session {session_id[:16]}...")
+
+        # Redirect back to user page
+        user_page_url = str(request.url_for("user_info_html"))
+
+        return HTMLResponse(
+            f"""
+            <!DOCTYPE html>
+            <html lang="en">
+            <head>
+                <meta charset="UTF-8">
+                <meta http-equiv="refresh" content="2;url={user_page_url}">
+                <title>Background Access Revoked</title>
+                <style>
+                    body {{
+                        font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
+                        max-width: 600px;
+                        margin: 50px auto;
+                        padding: 20px;
+                        text-align: center;
+                    }}
+                    .success {{
+                        background-color: #e8f5e9;
+                        border: 2px solid #4caf50;
+                        padding: 30px;
+                        border-radius: 8px;
+                    }}
+                    h1 {{
+                        color: #4caf50;
+                    }}
+                </style>
+            </head>
+            <body>
+                <div class="success">
+                    <h1>✓ Background Access Revoked</h1>
+                    <p>Your refresh token has been deleted successfully.</p>
+                    <p>Browser session remains active.</p>
+                    <p>Redirecting back to user page...</p>
+                </div>
+            </body>
+            </html>
+            """
+        )
+
+    except Exception as e:
+        logger.error(f"Failed to revoke background access: {e}")
+        return HTMLResponse(
+            f"""
+            <!DOCTYPE html>
+            <html>
+            <head><title>Error</title></head>
+            <body>
+                <h1>Error</h1>
+                <p>Failed to revoke background access: {e}</p>
+            </body>
+            </html>
+            """,
+            status_code=500,
+        )

nextcloud_mcp_server/client/calendar.py

@@ -100,7 +100,7 @@ class CalendarClient:
         # Use custom PROPFIND with CalendarServer namespace (cs:) for calendar-color.
         # caldav library's nsmap lacks "CS" namespace, and its CalendarColor uses
         # Apple iCal namespace which Nextcloud doesn't recognize.
-        from lxml import etree
+        from lxml import etree  # type: ignore[import-untyped]
 
         propfind_body = """<?xml version="1.0" encoding="utf-8"?>
 <d:propfind xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/" xmlns:c="urn:ietf:params:xml:ns:caldav">
@@ -261,11 +261,12 @@ class CalendarClient:
         result = []
         for event in events:
             await event.load(only_if_unloaded=True)
-            event_dict = self._parse_ical_event(event.data)
-            if event_dict:
-                event_dict["href"] = str(event.url)
-                event_dict["etag"] = ""
-                result.append(event_dict)
+            if event.data:
+                event_dict = self._parse_ical_event(event.data)
+                if event_dict:
+                    event_dict["href"] = str(event.url)
+                    event_dict["etag"] = ""
+                    result.append(event_dict)
 
             if len(result) >= limit:
                 break
@@ -314,8 +315,8 @@ class CalendarClient:
         await event.load(only_if_unloaded=True)
 
         # Merge updates into existing iCal data
-        updated_ical = self._merge_ical_properties(event.data, event_data, event_uid)
-        event.data = updated_ical
+        updated_ical = self._merge_ical_properties(event.data, event_data, event_uid)  # type: ignore[arg-type]
+        event.data = updated_ical  # type: ignore[misc]
 
         await event.save()
 
@@ -349,7 +350,7 @@ class CalendarClient:
         event = await calendar.event_by_uid(event_uid)
         await event.load(only_if_unloaded=True)
 
-        event_data = self._parse_ical_event(event.data)
+        event_data = self._parse_ical_event(event.data) if event.data else None  # type: ignore[arg-type]
         if not event_data:
             raise ValueError(f"Failed to parse event data for {event_uid}")
 
@@ -416,7 +417,10 @@ class CalendarClient:
             # Only load if data not already present from REPORT response
             # This avoids 404 errors for virtual calendars (e.g., Deck boards)
             await todo.load(only_if_unloaded=True)
-            todo_dict = self._parse_ical_todo(todo.data)
+            if todo.data:
+                todo_dict = self._parse_ical_todo(todo.data)  # type: ignore[arg-type]
+            else:
+                continue
             if todo_dict:
                 todo_dict["href"] = str(todo.url)
                 todo_dict["etag"] = ""
@@ -470,12 +474,14 @@ class CalendarClient:
             await todo.load(only_if_unloaded=True)
 
             logger.debug(
-                f"Loaded todo {todo_uid}, current data length: {len(todo.data)}"
+                f"Loaded todo {todo_uid}, current data length: {len(todo.data)}"  # type: ignore
             )
 
             # Merge updates into existing iCal data
             updated_ical = self._merge_ical_todo_properties(
-                todo.data, todo_data, todo_uid
+                todo.data,  # type: ignore[arg-type]
+                todo_data,
+                todo_uid,
             )
             logger.debug(f"Merged iCal data length: {len(updated_ical)}")
             logger.debug(f"Updated iCal content:\n{updated_ical}")

nextcloud_mcp_server/client/contacts.py

@@ -124,7 +124,7 @@ class ContactsClient(BaseNextcloudClient):
         carddav_path = self._get_carddav_base_path()
         url = f"{carddav_path}/{addressbook}/{uid}.vcf"
 
-        contact = Contact(fn=contact_data.get("fn"), uid=uid)
+        contact = Contact(fn=contact_data.get("fn"), uid=uid)  # type: ignore
         if "email" in contact_data:
             contact.email = [{"value": contact_data["email"], "type": ["HOME"]}]
         if "tel" in contact_data:
@@ -174,7 +174,7 @@ class ContactsClient(BaseNextcloudClient):
             )
         else:
             # Fallback to creating new vCard if we couldn't get existing
-            contact = Contact(fn=contact_data.get("fn"), uid=uid)
+            contact = Contact(fn=contact_data.get("fn"), uid=uid)  # type: ignore
             if "email" in contact_data:
                 contact.email = [{"value": contact_data["email"], "type": ["HOME"]}]
             if "tel" in contact_data:

nextcloud_mcp_server/config.py

@@ -129,16 +129,29 @@ class Settings:
     oidc_discovery_url: Optional[str] = None
     oidc_client_id: Optional[str] = None
     oidc_client_secret: Optional[str] = None
+    oidc_issuer: Optional[str] = None
 
     # Nextcloud settings
     nextcloud_host: Optional[str] = None
     nextcloud_username: Optional[str] = None
     nextcloud_password: Optional[str] = None
 
+    # ADR-005: Token Audience Validation (required for OAuth mode)
+    nextcloud_mcp_server_url: Optional[str] = None  # MCP server URL (used as audience)
+    nextcloud_resource_uri: Optional[str] = None  # Nextcloud resource identifier
+
+    # Token verification endpoints
+    jwks_uri: Optional[str] = None
+    introspection_uri: Optional[str] = None
+    userinfo_uri: Optional[str] = None
+
     # Progressive Consent settings (always enabled - no flag needed)
     enable_token_exchange: bool = False
     enable_offline_access: bool = False
 
+    # Token exchange cache settings
+    token_exchange_cache_ttl: int = 300  # seconds (5 minutes default)
+
     # Token settings
     token_encryption_key: Optional[str] = None
     token_storage_db: Optional[str] = None
@@ -155,10 +168,18 @@ def get_settings() -> Settings:
         oidc_discovery_url=os.getenv("OIDC_DISCOVERY_URL"),
         oidc_client_id=os.getenv("OIDC_CLIENT_ID"),
         oidc_client_secret=os.getenv("OIDC_CLIENT_SECRET"),
+        oidc_issuer=os.getenv("OIDC_ISSUER"),
         # Nextcloud settings
         nextcloud_host=os.getenv("NEXTCLOUD_HOST"),
         nextcloud_username=os.getenv("NEXTCLOUD_USERNAME"),
         nextcloud_password=os.getenv("NEXTCLOUD_PASSWORD"),
+        # ADR-005: Token Audience Validation
+        nextcloud_mcp_server_url=os.getenv("NEXTCLOUD_MCP_SERVER_URL"),
+        nextcloud_resource_uri=os.getenv("NEXTCLOUD_RESOURCE_URI"),
+        # Token verification endpoints
+        jwks_uri=os.getenv("JWKS_URI"),
+        introspection_uri=os.getenv("INTROSPECTION_URI"),
+        userinfo_uri=os.getenv("USERINFO_URI"),
         # Progressive Consent settings (always enabled)
         enable_token_exchange=(
             os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true"
@@ -166,6 +187,8 @@ def get_settings() -> Settings:
         enable_offline_access=(
             os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true"
         ),
+        # Token exchange cache settings
+        token_exchange_cache_ttl=int(os.getenv("TOKEN_EXCHANGE_CACHE_TTL", "300")),
         # Token settings
         token_encryption_key=os.getenv("TOKEN_ENCRYPTION_KEY"),
         token_storage_db=os.getenv("TOKEN_STORAGE_DB", "/tmp/tokens.db"),

nextcloud_mcp_server/context.py

@@ -10,12 +10,15 @@ async def get_client(ctx: Context) -> NextcloudClient:
     """
     Get the appropriate Nextcloud client based on authentication mode.
 
-    This function handles three modes:
+    ADR-005 compliant implementation supporting two modes:
     1. BasicAuth mode: Returns shared client from lifespan context
-    2. OAuth pass-through mode (ENABLE_TOKEN_EXCHANGE=false, default):
-       Verifies Flow 1 token and passes it to Nextcloud
-    3. OAuth token exchange mode (ENABLE_TOKEN_EXCHANGE=true):
-       Exchanges Flow 1 token for ephemeral Nextcloud token via RFC 8693
+    2. Multi-audience mode (ENABLE_TOKEN_EXCHANGE=false, default):
+       Token already contains both MCP and Nextcloud audiences - use directly
+    3. Token exchange mode (ENABLE_TOKEN_EXCHANGE=true):
+       Exchange MCP token for Nextcloud token via RFC 8693
+
+    SECURITY: Token passthrough has been REMOVED. All OAuth modes validate
+    proper token audiences per MCP Security Best Practices specification.
 
     Note: Nextcloud doesn't support OAuth scopes natively. Scopes are enforced
     by the MCP server via @require_scopes decorator, not by the IdP.
@@ -49,20 +52,22 @@ async def get_client(ctx: Context) -> NextcloudClient:
 
     # OAuth mode (has 'nextcloud_host' attribute)
     if hasattr(lifespan_ctx, "nextcloud_host"):
-        # Check if token exchange is enabled
-        if settings.enable_token_exchange:
-            from nextcloud_mcp_server.auth.context_helper import (
-                get_session_client_from_context,
-            )
+        from nextcloud_mcp_server.auth.context_helper import (
+            get_client_from_context,
+            get_session_client_from_context,
+        )
 
-            # Token exchange mode: Exchange Flow 1 token for ephemeral Nextcloud token
+        if settings.enable_token_exchange:
+            # Mode 2: Exchange MCP token for Nextcloud token
+            # Token was validated to have MCP audience in UnifiedTokenVerifier
+            # Now exchange it for Nextcloud audience
             return await get_session_client_from_context(
                 ctx, lifespan_ctx.nextcloud_host
             )
         else:
-            # Pass-through mode (default): Verify and pass Flow 1 token to Nextcloud
-            from nextcloud_mcp_server.auth import get_client_from_context
-
+            # Mode 1: Multi-audience token - use directly
+            # Token was validated to have MCP audience in UnifiedTokenVerifier
+            # Nextcloud will independently validate its own audience when receiving API calls
             return get_client_from_context(ctx, lifespan_ctx.nextcloud_host)
 
     # Unknown context type

nextcloud_mcp_server/document_processors/tesseract.py

@@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
 try:
     import io
 
-    import pytesseract
+    import pytesseract  # type: ignore
     from PIL import Image
 
     TESSERACT_AVAILABLE = True

nextcloud_mcp_server/document_processors/unstructured.py

@@ -112,10 +112,10 @@ class UnstructuredProcessor(DocumentProcessor):
                         f"Processing document with unstructured... ({elapsed}s elapsed)"
                     )
                     try:
-                        await progress_callback(
-                            progress=float(elapsed),
-                            total=None,  # Unknown total duration
-                            message=message,
+                        await progress_callback(  # type: ignore
+                            progress=float(elapsed),  # type: ignore
+                            total=None,  # Unknown total duration  # type: ignore
+                            message=message,  # type: ignore
                         )
                         logger.debug(f"Progress update sent: {elapsed}s elapsed")
                     except Exception as e:
@@ -293,7 +293,7 @@ class UnstructuredProcessor(DocumentProcessor):
                 self._run_progress_poller, stop_event, progress_callback, start_time
             )
 
-        return result
+        return result  # type: ignore
 
     async def health_check(self) -> bool:
         """Check if Unstructured API is available.

nextcloud_mcp_server/server/cookbook.py

@@ -191,7 +191,7 @@ def configure_cookbook_tools(mcp: FastMCP):
         recipe_yield: int | None = None,
         category: str | None = None,
         keywords: str | None = None,
-        ctx: Context = None,
+        ctx: Context = None,  # type: ignore
     ) -> CreateRecipeResponse:
         """Create a new recipe.
 
@@ -271,7 +271,7 @@ def configure_cookbook_tools(mcp: FastMCP):
         recipe_yield: int | None = None,
         category: str | None = None,
         keywords: str | None = None,
-        ctx: Context = None,
+        ctx: Context = None,  # type: ignore
     ) -> UpdateRecipeResponse:
         """Update an existing recipe.
 
@@ -544,7 +544,7 @@ def configure_cookbook_tools(mcp: FastMCP):
         folder: str | None = None,
         update_interval: int | None = None,
         print_image: bool | None = None,
-        ctx: Context = None,
+        ctx: Context = None,  # type: ignore
     ) -> ReindexResponse:
         """Set Cookbook app configuration.
 

nextcloud_mcp_server/server/notes.py

@@ -331,7 +331,7 @@ def configure_notes_tools(mcp: FastMCP):
             content, mime_type = await client.webdav.get_note_attachment(
                 note_id=note_id, filename=attachment_filename
             )
-            return {
+            return {  # type: ignore
                 "uri": f"nc://Notes/{note_id}/attachments/{attachment_filename}",
                 "mimeType": mime_type,
                 "data": content,

nextcloud_mcp_server/server/oauth_tools.py

@@ -11,16 +11,88 @@ import secrets
 from typing import Optional
 from urllib.parse import urlencode
 
+import httpx
+from mcp.server.auth.middleware.auth_context import get_access_token
+from mcp.server.auth.provider import AccessToken
 from mcp.server.fastmcp import Context
 from pydantic import BaseModel, Field
 
 from nextcloud_mcp_server.auth import require_scopes
 from nextcloud_mcp_server.auth.refresh_token_storage import RefreshTokenStorage
 from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
+from nextcloud_mcp_server.auth.userinfo_routes import _query_idp_userinfo
 
 logger = logging.getLogger(__name__)
 
 
+async def extract_user_id_from_token(ctx: Context) -> str:
+    """Extract user_id from the MCP access token (Flow 1).
+
+    Handles both JWT and opaque tokens:
+    - JWT: Decode and extract 'sub' claim
+    - Opaque: Call userinfo endpoint to get 'sub'
+
+    Args:
+        ctx: MCP context with access token
+
+    Returns:
+        user_id extracted from token, or "default_user" as fallback
+    """
+    # Use MCP SDK's get_access_token() which uses contextvars
+    access_token: AccessToken | None = get_access_token()
+
+    if not access_token or not access_token.token:
+        logger.warning("  ✗ No access token found via get_access_token()")
+        return "default_user"
+
+    token = access_token.token
+    is_jwt = "." in token and token.count(".") >= 2
+    logger.info(f"  Token type: {'JWT' if is_jwt else 'Opaque'}")
+
+    # Try JWT decode first
+    if is_jwt:
+        try:
+            import jwt
+
+            payload = jwt.decode(token, options={"verify_signature": False})
+            user_id = payload.get("sub", "unknown")
+            logger.info(f"  ✓ JWT decode successful: user_id={user_id}")
+            return user_id
+        except Exception as e:
+            logger.error(f"  ✗ JWT decode failed: {type(e).__name__}: {e}")
+
+    # Opaque token - call userinfo endpoint
+    logger.info("  Opaque token detected, calling userinfo endpoint...")
+    try:
+        # Get userinfo endpoint from OIDC discovery
+        oidc_discovery_uri = os.getenv(
+            "OIDC_DISCOVERY_URI",
+            "http://localhost:8080/.well-known/openid-configuration",
+        )
+        async with httpx.AsyncClient() as http_client:
+            discovery_response = await http_client.get(oidc_discovery_uri)
+            discovery_response.raise_for_status()
+            discovery = discovery_response.json()
+            userinfo_endpoint = discovery.get("userinfo_endpoint")
+
+        if userinfo_endpoint:
+            userinfo = await _query_idp_userinfo(token, userinfo_endpoint)
+            if userinfo:
+                user_id = userinfo.get("sub", "unknown")
+                logger.info(f"  ✓ Userinfo query successful: user_id={user_id}")
+                return user_id
+            else:
+                logger.error("  ✗ Userinfo query failed")
+        else:
+            logger.error("  ✗ No userinfo_endpoint available")
+    except Exception as e:
+        logger.error(f"  ✗ Userinfo query failed: {type(e).__name__}: {e}")
+
+    # Fallback
+    logger.warning("  Using fallback user_id: default_user")
+    return "default_user"
+
+
 class ProvisioningStatus(BaseModel):
     """Status of Nextcloud provisioning for a user."""
 
@@ -57,6 +129,15 @@ class RevocationResult(BaseModel):
     message: str = Field(description="Status message for the user")
 
 
+class LoginConfirmation(BaseModel):
+    """Schema for login confirmation elicitation."""
+
+    acknowledged: bool = Field(
+        default=False,
+        description="Check this box after completing login at the provided URL",
+    )
+
+
 async def get_provisioning_status(ctx: Context, user_id: str) -> ProvisioningStatus:
     """
     Check the provisioning status for Nextcloud access.
@@ -71,14 +152,28 @@ async def get_provisioning_status(ctx: Context, user_id: str) -> ProvisioningSta
     Returns:
         ProvisioningStatus with current provisioning state
     """
+    logger.info(
+        f"  get_provisioning_status: Looking up refresh token for user_id={user_id}"
+    )
     storage = RefreshTokenStorage.from_env()
     await storage.initialize()
 
     token_data = await storage.get_refresh_token(user_id)
 
     if not token_data:
+        logger.info(
+            f"  get_provisioning_status: ✗ No refresh token found for user_id={user_id}"
+        )
         return ProvisioningStatus(is_provisioned=False)
 
+    logger.info(
+        f"  get_provisioning_status: ✓ Refresh token FOUND for user_id={user_id}"
+    )
+    logger.info(f"    flow_type: {token_data.get('flow_type')}")
+    logger.info(
+        f"    provisioning_client_id: {token_data.get('provisioning_client_id', 'N/A')}"
+    )
+
     # Convert timestamp to ISO format if present
     provisioned_at_str = None
     if token_data.get("provisioned_at"):
@@ -106,36 +201,33 @@ def generate_oauth_url_for_flow2(
     """
     Generate OAuth authorization URL for Flow 2 (Resource Provisioning).
 
-    This creates the URL that the MCP server uses to get delegated
-    access to Nextcloud on behalf of the user.
+    This returns the MCP server's Flow 2 authorization endpoint, which will:
+    1. Generate PKCE parameters (required by Nextcloud OIDC)
+    2. Store code_verifier in session
+    3. Redirect to Nextcloud IdP with PKCE
+    4. Handle the callback with code_verifier for token exchange
 
     Args:
-        oidc_discovery_url: OIDC provider discovery URL
-        server_client_id: MCP server's OAuth client ID
-        redirect_uri: Callback URL for the MCP server
+        oidc_discovery_url: OIDC provider discovery URL (unused, kept for compatibility)
+        server_client_id: MCP server's OAuth client ID (unused, kept for compatibility)
+        redirect_uri: Callback URL for the MCP server (unused, kept for compatibility)
         state: CSRF protection state
-        scopes: List of scopes to request
+        scopes: List of scopes to request (unused, kept for compatibility)
 
     Returns:
-        Complete authorization URL for Flow 2
+        MCP server's Flow 2 authorization URL with state parameter
     """
-    # Extract base URL from discovery URL
-    # Format: https://example.com/.well-known/openid-configuration
-    # We need: https://example.com/apps/oidc/authorize
-    base_url = oidc_discovery_url.replace("/.well-known/openid-configuration", "")
-    auth_endpoint = f"{base_url}/apps/oidc/authorize"
+    # Use the MCP server's Flow 2 endpoint which handles PKCE internally
+    # This endpoint will:
+    # - Generate code_verifier and code_challenge (PKCE)
+    # - Store code_verifier in session storage
+    # - Redirect to Nextcloud with PKCE parameters
+    # - Handle the callback with proper code_verifier
+    mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
+    auth_endpoint = f"{mcp_server_url}/oauth/authorize-nextcloud"
 
-    # Build OAuth parameters
-    params = {
-        "response_type": "code",
-        "client_id": server_client_id,
-        "redirect_uri": redirect_uri,
-        "scope": " ".join(scopes),
-        "state": state,
-        # Request offline access for background operations
-        "access_type": "offline",
-        "prompt": "consent",  # Force consent screen to show scopes
-    }
+    # Only pass state parameter - the endpoint handles everything else
+    params = {"state": state}
 
     return f"{auth_endpoint}?{urlencode(params)}"
 
@@ -163,7 +255,7 @@ async def provision_nextcloud_access(
         if not user_id:
             # Get the authorization token from context
             if hasattr(ctx, "authorization") and ctx.authorization:
-                token = ctx.authorization.token
+                token = ctx.authorization.token  # type: ignore
                 # Decode token to get user info
                 try:
                     import jwt
@@ -190,27 +282,33 @@ async def provision_nextcloud_access(
             )
 
         # Get configuration
-        enable_progressive = (
-            os.getenv("ENABLE_PROGRESSIVE_CONSENT", "false").lower() == "true"
+        enable_offline_access = (
+            os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true"
         )
-        if not enable_progressive:
+        if not enable_offline_access:
             return ProvisioningResult(
                 success=False,
                 message=(
-                    "Progressive Consent is not enabled. "
-                    "Set ENABLE_PROGRESSIVE_CONSENT=true to use this feature."
+                    "Offline access is not enabled. "
+                    "Set ENABLE_OFFLINE_ACCESS=true to use this feature."
                 ),
             )
 
         # Get MCP server's OAuth client credentials
+        # Try environment variable first, then fall back to DCR client_id
         server_client_id = os.getenv("MCP_SERVER_CLIENT_ID")
         if not server_client_id:
-            # In production, would use Dynamic Client Registration here
+            # Try to get from lifespan context (DCR)
+            lifespan_ctx = ctx.request_context.lifespan_context
+            if hasattr(lifespan_ctx, "server_client_id"):
+                server_client_id = lifespan_ctx.server_client_id
+
+        if not server_client_id:
             return ProvisioningResult(
                 success=False,
                 message=(
                     "MCP server OAuth client not configured. "
-                    "Administrator must set MCP_SERVER_CLIENT_ID."
+                    "Set MCP_SERVER_CLIENT_ID environment variable or use Dynamic Client Registration."
                 ),
             )
 
@@ -229,7 +327,7 @@ async def provision_nextcloud_access(
 
         # Create OAuth session for Flow 2
         session_id = f"flow2_{user_id}_{secrets.token_hex(8)}"
-        redirect_uri = f"{os.getenv('NEXTCLOUD_MCP_SERVER_URL', 'http://localhost:8000')}/oauth/callback-nextcloud"
+        redirect_uri = f"{os.getenv('NEXTCLOUD_MCP_SERVER_URL', 'http://localhost:8000')}/oauth/callback"
 
         await storage.store_oauth_session(
             session_id=session_id,
@@ -301,13 +399,11 @@ async def revoke_nextcloud_access(
         RevocationResult with status
     """
     try:
-        # Get user ID from context if not provided
+        # Get user ID from token if not provided
         if not user_id:
-            user_id = (
-                ctx.context.get("user_id", "default_user")
-                if hasattr(ctx, "context")
-                else "default_user"
-            )
+            logger.info("Extracting user_id from access token for revoke...")
+            user_id = await extract_user_id_from_token(ctx)
+            logger.info(f"  Revoke using user_id: {user_id}")
 
         # Check current status
         status = await get_provisioning_status(ctx, user_id)
@@ -334,7 +430,7 @@ async def revoke_nextcloud_access(
                 "OIDC_DISCOVERY_URL",
                 f"{os.getenv('NEXTCLOUD_HOST')}/.well-known/openid-configuration",
             ),
-            nextcloud_host=os.getenv("NEXTCLOUD_HOST"),
+            nextcloud_host=os.getenv("NEXTCLOUD_HOST"),  # type: ignore
             encryption_key=encryption_key,
         )
 
@@ -382,7 +478,7 @@ async def check_provisioning_status(
     # Get user ID from context if not provided
     if not user_id:
         user_id = (
-            ctx.context.get("user_id", "default_user")
+            ctx.context.get("user_id", "default_user")  # type: ignore
             if hasattr(ctx, "context")
             else "default_user"
         )
@@ -390,6 +486,198 @@ async def check_provisioning_status(
     return await get_provisioning_status(ctx, user_id)
 
 
+async def check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str:
+    """
+    MCP Tool: Check if user is logged in and elicit login if needed.
+
+    This tool checks whether the user has completed Flow 2 (resource provisioning)
+    to grant offline access to Nextcloud. If not logged in, it uses MCP elicitation
+    to prompt the user to complete the login flow.
+
+    Args:
+        ctx: MCP context with user's Flow 1 token
+        user_id: Optional user identifier (extracted from token if not provided)
+
+    Returns:
+        "yes" if logged in, or elicitation prompting for login
+    """
+    try:
+        # Extract user ID from the MCP access token (Flow 1 token)
+        logger.info("=" * 60)
+        logger.info("check_logged_in: Starting user_id extraction")
+        logger.info("=" * 60)
+
+        if not user_id:
+            user_id = await extract_user_id_from_token(ctx)
+            logger.info(f"  Final user_id for check_logged_in: {user_id}")
+        else:
+            logger.info(f"  user_id provided as argument: {user_id}")
+
+        # Check if already logged in
+        logger.info(f"Checking provisioning status for user_id: {user_id}")
+        status = await get_provisioning_status(ctx, user_id)
+        logger.info(f"  Provisioning status: is_provisioned={status.is_provisioned}")
+
+        if status.is_provisioned:
+            logger.info(f"✓ User {user_id} is already logged in - returning 'yes'")
+            logger.info("=" * 60)
+            return "yes"
+
+        logger.info(f"✗ User {user_id} is NOT logged in - triggering elicitation")
+        logger.info("=" * 60)
+
+        # Not logged in - generate OAuth URL for Flow 2
+        enable_offline_access = (
+            os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true"
+        )
+        if not enable_offline_access:
+            return (
+                "Not logged in. Offline access is not enabled. "
+                "Set ENABLE_OFFLINE_ACCESS=true to use this feature."
+            )
+
+        # Get MCP server's OAuth client credentials
+        # Try environment variable first, then fall back to DCR client_id
+        server_client_id = os.getenv("MCP_SERVER_CLIENT_ID")
+        if not server_client_id:
+            # Try to get from lifespan context (DCR)
+            lifespan_ctx = ctx.request_context.lifespan_context
+            if hasattr(lifespan_ctx, "server_client_id"):
+                server_client_id = lifespan_ctx.server_client_id
+
+        if not server_client_id:
+            return (
+                "Not logged in. MCP server OAuth client not configured. "
+                "Set MCP_SERVER_CLIENT_ID environment variable or use Dynamic Client Registration."
+            )
+
+        # Generate OAuth URL for Flow 2
+        oidc_discovery_url = os.getenv(
+            "OIDC_DISCOVERY_URL",
+            f"{os.getenv('NEXTCLOUD_HOST')}/.well-known/openid-configuration",
+        )
+
+        # Generate secure state for CSRF protection
+        state = secrets.token_urlsafe(32)
+
+        # Store state in session for validation on callback
+        storage = RefreshTokenStorage.from_env()
+        await storage.initialize()
+
+        # Create OAuth session for Flow 2
+        session_id = f"flow2_{user_id}_{secrets.token_hex(8)}"
+        redirect_uri = f"{os.getenv('NEXTCLOUD_MCP_SERVER_URL', 'http://localhost:8000')}/oauth/callback"
+
+        await storage.store_oauth_session(
+            session_id=session_id,
+            client_redirect_uri="",  # No client redirect for Flow 2
+            state=state,
+            flow_type="flow2",
+            is_provisioning=True,
+            ttl_seconds=600,  # 10 minute TTL
+        )
+
+        # Define scopes for Nextcloud access
+        scopes = [
+            "openid",
+            "profile",
+            "email",
+            "offline_access",  # Critical for background operations
+            "notes:read",
+            "notes:write",
+            "calendar:read",
+            "calendar:write",
+            "contacts:read",
+            "contacts:write",
+            "files:read",
+            "files:write",
+        ]
+
+        # Generate authorization URL
+        auth_url = generate_oauth_url_for_flow2(
+            oidc_discovery_url=oidc_discovery_url,
+            server_client_id=server_client_id,
+            redirect_uri=redirect_uri,
+            state=state,
+            scopes=scopes,
+        )
+
+        # Use elicitation to prompt user to login
+        logger.info(f"Eliciting login for user {user_id} with URL: {auth_url}")
+
+        result = await ctx.elicit(
+            message=f"Please log in to Nextcloud at the following URL:\n\n{auth_url}\n\nAfter completing the login, check the box below and click OK.",
+            schema=LoginConfirmation,
+        )
+
+        if result.action == "accept":
+            # Check if login was successful by looking for refresh token
+            # Strategy: Try multiple lookup methods to handle both flows
+            logger.info("User accepted login prompt, checking for refresh token")
+            logger.info(f"  State parameter: {state[:16]}...")
+            logger.info(f"  User ID: {user_id}")
+
+            # First, try to find token by provisioning_client_id (Flow 2 from elicitation)
+            refresh_token_data = (
+                await storage.get_refresh_token_by_provisioning_client_id(state)
+            )
+
+            if refresh_token_data:
+                logger.info("✓ Refresh token found via provisioning_client_id lookup")
+                logger.info(
+                    f"  Flow type: {refresh_token_data.get('flow_type', 'unknown')}"
+                )
+                logger.info(
+                    f"  Provisioned at: {refresh_token_data.get('provisioned_at', 'unknown')}"
+                )
+                return "yes"
+
+            # Fallback: Try to find token by user_id (browser login or any other flow)
+            logger.info(f"✗ No token found with provisioning_client_id={state[:16]}...")
+            logger.info(f"  Trying fallback lookup by user_id: {user_id}")
+
+            refresh_token_data = await storage.get_refresh_token(user_id)
+
+            if refresh_token_data:
+                logger.info("✓ Refresh token found via user_id lookup")
+                logger.info(
+                    f"  Flow type: {refresh_token_data.get('flow_type', 'unknown')}"
+                )
+                logger.info(
+                    f"  Provisioned at: {refresh_token_data.get('provisioned_at', 'unknown')}"
+                )
+                logger.info(
+                    f"  Provisioning client ID: {refresh_token_data.get('provisioning_client_id', 'NULL')}"
+                )
+                logger.info(
+                    "  Note: This token was created via browser login or different flow"
+                )
+                return "yes"
+
+            # No token found by either method
+            logger.warning(f"✗ No refresh token found for user {user_id}")
+            logger.warning(
+                f"  Checked provisioning_client_id={state[:16]}... - NOT FOUND"
+            )
+            logger.warning(f"  Checked user_id={user_id} - NOT FOUND")
+            logger.warning(
+                "  This may indicate the user completed login but token wasn't stored"
+            )
+
+            return (
+                "Login not detected. Please ensure you completed the login "
+                "at the provided URL before clicking OK."
+            )
+        elif result.action == "decline":
+            return "Login declined by user."
+        else:
+            return "Login cancelled by user."
+
+    except Exception as e:
+        logger.error(f"Failed to check login status: {e}")
+        return f"Error checking login status: {str(e)}"
+
+
 # Register MCP tools
 def register_oauth_tools(mcp):
     """Register OAuth and provisioning tools with the MCP server."""
@@ -428,3 +716,14 @@ def register_oauth_tools(mcp):
         ctx: Context, user_id: Optional[str] = None
     ) -> ProvisioningStatus:
         return await check_provisioning_status(ctx, user_id)
+
+    @mcp.tool(
+        name="check_logged_in",
+        description=(
+            "Check if you are logged in to Nextcloud. "
+            "If not logged in, this tool will prompt you to complete the login flow."
+        ),
+    )
+    @require_scopes("openid")
+    async def tool_check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str:
+        return await check_logged_in(ctx, user_id)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nextcloud-mcp-server"
-version = "0.23.0"
+version = "0.26.0"
 description = "Model Context Protocol (MCP) server for Nextcloud integration - enables AI assistants to interact with Nextcloud data"
 authors = [
     {name = "Chris Coutinho", email = "chris@coutinho.io"}
@@ -10,7 +10,7 @@ license = {text = "AGPL-3.0-only"}
 requires-python = ">=3.11"
 keywords = ["nextcloud", "mcp", "model-context-protocol", "llm", "ai", "claude", "webdav", "caldav", "carddav"]
 dependencies = [
-    "mcp[cli] (>=1.19,<1.20)",
+    "mcp[cli] (>=1.20,<1.21)",
     "httpx (>=0.28.1,<0.29.0)",
     "pillow (>=12.0.0,<12.1.0)",
     "icalendar (>=6.0.0,<7.0.0)",
@@ -102,6 +102,7 @@ dev = [
     "pytest-timeout>=2.3.1",
     "ruff>=0.11.13",
     "reportlab>=4.0.0",
+    "ty>=0.0.1a25",
 ]
 
 [project.scripts]

tests/conftest.py

@@ -8,7 +8,9 @@ import httpx
 import pytest
 from httpx import HTTPStatusError
 from mcp import ClientSession
+from mcp.client.session import RequestContext
 from mcp.client.streamable_http import streamablehttp_client
+from mcp.types import ElicitRequestParams, ElicitResult, ErrorData
 
 from nextcloud_mcp_server.client import NextcloudClient
 
@@ -110,6 +112,7 @@ async def create_mcp_client_session(
     url: str,
     token: str | None = None,
     client_name: str = "MCP",
+    elicitation_callback: Any = None,
 ) -> AsyncGenerator[ClientSession, Any]:
     """
     Factory function to create an MCP client session with proper lifecycle management.
@@ -127,6 +130,8 @@ async def create_mcp_client_session(
         url: MCP server URL (e.g., "http://localhost:8000/mcp")
         token: Optional OAuth access token for Bearer authentication
         client_name: Client name for logging (e.g., "OAuth MCP (Playwright)")
+        elicitation_callback: Optional callback for handling elicitation requests.
+            Should match signature: async def callback(context: RequestContext, params: ElicitRequestParams) -> ElicitResult | ErrorData
 
     Yields:
         Initialized MCP ClientSession
@@ -149,7 +154,9 @@ async def create_mcp_client_session(
         write_stream,
         _,
     ):
-        async with ClientSession(read_stream, write_stream) as session:
+        async with ClientSession(
+            read_stream, write_stream, elicitation_callback=elicitation_callback
+        ) as session:
             await session.initialize()
             logger.info(f"{client_name} client session initialized successfully")
             yield session
@@ -251,6 +258,163 @@ async def nc_mcp_oauth_jwt_client(
         yield session
 
 
+@pytest.fixture
+async def nc_mcp_oauth_client_with_elicitation(
+    anyio_backend,
+    playwright_oauth_token: str,
+    browser,
+) -> AsyncGenerator[ClientSession, Any]:
+    """
+    Fixture to create an MCP client session with elicitation callback support.
+
+    This fixture enables REAL elicitation testing by providing a callback that:
+    1. Extracts OAuth URL from elicitation message
+    2. Uses Playwright to complete OAuth flow automatically
+    3. Returns acceptance to confirm completion
+
+    This allows testing the complete login elicitation flow (ADR-006) end-to-end,
+    verifying that:
+    - The check_logged_in tool triggers elicitation for unauthenticated users
+    - The OAuth flow completes successfully via automated browser
+    - Refresh token is stored after OAuth completion
+    - The tool returns "yes" after successful login
+
+    Uses function scope to allow each test to have independent elicitation state.
+    """
+    # Get credentials from environment
+    username = os.getenv("NEXTCLOUD_USERNAME")
+    password = os.getenv("NEXTCLOUD_PASSWORD")
+
+    if not all([username, password]):
+        pytest.skip(
+            "Elicitation test requires NEXTCLOUD_USERNAME and NEXTCLOUD_PASSWORD"
+        )
+
+    # Track whether elicitation was triggered (for test validation)
+    elicitation_triggered = {"count": 0}
+
+    async def elicitation_callback(
+        context: RequestContext[ClientSession, Any],
+        params: ElicitRequestParams,
+    ) -> ElicitResult | ErrorData:
+        """Handle elicitation by completing OAuth flow with Playwright."""
+        elicitation_triggered["count"] += 1
+
+        logger.info("🎯 Elicitation callback invoked!")
+        logger.info(f"  Message: {params.message[:100]}...")
+        logger.info(f"  Schema: {params.schema}")
+
+        # Extract OAuth URL from elicitation message
+        import re
+
+        url_pattern = r"https?://[^\s]+"
+        urls = re.findall(url_pattern, params.message)
+
+        if not urls:
+            error_msg = "No URL found in elicitation message"
+            logger.error(f"❌ {error_msg}")
+            return ErrorData(code=-32602, message=error_msg)
+
+        oauth_url = urls[0]
+        logger.info(f"  Extracted URL: {oauth_url}")
+
+        # Complete OAuth flow with Playwright
+        page = await browser.new_page()
+        try:
+            logger.info("🌐 Navigating to OAuth URL...")
+            await page.goto(oauth_url, timeout=60000)
+
+            current_url = page.url
+            logger.info(f"  Current URL after navigation: {current_url}")
+
+            # Handle login form if present
+            if "/login" in current_url or "/index.php/login" in current_url:
+                logger.info("🔐 Login page detected, filling credentials...")
+                await page.wait_for_selector('input[name="user"]', timeout=10000)
+                await page.fill('input[name="user"]', username)
+                await page.fill('input[name="password"]', password)
+                await page.click('button[type="submit"]')
+                await page.wait_for_load_state("networkidle", timeout=60000)
+                logger.info("  ✓ Login completed")
+
+            # Handle consent screen if present
+            try:
+                logger.info(f"  Current URL before consent: {page.url}")
+                consent_handled = await _handle_oauth_consent_screen(page, username)
+                if consent_handled:
+                    logger.info("  ✓ Consent granted")
+                else:
+                    logger.warning("  ⚠ No consent screen detected")
+                    # Take screenshot for debugging
+                    screenshot_path = f"/tmp/elicitation_no_consent_{uuid.uuid4()}.png"
+                    await page.screenshot(path=screenshot_path)
+                    logger.info(f"  Screenshot saved: {screenshot_path}")
+                    # Log page title for debugging
+                    page_title = await page.title()
+                    logger.info(f"  Page title: {page_title}")
+            except Exception as e:
+                logger.warning(f"  ⚠ Consent screen handling failed: {e}")
+                # Take screenshot for debugging
+                screenshot_path = f"/tmp/elicitation_consent_error_{uuid.uuid4()}.png"
+                await page.screenshot(path=screenshot_path)
+                logger.info(f"  Screenshot saved: {screenshot_path}")
+
+            # Wait for OAuth callback URL to be reached
+            # The MCP server's callback endpoint will handle token exchange
+            logger.info("⏳ Waiting for OAuth callback to complete...")
+
+            # Wait for URL to contain /oauth/callback or a success page
+            # Give it up to 30 seconds for the redirect and token exchange
+            for _ in range(60):  # 60 * 0.5s = 30s max wait
+                await anyio.sleep(0.5)
+                current_url = page.url
+                if "/oauth/callback" in current_url or "/user" in current_url:
+                    logger.info(f"  ✓ Callback URL reached: {current_url}")
+                    break
+            else:
+                logger.warning(
+                    f"  ⚠ Timeout waiting for callback, final URL: {page.url}"
+                )
+
+            # Wait a bit more to ensure the server processed the callback
+            await anyio.sleep(2)
+
+            final_url = page.url
+            logger.info(f"  Final URL: {final_url}")
+
+            # Return success - user "accepted" the elicitation
+            logger.info("✅ OAuth flow completed, returning accept")
+            return ElicitResult(action="accept", content={"acknowledged": True})
+
+        except Exception as e:
+            logger.error(f"❌ Elicitation OAuth flow failed: {e}")
+            # Take screenshot for debugging
+            try:
+                screenshot_path = f"/tmp/elicitation_oauth_failure_{uuid.uuid4()}.png"
+                await page.screenshot(path=screenshot_path)
+                logger.error(f"  Screenshot saved: {screenshot_path}")
+            except Exception:
+                pass
+
+            return ErrorData(
+                code=-32603, message=f"Failed to complete OAuth flow: {str(e)}"
+            )
+
+        finally:
+            await page.close()
+
+    # Create client session with elicitation callback
+    async for session in create_mcp_client_session(
+        url="http://localhost:8001/mcp",
+        token=playwright_oauth_token,
+        client_name="OAuth MCP with Elicitation",
+        elicitation_callback=elicitation_callback,
+    ):
+        # Attach elicitation metadata for test validation
+        session.elicitation_triggered = elicitation_triggered
+        yield session
+
+
 @pytest.fixture(scope="session")
 async def nc_mcp_oauth_client_read_only(
     anyio_backend,
@@ -2193,17 +2357,35 @@ async def _get_oauth_token_for_user(
     logger.info(f"Getting OAuth token for user: {username}...")
     logger.info(f"Using shared OAuth client: {client_id[:16]}...")
 
+    # Fetch resource identifier from PRM endpoint (RFC 9728)
+    mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8001")
+    prm_url = f"{mcp_server_url}/.well-known/oauth-protected-resource"
+
+    logger.debug(f"Fetching PRM metadata from: {prm_url}")
+    async with httpx.AsyncClient() as client:
+        prm_response = await client.get(prm_url, timeout=10)
+        if prm_response.status_code != 200:
+            logger.warning(f"Failed to fetch PRM metadata: {prm_response.status_code}")
+            # Fallback to default if PRM fetch fails
+            mcp_server_resource = f"{mcp_server_url}/mcp"
+        else:
+            prm_data = prm_response.json()
+            mcp_server_resource = prm_data.get("resource", f"{mcp_server_url}/mcp")
+            logger.info(f"Using resource from PRM: {mcp_server_resource}")
+
     # Generate unique state parameter for this OAuth flow
     state = secrets.token_urlsafe(32)
     logger.debug(f"Generated state for {username}: {state[:16]}...")
 
     # Construct authorization URL with state parameter
+    # Include resource parameter discovered from PRM endpoint
     auth_url = (
         f"{authorization_endpoint}?"
         f"response_type=code&"
         f"client_id={client_id}&"
         f"redirect_uri={quote(callback_url, safe='')}&"
         f"state={state}&"
+        f"resource={quote(mcp_server_resource, safe='')}&"  # Resource URI from PRM
         f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
     )
 

tests/server/oauth/test_elicitation_integration.py

@@ -0,0 +1,206 @@
+"""Integration tests for login elicitation with real MCP client callback support.
+
+These tests verify the complete end-to-end login elicitation flow (ADR-006)
+using the python-sdk MCP client with actual elicitation callback implementation.
+
+Unlike test_login_elicitation.py which validates response formats, these tests
+exercise the REAL elicitation protocol:
+1. MCP client with elicitation callback connects to server
+2. Tool triggers elicitation (ctx.elicit())
+3. Client callback receives elicitation request
+4. Callback completes OAuth flow via Playwright automation
+5. Client returns acceptance
+6. Tool proceeds with authenticated operation
+
+This validates that:
+- python-sdk MCP client can handle elicitation requests
+- OAuth flow completion via callback works end-to-end
+- Refresh tokens are properly stored after elicitation
+- check_logged_in returns "yes" after successful OAuth
+"""
+
+import logging
+
+import pytest
+
+logger = logging.getLogger(__name__)
+
+pytestmark = [pytest.mark.integration, pytest.mark.oauth]
+
+
+async def revoke_refresh_tokens(client):
+    """Helper to revoke all refresh tokens from MCP server.
+
+    This forces check_logged_in to trigger elicitation by removing
+    any existing refresh tokens via the revoke_nextcloud_access tool.
+    """
+    logger.info("Revoking refresh tokens via revoke_nextcloud_access tool...")
+
+    result = await client.call_tool("revoke_nextcloud_access", arguments={})
+
+    logger.info(f"Revoke result: isError={result.isError}")
+    if not result.isError:
+        logger.info(f"✓ Revoke response: {result.content[0].text}")
+    else:
+        logger.warning(f"Revoke failed: {result.content}")
+
+
+async def test_check_logged_in_with_real_elicitation_callback(
+    nc_mcp_oauth_client_with_elicitation,
+):
+    """Test check_logged_in with actual elicitation callback that completes OAuth.
+
+    This test validates the COMPLETE elicitation flow:
+    1. Call check_logged_in tool (which triggers elicitation)
+    2. Elicitation callback extracts OAuth URL
+    3. Playwright automation completes OAuth flow
+    4. Callback returns acceptance
+    5. Tool returns "yes" (logged in)
+    6. Refresh token is stored
+
+    This is the ONLY test that exercises the real MCP elicitation protocol
+    with python-sdk's ClientSession elicitation callback support.
+    """
+    client = nc_mcp_oauth_client_with_elicitation
+
+    logger.info("=" * 80)
+    logger.info("TEST: Real elicitation callback with OAuth completion")
+    logger.info("=" * 80)
+
+    # Revoke refresh tokens to force elicitation
+    await revoke_refresh_tokens(client)
+
+    # Call check_logged_in - this should trigger elicitation
+    logger.info("Calling check_logged_in tool...")
+    result = await client.call_tool("check_logged_in", arguments={})
+
+    logger.info("Tool execution completed")
+    logger.info(f"  Is error: {result.isError}")
+    if result.content:
+        response_text = result.content[0].text
+        logger.info(f"  Response: {response_text}")
+    else:
+        logger.warning("  No content in response")
+
+    # Validate tool execution succeeded
+    assert result.isError is False, f"Tool execution failed: {result.content}"
+    assert result.content is not None, "No content in tool response"
+
+    response_text = result.content[0].text.lower()
+
+    # Validate elicitation was triggered
+    elicitation_count = client.elicitation_triggered["count"]
+    logger.info(f"✓ Elicitation triggered {elicitation_count} time(s)")
+    assert elicitation_count >= 1, (
+        "Elicitation callback should have been invoked at least once"
+    )
+
+    # Validate OAuth completed successfully and tool returned "yes"
+    assert "yes" in response_text, (
+        f"Expected 'yes' after successful OAuth via elicitation, got: {response_text}"
+    )
+
+    logger.info("✅ Test passed: Real elicitation callback completed OAuth flow")
+    logger.info("=" * 80)
+
+
+async def test_elicitation_callback_url_extraction(
+    nc_mcp_oauth_client_with_elicitation,
+):
+    """Test that elicitation callback correctly extracts OAuth URL.
+
+    This validates the URL extraction logic in the callback by examining
+    the elicitation message format returned by check_logged_in.
+    """
+    client = nc_mcp_oauth_client_with_elicitation
+
+    logger.info("Testing OAuth URL extraction from elicitation message...")
+
+    # Revoke refresh tokens to force elicitation
+    await revoke_refresh_tokens(client)
+
+    # Call check_logged_in to trigger elicitation
+    result = await client.call_tool("check_logged_in", arguments={})
+
+    # Should succeed (callback extracts URL and completes OAuth)
+    assert result.isError is False
+    assert "yes" in result.content[0].text.lower()
+
+    # Elicitation should have been triggered
+    assert client.elicitation_triggered["count"] >= 1
+
+    logger.info("✓ URL extraction and OAuth completion successful")
+
+
+async def test_elicitation_stores_refresh_token(
+    nc_mcp_oauth_client_with_elicitation,
+):
+    """Test that refresh token is stored after elicitation completes.
+
+    Validates that after successful OAuth via elicitation:
+    1. check_logged_in returns "yes"
+    2. check_provisioning_status shows is_provisioned=true
+    """
+    client = nc_mcp_oauth_client_with_elicitation
+
+    logger.info("Testing refresh token storage after elicitation...")
+
+    # Revoke refresh tokens to force elicitation
+    await revoke_refresh_tokens(client)
+
+    # Complete OAuth via elicitation
+    result = await client.call_tool("check_logged_in", arguments={})
+    assert result.isError is False
+    assert "yes" in result.content[0].text.lower()
+
+    # Verify refresh token was stored
+    logger.info("Checking provisioning status...")
+    status_result = await client.call_tool("check_provisioning_status", arguments={})
+
+    assert status_result.isError is False
+    status_text = status_result.content[0].text.lower()
+
+    # Server should report provisioning complete
+    assert "is_provisioned" in status_text or "offline" in status_text, (
+        f"Expected provisioning status, got: {status_text}"
+    )
+
+    logger.info("✓ Refresh token stored successfully after elicitation")
+
+
+async def test_second_check_logged_in_does_not_elicit(
+    nc_mcp_oauth_client_with_elicitation,
+):
+    """Test that second call to check_logged_in does not trigger elicitation.
+
+    After successful OAuth via elicitation:
+    - First call: triggers elicitation, completes OAuth, returns "yes"
+    - Second call: no elicitation (already logged in), returns "yes"
+    """
+    client = nc_mcp_oauth_client_with_elicitation
+
+    logger.info("Testing that already-logged-in users don't get elicited...")
+
+    # First call: triggers elicitation
+    result1 = await client.call_tool("check_logged_in", arguments={})
+    assert result1.isError is False
+    assert "yes" in result1.content[0].text.lower()
+
+    elicitation_count_after_first = client.elicitation_triggered["count"]
+    logger.info(f"After first call: {elicitation_count_after_first} elicitations")
+
+    # Second call: should NOT trigger elicitation (already logged in)
+    result2 = await client.call_tool("check_logged_in", arguments={})
+    assert result2.isError is False
+    assert "yes" in result2.content[0].text.lower()
+
+    elicitation_count_after_second = client.elicitation_triggered["count"]
+    logger.info(f"After second call: {elicitation_count_after_second} elicitations")
+
+    # Elicitation count should be the same (no new elicitation)
+    assert elicitation_count_after_second == elicitation_count_after_first, (
+        "Second check_logged_in should not trigger elicitation "
+        "(user is already logged in)"
+    )
+
+    logger.info("✓ Already-logged-in users don't get redundant elicitations")

tests/server/oauth/test_login_elicitation.py

@@ -0,0 +1,246 @@
+"""Integration tests for login elicitation flow (ADR-006 Interim Implementation).
+
+Tests verify:
+1. check_logged_in tool with elicitation for unauthenticated users
+2. Elicitation contains login URL in message
+3. User can complete login via OAuth
+4. After login, check_logged_in returns "yes"
+5. Already-authenticated users get immediate "yes" response
+6. Elicitation decline/cancel handling
+"""
+
+import logging
+import re
+
+import pytest
+
+logger = logging.getLogger(__name__)
+
+pytestmark = [pytest.mark.integration, pytest.mark.oauth]
+
+
+async def test_check_logged_in_elicitation_flow(
+    nc_mcp_oauth_client, browser, oauth_callback_server
+):
+    """Test that check_logged_in elicits login for unauthenticated user.
+
+    This test validates the complete elicitation flow:
+    1. Call check_logged_in on authenticated client (already has refresh token)
+    2. Verify tool returns "yes" without elicitation
+    3. Extract and validate the elicitation URL format from response
+    4. Verify refresh token exists after successful OAuth flow
+
+    Note: Actual elicitation handling requires MCP protocol support in the test client.
+    This test validates the response format and token storage.
+    """
+    # Call check_logged_in tool on authenticated client
+    logger.info("Calling check_logged_in on authenticated client")
+    result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={})
+
+    assert result.isError is False, f"Tool execution failed: {result.content}"
+    assert result.content is not None
+
+    response_text = result.content[0].text
+    logger.info(f"check_logged_in response: {response_text}")
+
+    # Since nc_mcp_oauth_client fixture already completes OAuth during setup,
+    # the user should already be provisioned and we expect "yes"
+    # For unauthenticated users, the response would contain an elicitation URL
+    # Note: Test framework may return "elicitation not supported" if MCP elicitation is unavailable
+    assert (
+        "yes" in response_text.lower()
+        or "http" in response_text.lower()
+        or "elicitation not supported" in response_text.lower()
+    ), f"Unexpected response: {response_text}"
+
+    # If response contains a URL (elicitation case), validate its format
+    if "http" in response_text:
+        url_pattern = r"https?://[^\s]+"
+        urls = re.findall(url_pattern, response_text)
+        assert len(urls) > 0, "Expected elicitation URL in response"
+
+        login_url = urls[0]
+        logger.info(f"Elicitation URL: {login_url}")
+
+        # Validate URL points to MCP server's Flow 2 endpoint
+        assert "/oauth/authorize-nextcloud" in login_url, (
+            f"Expected URL to point to MCP server Flow 2 endpoint, got: {login_url}"
+        )
+        # Validate URL contains state parameter
+        assert "state=" in login_url, "Expected state parameter in elicitation URL"
+    elif "elicitation not supported" in response_text.lower():
+        logger.info(
+            "✓ Test client doesn't support elicitation - this is expected in test environment"
+        )
+
+
+async def test_check_logged_in_already_authenticated(nc_mcp_oauth_client):
+    """Test that check_logged_in returns 'yes' for authenticated user.
+
+    This test verifies that if the user has already completed Flow 2
+    (resource provisioning), the tool immediately returns "yes" without
+    elicitation.
+    """
+    logger.info("Calling check_logged_in on authenticated client")
+
+    # Since we're using the nc_mcp_oauth_client fixture which completes
+    # OAuth during setup, the user should already be provisioned
+    result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={})
+
+    assert result.isError is False, f"Tool execution failed: {result.content}"
+    assert result.content is not None
+
+    response_text = result.content[0].text
+    logger.info(f"Response: {response_text}")
+
+    # Check for valid responses:
+    # - "yes" (already logged in)
+    # - "not enabled" (offline access not enabled)
+    # - "not configured" (MCP_SERVER_CLIENT_ID not set)
+    # - "elicitation not supported" (test environment limitation)
+    assert (
+        "yes" in response_text.lower()
+        or "not enabled" in response_text.lower()
+        or "not configured" in response_text.lower()
+        or "elicitation not supported" in response_text.lower()
+    )
+
+
+async def test_check_logged_in_url_format(nc_mcp_oauth_client):
+    """Test that login URL (when needed) follows correct OAuth format.
+
+    This test verifies that if the tool needs to provide a login URL,
+    the URL contains the correct OAuth parameters for Flow 2.
+    """
+    # Call the tool
+    result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={})
+
+    assert result.isError is False, f"Tool execution failed: {result.content}"
+    assert result.content is not None
+
+    response_text = result.content[0].text
+    logger.info(f"Response: {response_text}")
+
+    # If response contains a URL, validate it
+    url_pattern = r"https?://[^\s]+"
+    urls = re.findall(url_pattern, response_text)
+
+    if urls:
+        login_url = urls[0]
+        logger.info(f"Found login URL: {login_url}")
+
+        # Validate OAuth parameters
+        assert "response_type=code" in login_url
+        assert "client_id=" in login_url
+        assert "redirect_uri=" in login_url
+        assert "scope=" in login_url
+        assert "state=" in login_url
+        assert "openid" in login_url  # Should request openid scope
+
+        # Validate callback URL (unified endpoint without query params)
+        # Note: redirect_uri should be /oauth/callback (no query params)
+        # Flow type is determined by session lookup, not URL params
+        assert (
+            "/oauth/callback" in login_url
+            or "callback-nextcloud" in login_url  # Legacy support
+            or "authorize-nextcloud" in login_url
+        )
+
+
+async def test_check_logged_in_with_user_id(nc_mcp_oauth_client):
+    """Test that check_logged_in accepts optional user_id parameter.
+
+    This verifies the tool can be called with an explicit user_id.
+    """
+    result = await nc_mcp_oauth_client.call_tool(
+        "check_logged_in", arguments={"user_id": "testuser"}
+    )
+
+    assert result.isError is False, f"Tool execution failed: {result.content}"
+    assert result.content is not None
+
+    response_text = result.content[0].text
+    logger.info(f"Response with user_id: {response_text}")
+
+    # Should get some response (either yes or not logged in)
+    assert len(response_text) > 0
+
+
+async def test_check_logged_in_tool_metadata(nc_mcp_oauth_client):
+    """Test that check_logged_in tool has correct metadata."""
+    tools = await nc_mcp_oauth_client.list_tools()
+    assert tools is not None
+
+    # Find the check_logged_in tool
+    check_logged_in_tool = None
+    for tool in tools.tools:
+        if tool.name == "check_logged_in":
+            check_logged_in_tool = tool
+            break
+
+    assert check_logged_in_tool is not None, "check_logged_in tool not found"
+    logger.info(f"Tool: {check_logged_in_tool.name}")
+    logger.info(f"Description: {check_logged_in_tool.description}")
+
+    # Verify description mentions login
+    assert "login" in check_logged_in_tool.description.lower()
+
+    # Tool should have openid scope requirement
+    # (This would need to be verified via tool schema if exposed)
+
+
+async def test_elicitation_url_and_refresh_token_flow(nc_mcp_oauth_client):
+    """Test that MCP server validates refresh tokens after OAuth completion.
+
+    This test validates the server's refresh token handling through its API:
+    1. Call check_provisioning_status to verify server-side token validation
+    2. Server responses indicate token state:
+       - is_provisioned=True: Server has valid refresh token
+       - is_provisioned=False: No token or invalid token
+       - Error response: Token validation failed
+
+    The test does NOT directly access refresh token storage - it relies on
+    the MCP server to validate tokens internally and report status via API.
+    """
+    logger.info("Testing server-side refresh token validation via API")
+
+    # Call check_provisioning_status - the server will internally:
+    # 1. Check if refresh token exists for the user
+    # 2. Validate the refresh token is not expired
+    # 3. Return provisioning status
+    result = await nc_mcp_oauth_client.call_tool(
+        "check_provisioning_status", arguments={}
+    )
+
+    assert result.isError is False, f"Tool execution failed: {result.content}"
+    assert result.content is not None
+
+    response_text = result.content[0].text
+    logger.info(f"Provisioning status response: {response_text}")
+
+    # Parse the response to validate server's token validation
+    # Expected responses:
+    # 1. "is_provisioned: true" - server validated token successfully
+    # 2. "is_provisioned: false" - no token or invalid token
+    # 3. Error message - token validation failed
+
+    if "is_provisioned" in response_text.lower():
+        if "true" in response_text.lower():
+            logger.info("✓ Server validated refresh token: is_provisioned=True")
+            logger.info("  This confirms the server has a valid refresh token stored")
+        else:
+            logger.info("Server reports: is_provisioned=False (no valid token)")
+    elif "error" in response_text.lower():
+        logger.warning(
+            f"Server returned error during token validation: {response_text}"
+        )
+    else:
+        logger.info(f"Server response: {response_text}")
+
+    # The key validation: Server must return a valid response
+    # (not an error), proving it can check its own refresh token state
+    assert (
+        "is_provisioned" in response_text.lower() or "offline" in response_text.lower()
+    ), f"Expected provisioning status response from server, got: {response_text}"
+
+    logger.info("✓ Server successfully validated refresh token state via API")

tests/server/oauth/test_scope_authorization.py

@@ -412,7 +412,7 @@ async def test_jwt_with_no_custom_scopes_returns_zero_tools(
 
     tool_names = [tool.name for tool in result.tools]
     logger.info(
-        f"JWT token with no custom scopes sees {len(tool_names)} tools (should be 3 OAuth tools)"
+        f"JWT token with no custom scopes sees {len(tool_names)} tools (should be 4 OAuth tools)"
     )
 
     # Only OAuth provisioning tools should be visible (they require 'openid' scope)
@@ -420,6 +420,7 @@ async def test_jwt_with_no_custom_scopes_returns_zero_tools(
         "provision_nextcloud_access",
         "revoke_nextcloud_access",
         "check_provisioning_status",
+        "check_logged_in",  # Login elicitation tool (ADR-006)
     ]
 
     assert set(tool_names) == set(expected_oauth_tools), (

tests/unit/test_unified_verifier.py

@@ -0,0 +1,523 @@
+"""
+Unit tests for UnifiedTokenVerifier (ADR-005).
+
+Tests token audience validation for both multi-audience and token exchange modes
+without requiring real network calls or IdP connections.
+"""
+
+import time
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import jwt
+import pytest
+
+from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
+from nextcloud_mcp_server.config import Settings
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def base_settings():
+    """Create base settings for testing."""
+    return Settings(
+        oidc_client_id="test-client-id",
+        oidc_client_secret="test-client-secret",
+        oidc_issuer="https://idp.example.com",
+        nextcloud_host="https://nextcloud.example.com",
+        nextcloud_mcp_server_url="http://localhost:8000",
+        nextcloud_resource_uri="http://localhost:8080",
+        jwks_uri="https://idp.example.com/jwks",
+        introspection_uri="https://idp.example.com/introspect",
+        enable_token_exchange=False,  # Multi-audience mode
+        token_exchange_cache_ttl=300,
+    )
+
+
+@pytest.fixture
+def exchange_settings(base_settings):
+    """Create settings for token exchange mode."""
+    base_settings.enable_token_exchange = True
+    return base_settings
+
+
+class TestUnifiedTokenVerifierInit:
+    """Test UnifiedTokenVerifier initialization."""
+
+    def test_init_multi_audience_mode(self, base_settings):
+        """Test verifier initialization in multi-audience mode."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        assert verifier.mode == "multi-audience"
+        assert verifier.settings == base_settings
+
+    def test_init_exchange_mode(self, exchange_settings):
+        """Test verifier initialization in token exchange mode."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+        assert verifier.mode == "exchange"
+        assert verifier.settings == exchange_settings
+
+
+class TestAudienceValidation:
+    """Test audience validation logic."""
+
+    def test_validate_multi_audience_both_present(self, base_settings):
+        """Test MCP audience validation with both audiences present."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        payload = {
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        assert verifier._has_mcp_audience(payload) is True
+
+    def test_validate_multi_audience_server_url_and_resource(self, base_settings):
+        """Test MCP audience validation with server URL instead of client ID."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        payload = {
+            "aud": ["http://localhost:8000", "http://localhost:8080"],
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        assert verifier._has_mcp_audience(payload) is True
+
+    def test_validate_multi_audience_missing_mcp(self, base_settings):
+        """Test MCP audience validation fails without MCP audience."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        payload = {
+            "aud": ["http://localhost:8080"],  # Only Nextcloud
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        assert verifier._has_mcp_audience(payload) is False
+
+    def test_validate_multi_audience_missing_nextcloud(self, base_settings):
+        """Test MCP audience validation succeeds with only MCP audience (RFC 7519 compliant)."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        payload = {
+            "aud": ["test-client-id"],  # Only MCP
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        # Per RFC 7519, we only validate MCP audience. Nextcloud validates its own.
+        assert verifier._has_mcp_audience(payload) is True
+
+    def test_validate_multi_audience_string_audience(self, base_settings):
+        """Test MCP audience validation with string audience works (RFC 7519 compliant)."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        payload = {
+            "aud": "test-client-id",  # Single audience as string
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        # Should pass - we only validate MCP audience per RFC 7519
+        assert verifier._has_mcp_audience(payload) is True
+
+    def test_has_mcp_audience_with_client_id(self, exchange_settings):
+        """Test MCP audience validation with client ID."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+        payload = {
+            "aud": ["test-client-id"],
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        assert verifier._has_mcp_audience(payload) is True
+
+    def test_has_mcp_audience_with_server_url(self, exchange_settings):
+        """Test MCP audience validation with server URL."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+        payload = {
+            "aud": ["http://localhost:8000"],
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        assert verifier._has_mcp_audience(payload) is True
+
+    def test_has_mcp_audience_missing(self, exchange_settings):
+        """Test MCP audience validation fails without MCP audience."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+        payload = {
+            "aud": ["http://localhost:8080"],  # Wrong audience
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+
+        assert verifier._has_mcp_audience(payload) is False
+
+
+class TestTokenFormatDetection:
+    """Test JWT format detection."""
+
+    def test_is_jwt_format_valid(self, base_settings):
+        """Test JWT format detection with valid JWT."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        jwt_token = "eyJhbGc.eyJzdWI.signature"
+        assert verifier._is_jwt_format(jwt_token) is True
+
+    def test_is_jwt_format_opaque(self, base_settings):
+        """Test JWT format detection with opaque token."""
+        verifier = UnifiedTokenVerifier(base_settings)
+        opaque_token = "opaque-token-12345"
+        assert verifier._is_jwt_format(opaque_token) is False
+
+
+class TestTokenCaching:
+    """Test token caching functionality."""
+
+    async def test_cache_stores_and_retrieves(self, base_settings):
+        """Test token caching stores and retrieves tokens."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Create a valid access token
+        payload = {
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "sub": "testuser",
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+            "client_id": "test-client-id",
+        }
+        test_token = jwt.encode(payload, "secret", algorithm="HS256")
+
+        # Create AccessToken and cache it
+        access_token = verifier._create_access_token(test_token, payload)
+        assert access_token is not None
+
+        # Should retrieve from cache
+        cached = verifier._get_cached_token(test_token)
+        assert cached is not None
+        assert cached.resource == "testuser"
+        assert cached.scopes == ["openid", "profile"]
+
+    async def test_cache_respects_expiry(self, base_settings):
+        """Test that expired tokens are not returned from cache."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Create expired token payload
+        payload = {
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "sub": "testuser",
+            "scope": "openid profile",
+            "exp": int(time.time() - 100),  # Expired 100 seconds ago
+            "client_id": "test-client-id",
+        }
+        test_token = jwt.encode(payload, "secret", algorithm="HS256")
+
+        # Create and cache
+        access_token = verifier._create_access_token(test_token, payload)
+        assert access_token is not None
+
+        # Should not retrieve expired token
+        cached = verifier._get_cached_token(test_token)
+        assert cached is None
+
+    async def test_cache_clear(self, base_settings):
+        """Test cache clearing."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Create and cache token
+        payload = {
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "sub": "testuser",
+            "exp": int(time.time() + 3600),
+        }
+        test_token = jwt.encode(payload, "secret", algorithm="HS256")
+        verifier._create_access_token(test_token, payload)
+
+        # Clear cache
+        verifier.clear_cache()
+
+        # Should not retrieve after clear
+        cached = verifier._get_cached_token(test_token)
+        assert cached is None
+
+
+class TestMultiAudienceVerification:
+    """Test multi-audience token verification."""
+
+    async def test_verify_multi_audience_with_introspection(self, base_settings):
+        """Test multi-audience verification using introspection."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Mock introspection response
+        introspection_response = {
+            "active": True,
+            "sub": "testuser",
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+            "client_id": "test-client-id",
+        }
+
+        with patch.object(
+            verifier, "_introspect_token", return_value=introspection_response
+        ):
+            opaque_token = "opaque-token-12345"
+            result = await verifier._verify_mcp_audience(opaque_token)
+
+            assert result is not None
+            assert result.resource == "testuser"
+            assert result.scopes == ["openid", "profile"]
+
+    async def test_verify_multi_audience_fails_without_both_audiences(
+        self, base_settings
+    ):
+        """Test MCP audience verification succeeds with only MCP audience (RFC 7519 compliant)."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Mock introspection response with only MCP audience
+        introspection_response = {
+            "active": True,
+            "sub": "testuser",
+            "aud": [
+                "test-client-id"
+            ],  # Only MCP audience (Nextcloud validates its own)
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+
+        with patch.object(
+            verifier, "_introspect_token", return_value=introspection_response
+        ):
+            opaque_token = "opaque-token-12345"
+            result = await verifier._verify_mcp_audience(opaque_token)
+
+            # Should succeed with only MCP audience per RFC 7519
+            assert result is not None
+            assert result.resource == "testuser"
+
+
+class TestExchangeModeVerification:
+    """Test token exchange mode verification."""
+
+    async def test_verify_mcp_audience_only_success(self, exchange_settings):
+        """Test MCP-only audience verification succeeds with MCP audience."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+
+        # Mock introspection response with MCP audience only
+        introspection_response = {
+            "active": True,
+            "sub": "testuser",
+            "aud": ["test-client-id"],
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+            "client_id": "test-client-id",
+        }
+
+        with patch.object(
+            verifier, "_introspect_token", return_value=introspection_response
+        ):
+            opaque_token = "opaque-token-12345"
+            result = await verifier._verify_mcp_audience(opaque_token)
+
+            assert result is not None
+            assert result.resource == "testuser"
+
+    async def test_verify_mcp_audience_only_fails_without_mcp(self, exchange_settings):
+        """Test MCP audience verification fails without MCP audience."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+
+        # Mock introspection response without MCP audience
+        introspection_response = {
+            "active": True,
+            "sub": "testuser",
+            "aud": ["http://localhost:8080"],  # Wrong audience
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+
+        with patch.object(
+            verifier, "_introspect_token", return_value=introspection_response
+        ):
+            opaque_token = "opaque-token-12345"
+            result = await verifier._verify_mcp_audience(opaque_token)
+
+            assert result is None
+
+
+class TestIntrospection:
+    """Test token introspection."""
+
+    async def test_introspect_active_token(self, base_settings):
+        """Test introspection of active token."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Mock HTTP response
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "active": True,
+            "sub": "testuser",
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+            "client_id": "test-client-id",
+        }
+
+        verifier.http_client.post = AsyncMock(return_value=mock_response)
+
+        result = await verifier._introspect_token("test-token")
+        assert result is not None
+        assert result["active"] is True
+        assert result["sub"] == "testuser"
+
+    async def test_introspect_inactive_token(self, base_settings):
+        """Test introspection of inactive token."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        # Mock HTTP response
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"active": False}
+
+        verifier.http_client.post = AsyncMock(return_value=mock_response)
+
+        result = await verifier._introspect_token("test-token")
+        assert result is None
+
+    async def test_introspect_without_endpoint(self, base_settings):
+        """Test introspection when endpoint not configured."""
+        base_settings.introspection_uri = None
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        result = await verifier._introspect_token("test-token")
+        assert result is None
+
+
+class TestAccessTokenCreation:
+    """Test AccessToken object creation."""
+
+    def test_create_access_token_success(self, base_settings):
+        """Test successful AccessToken creation."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        payload = {
+            "sub": "testuser",
+            "scope": "openid profile email",
+            "exp": int(time.time() + 3600),
+            "client_id": "test-client-id",
+        }
+        token = "test-token-123"
+
+        result = verifier._create_access_token(token, payload)
+        assert result is not None
+        assert result.token == token
+        assert result.resource == "testuser"
+        assert result.scopes == ["openid", "profile", "email"]
+        assert result.client_id == "test-client-id"
+
+    def test_create_access_token_with_preferred_username(self, base_settings):
+        """Test AccessToken creation with preferred_username fallback."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        payload = {
+            "preferred_username": "testuser",  # No 'sub' claim
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+        token = "test-token-123"
+
+        result = verifier._create_access_token(token, payload)
+        assert result is not None
+        assert result.resource == "testuser"
+
+    def test_create_access_token_no_username(self, base_settings):
+        """Test AccessToken creation fails without username."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        payload = {
+            # No sub or preferred_username
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+        token = "test-token-123"
+
+        result = verifier._create_access_token(token, payload)
+        assert result is None
+
+    def test_create_access_token_no_expiry(self, base_settings):
+        """Test AccessToken creation uses default TTL without expiry."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        payload = {
+            "sub": "testuser",
+            "scope": "openid profile",
+            # No exp claim
+        }
+        token = "test-token-123"
+
+        result = verifier._create_access_token(token, payload)
+        assert result is not None
+        # Should have set a default expiry
+        assert result.expires_at > int(time.time())
+
+
+class TestVerifyTokenFlow:
+    """Test complete verify_token flow."""
+
+    async def test_verify_token_from_cache(self, base_settings):
+        """Test verify_token returns cached token."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        payload = {
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "sub": "testuser",
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+        token = jwt.encode(payload, "secret", algorithm="HS256")
+
+        # First call - should cache
+        result1 = verifier._create_access_token(token, payload)
+        assert result1 is not None
+
+        # Mock _verify_mcp_audience to ensure it's not called
+        with patch.object(verifier, "_verify_mcp_audience") as mock_verify:
+            result2 = await verifier.verify_token(token)
+            assert result2 is not None
+            assert result2.resource == "testuser"
+            # Should not call verification since it's cached
+            mock_verify.assert_not_called()
+
+    async def test_verify_token_multi_audience_mode(self, base_settings):
+        """Test verify_token in multi-audience mode."""
+        verifier = UnifiedTokenVerifier(base_settings)
+
+        introspection_response = {
+            "active": True,
+            "sub": "testuser",
+            "aud": ["test-client-id", "http://localhost:8080"],
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+
+        with patch.object(
+            verifier, "_introspect_token", return_value=introspection_response
+        ):
+            result = await verifier.verify_token("opaque-token")
+            assert result is not None
+            assert result.resource == "testuser"
+
+    async def test_verify_token_exchange_mode(self, exchange_settings):
+        """Test verify_token in exchange mode."""
+        verifier = UnifiedTokenVerifier(exchange_settings)
+
+        introspection_response = {
+            "active": True,
+            "sub": "testuser",
+            "aud": ["test-client-id"],  # MCP audience only
+            "scope": "openid profile",
+            "exp": int(time.time() + 3600),
+        }
+
+        with patch.object(
+            verifier, "_introspect_token", return_value=introspection_response
+        ):
+            result = await verifier.verify_token("opaque-token")
+            assert result is not None
+            assert result.resource == "testuser"

uv.lock

@@ -501,6 +501,8 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/1f/8e/abdd3f14d735b2929290a018ecf133c901be4874b858dd1c604b9319f064/greenlet-3.2.4-cp311-cp311-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:2523e5246274f54fdadbce8494458a2ebdcdbc7b802318466ac5606d3cded1f8", size = 587684, upload-time = "2025-08-07T13:18:25.164Z" },
     { url = "https://files.pythonhosted.org/packages/5d/65/deb2a69c3e5996439b0176f6651e0052542bb6c8f8ec2e3fba97c9768805/greenlet-3.2.4-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:1987de92fec508535687fb807a5cea1560f6196285a4cde35c100b8cd632cc52", size = 1116647, upload-time = "2025-08-07T13:42:38.655Z" },
     { url = "https://files.pythonhosted.org/packages/3f/cc/b07000438a29ac5cfb2194bfc128151d52f333cee74dd7dfe3fb733fc16c/greenlet-3.2.4-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:55e9c5affaa6775e2c6b67659f3a71684de4c549b3dd9afca3bc773533d284fa", size = 1142073, upload-time = "2025-08-07T13:18:21.737Z" },
+    { url = "https://files.pythonhosted.org/packages/67/24/28a5b2fa42d12b3d7e5614145f0bd89714c34c08be6aabe39c14dd52db34/greenlet-3.2.4-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:c9c6de1940a7d828635fbd254d69db79e54619f165ee7ce32fda763a9cb6a58c", size = 1548385, upload-time = "2025-11-04T12:42:11.067Z" },
+    { url = "https://files.pythonhosted.org/packages/6a/05/03f2f0bdd0b0ff9a4f7b99333d57b53a7709c27723ec8123056b084e69cd/greenlet-3.2.4-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:03c5136e7be905045160b1b9fdca93dd6727b180feeafda6818e6496434ed8c5", size = 1613329, upload-time = "2025-11-04T12:42:12.928Z" },
     { url = "https://files.pythonhosted.org/packages/d8/0f/30aef242fcab550b0b3520b8e3561156857c94288f0332a79928c31a52cf/greenlet-3.2.4-cp311-cp311-win_amd64.whl", hash = "sha256:9c40adce87eaa9ddb593ccb0fa6a07caf34015a29bf8d344811665b573138db9", size = 299100, upload-time = "2025-08-07T13:44:12.287Z" },
     { url = "https://files.pythonhosted.org/packages/44/69/9b804adb5fd0671f367781560eb5eb586c4d495277c93bde4307b9e28068/greenlet-3.2.4-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:3b67ca49f54cede0186854a008109d6ee71f66bd57bb36abd6d0a0267b540cdd", size = 274079, upload-time = "2025-08-07T13:15:45.033Z" },
     { url = "https://files.pythonhosted.org/packages/46/e9/d2a80c99f19a153eff70bc451ab78615583b8dac0754cfb942223d2c1a0d/greenlet-3.2.4-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:ddf9164e7a5b08e9d22511526865780a576f19ddd00d62f8a665949327fde8bb", size = 640997, upload-time = "2025-08-07T13:42:56.234Z" },
@@ -510,6 +512,8 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/19/0d/6660d55f7373b2ff8152401a83e02084956da23ae58cddbfb0b330978fe9/greenlet-3.2.4-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:3b3812d8d0c9579967815af437d96623f45c0f2ae5f04e366de62a12d83a8fb0", size = 607586, upload-time = "2025-08-07T13:18:28.544Z" },
     { url = "https://files.pythonhosted.org/packages/8e/1a/c953fdedd22d81ee4629afbb38d2f9d71e37d23caace44775a3a969147d4/greenlet-3.2.4-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:abbf57b5a870d30c4675928c37278493044d7c14378350b3aa5d484fa65575f0", size = 1123281, upload-time = "2025-08-07T13:42:39.858Z" },
     { url = "https://files.pythonhosted.org/packages/3f/c7/12381b18e21aef2c6bd3a636da1088b888b97b7a0362fac2e4de92405f97/greenlet-3.2.4-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:20fb936b4652b6e307b8f347665e2c615540d4b42b3b4c8a321d8286da7e520f", size = 1151142, upload-time = "2025-08-07T13:18:22.981Z" },
+    { url = "https://files.pythonhosted.org/packages/27/45/80935968b53cfd3f33cf99ea5f08227f2646e044568c9b1555b58ffd61c2/greenlet-3.2.4-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:ee7a6ec486883397d70eec05059353b8e83eca9168b9f3f9a361971e77e0bcd0", size = 1564846, upload-time = "2025-11-04T12:42:15.191Z" },
+    { url = "https://files.pythonhosted.org/packages/69/02/b7c30e5e04752cb4db6202a3858b149c0710e5453b71a3b2aec5d78a1aab/greenlet-3.2.4-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:326d234cbf337c9c3def0676412eb7040a35a768efc92504b947b3e9cfc7543d", size = 1633814, upload-time = "2025-11-04T12:42:17.175Z" },
     { url = "https://files.pythonhosted.org/packages/e9/08/b0814846b79399e585f974bbeebf5580fbe59e258ea7be64d9dfb253c84f/greenlet-3.2.4-cp312-cp312-win_amd64.whl", hash = "sha256:a7d4e128405eea3814a12cc2605e0e6aedb4035bf32697f72deca74de4105e02", size = 299899, upload-time = "2025-08-07T13:38:53.448Z" },
     { url = "https://files.pythonhosted.org/packages/49/e8/58c7f85958bda41dafea50497cbd59738c5c43dbbea5ee83d651234398f4/greenlet-3.2.4-cp313-cp313-macosx_11_0_universal2.whl", hash = "sha256:1a921e542453fe531144e91e1feedf12e07351b1cf6c9e8a3325ea600a715a31", size = 272814, upload-time = "2025-08-07T13:15:50.011Z" },
     { url = "https://files.pythonhosted.org/packages/62/dd/b9f59862e9e257a16e4e610480cfffd29e3fae018a68c2332090b53aac3d/greenlet-3.2.4-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:cd3c8e693bff0fff6ba55f140bf390fa92c994083f838fece0f63be121334945", size = 641073, upload-time = "2025-08-07T13:42:57.23Z" },
@@ -519,6 +523,8 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ee/43/3cecdc0349359e1a527cbf2e3e28e5f8f06d3343aaf82ca13437a9aa290f/greenlet-3.2.4-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:23768528f2911bcd7e475210822ffb5254ed10d71f4028387e5a99b4c6699671", size = 610497, upload-time = "2025-08-07T13:18:31.636Z" },
     { url = "https://files.pythonhosted.org/packages/b8/19/06b6cf5d604e2c382a6f31cafafd6f33d5dea706f4db7bdab184bad2b21d/greenlet-3.2.4-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:00fadb3fedccc447f517ee0d3fd8fe49eae949e1cd0f6a611818f4f6fb7dc83b", size = 1121662, upload-time = "2025-08-07T13:42:41.117Z" },
     { url = "https://files.pythonhosted.org/packages/a2/15/0d5e4e1a66fab130d98168fe984c509249c833c1a3c16806b90f253ce7b9/greenlet-3.2.4-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:d25c5091190f2dc0eaa3f950252122edbbadbb682aa7b1ef2f8af0f8c0afefae", size = 1149210, upload-time = "2025-08-07T13:18:24.072Z" },
+    { url = "https://files.pythonhosted.org/packages/1c/53/f9c440463b3057485b8594d7a638bed53ba531165ef0ca0e6c364b5cc807/greenlet-3.2.4-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:6e343822feb58ac4d0a1211bd9399de2b3a04963ddeec21530fc426cc121f19b", size = 1564759, upload-time = "2025-11-04T12:42:19.395Z" },
+    { url = "https://files.pythonhosted.org/packages/47/e4/3bb4240abdd0a8d23f4f88adec746a3099f0d86bfedb623f063b2e3b4df0/greenlet-3.2.4-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:ca7f6f1f2649b89ce02f6f229d7c19f680a6238af656f61e0115b24857917929", size = 1634288, upload-time = "2025-11-04T12:42:21.174Z" },
     { url = "https://files.pythonhosted.org/packages/0b/55/2321e43595e6801e105fcfdee02b34c0f996eb71e6ddffca6b10b7e1d771/greenlet-3.2.4-cp313-cp313-win_amd64.whl", hash = "sha256:554b03b6e73aaabec3745364d6239e9e012d64c68ccd0b8430c64ccc14939a8b", size = 299685, upload-time = "2025-08-07T13:24:38.824Z" },
     { url = "https://files.pythonhosted.org/packages/22/5c/85273fd7cc388285632b0498dbbab97596e04b154933dfe0f3e68156c68c/greenlet-3.2.4-cp314-cp314-macosx_11_0_universal2.whl", hash = "sha256:49a30d5fda2507ae77be16479bdb62a660fa51b1eb4928b524975b3bde77b3c0", size = 273586, upload-time = "2025-08-07T13:16:08.004Z" },
     { url = "https://files.pythonhosted.org/packages/d1/75/10aeeaa3da9332c2e761e4c50d4c3556c21113ee3f0afa2cf5769946f7a3/greenlet-3.2.4-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:299fd615cd8fc86267b47597123e3f43ad79c9d8a22bebdce535e53550763e2f", size = 686346, upload-time = "2025-08-07T13:42:59.944Z" },
@@ -526,6 +532,8 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/dc/8b/29aae55436521f1d6f8ff4e12fb676f3400de7fcf27fccd1d4d17fd8fecd/greenlet-3.2.4-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:b4a1870c51720687af7fa3e7cda6d08d801dae660f75a76f3845b642b4da6ee1", size = 694659, upload-time = "2025-08-07T13:53:17.759Z" },
     { url = "https://files.pythonhosted.org/packages/92/2e/ea25914b1ebfde93b6fc4ff46d6864564fba59024e928bdc7de475affc25/greenlet-3.2.4-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:061dc4cf2c34852b052a8620d40f36324554bc192be474b9e9770e8c042fd735", size = 695355, upload-time = "2025-08-07T13:18:34.517Z" },
     { url = "https://files.pythonhosted.org/packages/72/60/fc56c62046ec17f6b0d3060564562c64c862948c9d4bc8aa807cf5bd74f4/greenlet-3.2.4-cp314-cp314-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:44358b9bf66c8576a9f57a590d5f5d6e72fa4228b763d0e43fee6d3b06d3a337", size = 657512, upload-time = "2025-08-07T13:18:33.969Z" },
+    { url = "https://files.pythonhosted.org/packages/23/6e/74407aed965a4ab6ddd93a7ded3180b730d281c77b765788419484cdfeef/greenlet-3.2.4-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:2917bdf657f5859fbf3386b12d68ede4cf1f04c90c3a6bc1f013dd68a22e2269", size = 1612508, upload-time = "2025-11-04T12:42:23.427Z" },
+    { url = "https://files.pythonhosted.org/packages/0d/da/343cd760ab2f92bac1845ca07ee3faea9fe52bee65f7bcb19f16ad7de08b/greenlet-3.2.4-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:015d48959d4add5d6c9f6c5210ee3803a830dce46356e3bc326d6776bde54681", size = 1680760, upload-time = "2025-11-04T12:42:25.341Z" },
     { url = "https://files.pythonhosted.org/packages/e3/a5/6ddab2b4c112be95601c13428db1d8b6608a8b6039816f2ba09c346c08fc/greenlet-3.2.4-cp314-cp314-win_amd64.whl", hash = "sha256:e37ab26028f12dbb0ff65f29a8d3d44a765c61e729647bf2ddfbbed621726f01", size = 303425, upload-time = "2025-08-07T13:32:27.59Z" },
 ]
 
@@ -929,7 +937,7 @@ wheels = [
 
 [[package]]
 name = "mcp"
-version = "1.19.0"
+version = "1.20.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "anyio" },
@@ -938,15 +946,16 @@ dependencies = [
     { name = "jsonschema" },
     { name = "pydantic" },
     { name = "pydantic-settings" },
+    { name = "pyjwt", extra = ["crypto"] },
     { name = "python-multipart" },
     { name = "pywin32", marker = "sys_platform == 'win32'" },
     { name = "sse-starlette" },
     { name = "starlette" },
     { name = "uvicorn", marker = "sys_platform != 'emscripten'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/69/2b/916852a5668f45d8787378461eaa1244876d77575ffef024483c94c0649c/mcp-1.19.0.tar.gz", hash = "sha256:213de0d3cd63f71bc08ffe9cc8d4409cc87acffd383f6195d2ce0457c021b5c1", size = 444163, upload-time = "2025-10-24T01:11:15.839Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/22/fae38092e6c2995c03232635028510d77e7decff31b4ae79dfa0ba99c635/mcp-1.20.0.tar.gz", hash = "sha256:9ccc09eaadbfbcbbdab1c9723cfe2e0d1d9e324d7d3ce7e332ef90b09ed35177", size = 451377, upload-time = "2025-10-30T22:14:53.421Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/ce/a3/3e71a875a08b6a830b88c40bc413bff01f1650f1efe8a054b5e90a9d4f56/mcp-1.19.0-py3-none-any.whl", hash = "sha256:f5907fe1c0167255f916718f376d05f09a830a215327a3ccdd5ec8a519f2e572", size = 170105, upload-time = "2025-10-24T01:11:14.151Z" },
+    { url = "https://files.pythonhosted.org/packages/df/00/76fc92f4892d47fecb37131d0e95ea69259f077d84c68f6793a0d96cfe80/mcp-1.20.0-py3-none-any.whl", hash = "sha256:d0dc06f93653f7432ff89f694721c87f79876b6f93741bf628ad1e48f7ac5e5d", size = 173136, upload-time = "2025-10-30T22:14:51.078Z" },
 ]
 
 [package.optional-dependencies]
@@ -966,7 +975,7 @@ wheels = [
 
 [[package]]
 name = "nextcloud-mcp-server"
-version = "0.23.0"
+version = "0.26.0"
 source = { editable = "." }
 dependencies = [
     { name = "aiosqlite" },
@@ -994,6 +1003,7 @@ dev = [
     { name = "pytest-timeout" },
     { name = "reportlab" },
     { name = "ruff" },
+    { name = "ty" },
 ]
 
 [package.metadata]
@@ -1004,7 +1014,7 @@ requires-dist = [
     { name = "click", specifier = ">=8.1.8" },
     { name = "httpx", specifier = ">=0.28.1,<0.29.0" },
     { name = "icalendar", specifier = ">=6.0.0,<7.0.0" },
-    { name = "mcp", extras = ["cli"], specifier = ">=1.19,<1.20" },
+    { name = "mcp", extras = ["cli"], specifier = ">=1.20,<1.21" },
     { name = "pillow", specifier = ">=12.0.0,<12.1.0" },
     { name = "pydantic", specifier = ">=2.11.4" },
     { name = "pyjwt", extras = ["crypto"], specifier = ">=2.8.0" },
@@ -1023,6 +1033,7 @@ dev = [
     { name = "pytest-timeout", specifier = ">=2.3.1" },
     { name = "reportlab", specifier = ">=4.0.0" },
     { name = "ruff", specifier = ">=0.11.13" },
+    { name = "ty", specifier = ">=0.0.1a25" },
 ]
 
 [[package]]
@@ -1954,6 +1965,31 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/00/c0/8f5d070730d7836adc9c9b6408dec68c6ced86b304a9b26a14df072a6e8c/traitlets-5.14.3-py3-none-any.whl", hash = "sha256:b74e89e397b1ed28cc831db7aea759ba6640cb3de13090ca145426688ff1ac4f", size = 85359, upload-time = "2024-04-19T11:11:46.763Z" },
 ]
 
+[[package]]
+name = "ty"
+version = "0.0.1a25"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f6/6b/e73bc3c1039ea72936158a08313155a49e5aa5e7db5205a149fe516a4660/ty-0.0.1a25.tar.gz", hash = "sha256:5550b24b9dd0e0f8b4b2c1f0fcc608a55d0421dd67b6c364bc7bf25762334511", size = 4403670, upload-time = "2025-10-29T19:40:23.647Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/8f/3b/4457231238a2eeb04cba4ba7cc33d735be68ee46ca40a98ae30e187de864/ty-0.0.1a25-py3-none-linux_armv6l.whl", hash = "sha256:d35b2c1f94a014a22875d2745aa0432761d2a9a8eb7212630d5caf547daeef6d", size = 8878803, upload-time = "2025-10-29T19:39:42.243Z" },
+    { url = "https://files.pythonhosted.org/packages/8a/fa/a328713dd310018fc7a381693d8588185baa2fdae913e01a6839187215df/ty-0.0.1a25-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:192edac94675a468bac7f6e04687a77a64698e4e1fe01f6a048bf9b6dde5b703", size = 8695667, upload-time = "2025-10-29T19:39:45.179Z" },
+    { url = "https://files.pythonhosted.org/packages/22/e8/5707939118992ced2bf5385adc3ede7723c1b717b07ad14c495eea1e47b4/ty-0.0.1a25-py3-none-macosx_11_0_arm64.whl", hash = "sha256:949523621f336e01bc7d687b7bd08fe838edadbdb6563c2c057ed1d264e820cf", size = 8159012, upload-time = "2025-10-29T19:39:47.011Z" },
+    { url = "https://files.pythonhosted.org/packages/eb/fb/ff313aa71602225cd78f1bce3017713d6d1b1c1e0fa8101ead4594a60d95/ty-0.0.1a25-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:94f78f621458c05e59e890061021198197f29a7b51a33eda82bbb036e7ed73d7", size = 8433675, upload-time = "2025-10-29T19:39:48.443Z" },
+    { url = "https://files.pythonhosted.org/packages/c0/8d/cc7e7fb57215a15b575a43ed042bdd92971871e0decec1b26d2e7d969465/ty-0.0.1a25-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d9656fca8062a2c6709c30d76d662c96d2e7dbfee8f70e55ec6b6afd67b5d447", size = 8668456, upload-time = "2025-10-29T19:39:50.412Z" },
+    { url = "https://files.pythonhosted.org/packages/b8/6d/d7bf5909ed2dcdcbc1e2ca7eea80929893e2d188d9c36b3fcb2b36532ff6/ty-0.0.1a25-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a9f3bbf523b49935bbd76e230408d858dce0d614f44f5807bbbd0954f64e0f01", size = 9023543, upload-time = "2025-10-29T19:39:52.292Z" },
+    { url = "https://files.pythonhosted.org/packages/b4/b8/72bcefb4be32e5a84f0b21de2552f16cdb4cae3eb271ac891c8199c26b1a/ty-0.0.1a25-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:f13ea9815f4a54a0a303ca7bf411b0650e3c2a24fc6c7889ffba2c94f5e97a6a", size = 9700013, upload-time = "2025-10-29T19:39:57.283Z" },
+    { url = "https://files.pythonhosted.org/packages/90/0d/cf7e794b840cf6b0bbecb022e593c543f85abad27a582241cf2095048cb1/ty-0.0.1a25-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:eab6e33ebe202a71a50c3d5a5580e3bc1a85cda3ffcdc48cec3f1c693b7a873b", size = 9372574, upload-time = "2025-10-29T19:40:04.532Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/71/2d35e7d51b48eabd330e2f7b7e0bce541cbd95950c4d2f780e85f3366af1/ty-0.0.1a25-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:f6b9a31da43424cdab483703a54a561b93aabba84630788505329fc5294a9c62", size = 9535726, upload-time = "2025-10-29T19:40:06.548Z" },
+    { url = "https://files.pythonhosted.org/packages/57/d3/01ecc23bbd8f3e0dfbcf9172d06d84e88155c5f416f1491137e8066fd859/ty-0.0.1a25-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0a90d897a7c1a5ae9b41a4c7b0a42262a06361476ad88d783dbedd7913edadbc", size = 9003380, upload-time = "2025-10-29T19:40:08.683Z" },
+    { url = "https://files.pythonhosted.org/packages/de/f9/cde9380d8a1a6ca61baeb9aecb12cbec90d489aa929be55cd78ad5c2ccd9/ty-0.0.1a25-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:93c7e7ab2859af0f866d34d27f4ae70dd4fb95b847387f082de1197f9f34e068", size = 8401833, upload-time = "2025-10-29T19:40:10.627Z" },
+    { url = "https://files.pythonhosted.org/packages/0b/39/0acf3625b0c495011795a391016b572f97a812aca1d67f7a76621fdb9ebf/ty-0.0.1a25-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:4a247061bd32bae3865a236d7f8b6c9916c80995db30ae1600999010f90623a9", size = 8706761, upload-time = "2025-10-29T19:40:12.575Z" },
+    { url = "https://files.pythonhosted.org/packages/25/73/7de1648f3563dd9d416d36ab5f1649bfd7b47a179135027f31d44b89a246/ty-0.0.1a25-py3-none-musllinux_1_2_i686.whl", hash = "sha256:1711dd587eccf04fd50c494dc39babe38f4cb345bc3901bf1d8149cac570e979", size = 8792426, upload-time = "2025-10-29T19:40:14.553Z" },
+    { url = "https://files.pythonhosted.org/packages/7d/8a/b6e761a65eac7acd10b2e452f49b2d8ae0ea163ca36bb6b18b2dadae251b/ty-0.0.1a25-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:5f4c9b0cf7995e2e3de9bab4d066063dea92019f2f62673b7574e3612643dd35", size = 9103991, upload-time = "2025-10-29T19:40:16.332Z" },
+    { url = "https://files.pythonhosted.org/packages/e4/25/9324ae947fcc4322470326cf8276a3fc2f08dc82adec1de79d963fdf7af5/ty-0.0.1a25-py3-none-win32.whl", hash = "sha256:168fc8aee396d617451acc44cd28baffa47359777342836060c27aa6f37e2445", size = 8387095, upload-time = "2025-10-29T19:40:18.368Z" },
+    { url = "https://files.pythonhosted.org/packages/3b/2b/cb12cbc7db1ba310aa7b1de9b4e018576f653105993736c086ee67d2ec02/ty-0.0.1a25-py3-none-win_amd64.whl", hash = "sha256:a2fad3d8e92bb4d57a8872a6f56b1aef54539d36f23ebb01abe88ac4338efafb", size = 9059225, upload-time = "2025-10-29T19:40:20.278Z" },
+    { url = "https://files.pythonhosted.org/packages/2f/c1/f6be8cdd0bf387c1d8ee9d14bb299b7b5d2c0532f550a6693216a32ec0c5/ty-0.0.1a25-py3-none-win_arm64.whl", hash = "sha256:dde2962d448ed87c48736e9a4bb13715a4cced705525e732b1c0dac1d4c66e3d", size = 8536832, upload-time = "2025-10-29T19:40:22.014Z" },
+]
+
 [[package]]
 name = "typer"
 version = "0.19.2"