feat: Enable SSE transport for mcp service and update test fixtures

Changes:
- Remove streamable-http transport override from mcp service in docker-compose.yml
- Service now uses CLI default SSE transport on /sse endpoint
- Add create_mcp_client_session_sse() helper for SSE connections
- Update nc_mcp_client fixture to use SSE transport
- Fix unpacking for SSE client (yields 2 values vs 3 for streamable-http)

Testing:
- All 4 smoke tests pass with SSE transport
- 32/34 affected tests pass (2 skipped for vector sync)
- OAuth services remain on streamable-http (unchanged)

Note: SSE transport is being deprecated in favor of streamable-http.
This enables minimal validation testing before deprecation.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

.github/workflows/test.yml

@@ -85,4 +85,4 @@ jobs:
           NEXTCLOUD_USERNAME: "admin"
           NEXTCLOUD_PASSWORD: "admin"
         run: |
-          uv run pytest -v --log-cli-level=WARN --ignore=tests/manual
+          uv run pytest -v --log-cli-level=WARN -m smoke

CHANGELOG.md

@@ -1,3 +1,15 @@
+## v0.34.2 (2025-11-13)
+
+### Fix
+
+- Use NEXTCLOUD_OIDC_CLIENT_ID/SECRET env vars consistently
+
+## v0.34.1 (2025-11-13)
+
+### Fix
+
+- return all notes when search query is empty
+
 ## v0.34.0 (2025-11-13)
 
 ### Feat

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.34.0
+version: 0.34.2
-appVersion: "0.34.0"
+appVersion: "0.34.2"
 keywords:
   - nextcloud
   - mcp

docker-compose.yml

@@ -3,7 +3,7 @@ services:
   # https://hub.docker.com/_/mariadb
   db:
     # Note: Check the recommend version here: https://docs.nextcloud.com/server/latest/admin_manual/installation/system_requirements.html#server
-    image: docker.io/library/mariadb:lts@sha256:404ebf26ed7a56fbab05c29f6f1e70188e5eadb51bba8cee8d355775776deb08
+    image: docker.io/library/mariadb:lts@sha256:6b848cb24fbbd87429917f6c4422ac53c343e85692eb0fef86553e99e4f422f3
     restart: always
     command: --transaction-isolation=READ-COMMITTED
     volumes:
@@ -69,7 +69,6 @@ services:
 
   mcp:
     build: .
-    command: ["--transport", "streamable-http"]
     restart: always
     depends_on:
       app:
@@ -156,7 +155,7 @@ services:
       - oauth-tokens:/app/data
 
   keycloak:
-    image: quay.io/keycloak/keycloak:26.4.4@sha256:c6459d5fae1b759f5d667ebdc6237ab3121379c3494e213898569014ede1846d
+    image: quay.io/keycloak/keycloak:26.4.5@sha256:653852bfdea2be6e958b9e90a976eff1c6de34edd55f2f679bdc48ef16bc528e
     command:
       - "start-dev"
       - "--import-realm"

nextcloud_mcp_server/app.py

@@ -507,9 +507,9 @@ async def setup_oauth_config():
     - External IdP mode: OIDC_DISCOVERY_URL points to external provider
       → External IdP for OAuth, Nextcloud user_oidc validates tokens and provides API access
 
-    Uses generic OIDC environment variables:
+    Uses OIDC environment variables:
     - OIDC_DISCOVERY_URL: OIDC discovery endpoint (optional, defaults to NEXTCLOUD_HOST)
-    - OIDC_CLIENT_ID / OIDC_CLIENT_SECRET: Static credentials (optional, uses DCR if not provided)
+    - NEXTCLOUD_OIDC_CLIENT_ID / NEXTCLOUD_OIDC_CLIENT_SECRET: Static credentials (optional, uses DCR if not provided)
     - NEXTCLOUD_OIDC_SCOPES: Requested OAuth scopes
 
     This is done synchronously before FastMCP initialization because FastMCP
@@ -633,19 +633,21 @@ async def setup_oauth_config():
             )
 
     # Load client credentials (static or dynamic registration)
-    client_id = os.getenv("OIDC_CLIENT_ID")
+    client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID")
-    client_secret = os.getenv("OIDC_CLIENT_SECRET")
+    client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET")
 
     if client_id and client_secret:
         logger.info(f"Using static OIDC client credentials: {client_id}")
     elif registration_endpoint:
-        logger.info("OIDC_CLIENT_ID not set, attempting Dynamic Client Registration")
+        logger.info(
+            "NEXTCLOUD_OIDC_CLIENT_ID not set, attempting Dynamic Client Registration"
+        )
         client_id, client_secret = await load_oauth_client_credentials(
             nextcloud_host=nextcloud_host, registration_endpoint=registration_endpoint
         )
     else:
         raise ValueError(
-            "OIDC_CLIENT_ID and OIDC_CLIENT_SECRET environment variables are required "
+            "NEXTCLOUD_OIDC_CLIENT_ID and NEXTCLOUD_OIDC_CLIENT_SECRET environment variables are required "
             "when the OIDC provider does not support Dynamic Client Registration. "
             f"Discovery URL: {discovery_url}"
         )

nextcloud_mcp_server/config.py

@@ -288,8 +288,8 @@ def get_settings() -> Settings:
     return Settings(
         # OAuth/OIDC settings
         oidc_discovery_url=os.getenv("OIDC_DISCOVERY_URL"),
-        oidc_client_id=os.getenv("OIDC_CLIENT_ID"),
+        oidc_client_id=os.getenv("NEXTCLOUD_OIDC_CLIENT_ID"),
-        oidc_client_secret=os.getenv("OIDC_CLIENT_SECRET"),
+        oidc_client_secret=os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET"),
         oidc_issuer=os.getenv("OIDC_ISSUER"),
         # Nextcloud settings
         nextcloud_host=os.getenv("NEXTCLOUD_HOST"),

nextcloud_mcp_server/controllers/notes_search.py

@@ -12,13 +12,24 @@ class NotesSearchController:
         """
         Search notes using token-based matching with relevance ranking.
         Returns notes sorted by relevance score.
+        If query is empty, returns all notes.
         """
         search_results = []
         query_tokens = self._process_query(query)
 
-        # If empty query after processing, return empty results
+        # If empty query after processing, return all notes
         if not query_tokens:
-            return []
+            async for note in notes:
+                search_results.append(
+                    {
+                        "id": note.get("id"),
+                        "title": note.get("title"),
+                        "category": note.get("category"),
+                        "modified": note.get("modified"),
+                        "_score": None,  # No score for unfiltered results
+                    }
+                )
+            return search_results
 
         # Process and score each note
         async for note in notes:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nextcloud-mcp-server"
-version = "0.34.0"
+version = "0.34.2"
 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"}

scripts/add_scope_decorators.py

@@ -1,307 +0,0 @@
-#!/usr/bin/env python3
-"""Script to automatically add @require_scopes decorators to MCP tools.
-
-This script parses server module files and adds appropriate scope decorators
-based on the operation type (read vs write).
-
-Usage:
-    python scripts/add_scope_decorators.py [--dry-run] [--file FILE]
-"""
-
-import argparse
-import ast
-import re
-from pathlib import Path
-from typing import List, Tuple
-
-# Operation patterns for classification
-READ_PATTERNS = [
-    r".*_get_.*",
-    r".*_get$",
-    r".*_list_.*",
-    r".*_list$",
-    r".*_search_.*",
-    r".*_search$",
-    r".*_read_.*",
-    r".*_read$",
-    r".*_find_.*",
-    r".*_find$",
-    r".*_fetch_.*",
-    r".*_fetch$",
-    r".*_retrieve_.*",
-    r".*_retrieve$",
-]
-
-WRITE_PATTERNS = [
-    r".*_create_.*",
-    r".*_create$",
-    r".*_update_.*",
-    r".*_update$",
-    r".*_delete_.*",
-    r".*_delete$",
-    r".*_append_.*",
-    r".*_append$",
-    r".*_modify_.*",
-    r".*_modify$",
-    r".*_set_.*",
-    r".*_set$",
-    r".*_add_.*",
-    r".*_add$",
-    r".*_remove_.*",
-    r".*_remove$",
-    r".*_edit_.*",
-    r".*_edit$",
-    r".*_move_.*",
-    r".*_move$",
-    r".*_copy_.*",
-    r".*_copy$",
-    r".*_upload_.*",
-    r".*_upload$",
-    r".*_download_.*",
-    r".*_download$",
-    r".*_share_.*",
-    r".*_share$",
-    r".*_unshare_.*",
-    r".*_unshare$",
-    r".*_bulk_.*",  # Bulk operations are typically writes
-]
-
-
-def classify_operation(func_name: str) -> str | None:
-    """Classify a function as read or write operation.
-
-    Args:
-        func_name: Function name to classify
-
-    Returns:
-        "nc:read", "nc:write", or None if cannot classify
-    """
-    # Check write patterns first (more specific)
-    for pattern in WRITE_PATTERNS:
-        if re.match(pattern, func_name):
-            return "nc:write"
-
-    # Check read patterns
-    for pattern in READ_PATTERNS:
-        if re.match(pattern, func_name):
-            return "nc:read"
-
-    return None
-
-
-def has_scope_decorator(decorators: List[ast.expr]) -> bool:
-    """Check if function already has @require_scopes decorator."""
-    for decorator in decorators:
-        if isinstance(decorator, ast.Call):
-            if (
-                isinstance(decorator.func, ast.Name)
-                and decorator.func.id == "require_scopes"
-            ):
-                return True
-        elif isinstance(decorator, ast.Name) and decorator.name == "require_scopes":
-            return True
-    return False
-
-
-def has_mcp_tool_decorator(decorators: List[ast.expr]) -> bool:
-    """Check if function has @mcp.tool() decorator."""
-    for decorator in decorators:
-        if isinstance(decorator, ast.Call):
-            if isinstance(decorator.func, ast.Attribute):
-                if decorator.func.attr == "tool":
-                    return True
-    return False
-
-
-def find_tools_needing_decorators(
-    file_path: Path, verbose: bool = False
-) -> List[Tuple[str, int, str]]:
-    """Find all tools that need scope decorators.
-
-    Returns:
-        List of (function_name, line_number, required_scope)
-    """
-    with open(file_path) as f:
-        content = f.read()
-
-    try:
-        tree = ast.parse(content)
-    except SyntaxError as e:
-        print(f"  ⚠️  Syntax error in {file_path}: {e}")
-        return []
-
-    tools_to_update = []
-    total_functions = 0
-    mcp_tools = 0
-    already_has_scope = 0
-    cannot_classify = 0
-
-    for node in ast.walk(tree):
-        if isinstance(node, ast.FunctionDef):
-            total_functions += 1
-
-            if verbose and node.decorator_list:
-                decorators_str = [
-                    ast.unparse(d) if hasattr(ast, "unparse") else str(d)
-                    for d in node.decorator_list
-                ]
-                print(f"  Function {node.name} has decorators: {decorators_str}")
-
-            # Check if it's an MCP tool
-            if not has_mcp_tool_decorator(node.decorator_list):
-                continue
-
-            mcp_tools += 1
-
-            # Check if it already has scope decorator
-            if has_scope_decorator(node.decorator_list):
-                already_has_scope += 1
-                continue
-
-            # Classify operation
-            scope = classify_operation(node.name)
-            if scope:
-                tools_to_update.append((node.name, node.lineno, scope))
-            else:
-                cannot_classify += 1
-                if verbose:
-                    print(f"  ⚠️  Cannot classify: {node.name}")
-
-    if verbose:
-        print(
-            f"  Debug: total_functions={total_functions}, mcp_tools={mcp_tools}, already_has_scope={already_has_scope}, cannot_classify={cannot_classify}"
-        )
-
-    return tools_to_update
-
-
-def add_decorator_to_file(
-    file_path: Path, dry_run: bool = False, verbose: bool = False
-) -> int:
-    """Add @require_scopes decorators to tools in a file.
-
-    Returns:
-        Number of decorators added
-    """
-    tools = find_tools_needing_decorators(file_path, verbose=verbose)
-
-    if not tools:
-        return 0
-
-    print(f"\n📝 {file_path.relative_to(Path.cwd())}")
-
-    with open(file_path) as f:
-        lines = f.readlines()
-
-    # Check if require_scopes is already imported
-    has_import = False
-    import_line_idx = None
-    for i, line in enumerate(lines):
-        if "from nextcloud_mcp_server.auth import" in line and "require_scopes" in line:
-            has_import = True
-            break
-        elif "from nextcloud_mcp_server.auth import" in line:
-            import_line_idx = i
-
-    # Add import if needed
-    if not has_import:
-        if import_line_idx is not None:
-            # Add require_scopes to existing import
-            old_line = lines[import_line_idx]
-            if "(" in old_line:
-                # Multi-line import
-                print(
-                    "  ⚠️  Multi-line import detected, please add manually: from nextcloud_mcp_server.auth import require_scopes"
-                )
-            else:
-                # Single line import - add require_scopes
-                lines[import_line_idx] = (
-                    old_line.rstrip().rstrip(")").rstrip() + ", require_scopes)\n"
-                )
-                print("  ✓ Added require_scopes to import")
-        else:
-            # No auth import exists, add new import
-            # Find first import line
-            for i, line in enumerate(lines):
-                if line.startswith("from nextcloud_mcp_server"):
-                    lines.insert(
-                        i, "from nextcloud_mcp_server.auth import require_scopes\n"
-                    )
-                    print(
-                        "  ✓ Added import: from nextcloud_mcp_server.auth import require_scopes"
-                    )
-                    break
-
-    # Add decorators to tools (in reverse order to preserve line numbers)
-    for func_name, line_num, scope in reversed(tools):
-        # Find the @mcp.tool() decorator line
-        for i in range(line_num - 1, max(0, line_num - 10), -1):
-            if "@mcp.tool()" in lines[i]:
-                # Get indentation from @mcp.tool() line
-                indent = len(lines[i]) - len(lines[i].lstrip())
-                decorator_line = " " * indent + f'@require_scopes("{scope}")\n'
-                lines.insert(i + 1, decorator_line)
-                print(f'  ✓ {func_name}:{line_num} → @require_scopes("{scope}")')
-                break
-
-    if not dry_run:
-        with open(file_path, "w") as f:
-            f.writelines(lines)
-        print("  💾 Saved changes")
-    else:
-        print("  🔍 DRY RUN - no changes written")
-
-    return len(tools)
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Add @require_scopes decorators to MCP tools"
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",
-        help="Show what would be changed without modifying files",
-    )
-    parser.add_argument(
-        "--file",
-        type=Path,
-        help="Process a single file instead of all server modules",
-    )
-    parser.add_argument(
-        "--verbose",
-        "-v",
-        action="store_true",
-        help="Show debug information",
-    )
-    args = parser.parse_args()
-
-    server_dir = Path(__file__).parent.parent / "nextcloud_mcp_server" / "server"
-
-    if args.file:
-        files = [args.file]
-    else:
-        files = sorted(server_dir.glob("*.py"))
-        files = [f for f in files if f.name != "__init__.py"]
-
-    print("🔍 Scanning for tools needing scope decorators...")
-    print(
-        f"   {'DRY RUN MODE - No changes will be made' if args.dry_run else 'LIVE MODE - Files will be modified'}"
-    )
-
-    total_added = 0
-    for file_path in files:
-        added = add_decorator_to_file(
-            file_path, dry_run=args.dry_run, verbose=args.verbose
-        )
-        total_added += added
-
-    print(f"\n{'📊 Summary (dry run)' if args.dry_run else '✅ Complete'}")
-    print(f"   Total decorators added: {total_added}")
-
-    if args.dry_run:
-        print("\n💡 Run without --dry-run to apply changes")
-
-
-if __name__ == "__main__":
-    main()

scripts/add_scope_decorators_simple.py

@@ -1,232 +0,0 @@
-#!/usr/bin/env python3
-"""Simpler script to add @require_scopes decorators using regex.
-
-This script uses regex patterns to find @mcp.tool() decorators and adds
-the appropriate @require_scopes decorator based on function name patterns.
-
-Usage:
-    python scripts/add_scope_decorators_simple.py [--dry-run]
-"""
-
-import argparse
-import re
-from pathlib import Path
-
-# Operation patterns for classification
-READ_KEYWORDS = [
-    "get",
-    "list",
-    "search",
-    "read",
-    "find",
-    "fetch",
-    "retrieve",
-    "upcoming",
-]
-WRITE_KEYWORDS = [
-    "create",
-    "update",
-    "delete",
-    "append",
-    "modify",
-    "set",
-    "add",
-    "remove",
-    "edit",
-    "move",
-    "copy",
-    "upload",
-    "download",
-    "share",
-    "unshare",
-    "bulk",
-    "manage",
-    "import",
-    "reindex",
-    "archive",
-    "unarchive",
-    "reorder",
-    "assign",
-    "unassign",
-    "insert",
-    "write",
-]
-
-
-def classify_function(func_name: str) -> str | None:
-    """Classify a function name as read or write operation."""
-    func_lower = func_name.lower()
-
-    # Check write keywords first (more specific)
-    for keyword in WRITE_KEYWORDS:
-        if f"_{keyword}_" in func_lower or func_lower.endswith(f"_{keyword}"):
-            return "nc:write"
-
-    # Check read keywords
-    for keyword in READ_KEYWORDS:
-        if f"_{keyword}_" in func_lower or func_lower.endswith(f"_{keyword}"):
-            return "nc:read"
-
-    return None
-
-
-def process_file(file_path: Path, dry_run: bool = False) -> int:
-    """Process a single file to add @require_scopes decorators.
-
-    Returns:
-        Number of decorators added
-    """
-    with open(file_path) as f:
-        lines = f.readlines()
-
-    # Check if require_scopes is already imported
-    has_import = False
-    import_line_idx = None
-
-    for i, line in enumerate(lines):
-        if "from nextcloud_mcp_server.auth import" in line:
-            if "require_scopes" in line:
-                has_import = True
-            else:
-                import_line_idx = i
-
-    modified = False
-    decorators_added = 0
-
-    # Find all @mcp.tool() decorators
-    i = 0
-    while i < len(lines):
-        line = lines[i]
-
-        # Look for @mcp.tool() decorator
-        if re.match(r"\s*@mcp\.tool\(\)", line):
-            # Check if next line already has @require_scopes
-            if i + 1 < len(lines) and "@require_scopes" in lines[i + 1]:
-                i += 1
-                continue
-
-            # Find the function definition (should be on next line or after other decorators)
-            func_line_idx = i + 1
-            while func_line_idx < len(lines) and not lines[
-                func_line_idx
-            ].strip().startswith("async def"):
-                func_line_idx += 1
-
-            if func_line_idx >= len(lines):
-                i += 1
-                continue
-
-            # Extract function name
-            func_match = re.match(r"\s*async def (\w+)\(", lines[func_line_idx])
-            if not func_match:
-                i += 1
-                continue
-
-            func_name = func_match.group(1)
-            scope = classify_function(func_name)
-
-            if scope:
-                # Get indentation from @mcp.tool() line
-                indent = len(line) - len(line.lstrip())
-                decorator_line = " " * indent + f'@require_scopes("{scope}")\n'
-
-                # Insert after @mcp.tool()
-                lines.insert(i + 1, decorator_line)
-                decorators_added += 1
-                modified = True
-                print(f'  ✓ {func_name} → @require_scopes("{scope}")')
-            else:
-                print(f"  ⚠️  Cannot classify: {func_name}")
-
-        i += 1
-
-    # Add import if needed and decorators were added
-    if decorators_added > 0 and not has_import:
-        if import_line_idx is not None:
-            # Add to existing import
-            old_line = lines[import_line_idx]
-            if old_line.rstrip().endswith(")"):
-                lines[import_line_idx] = old_line.rstrip()[:-1] + ", require_scopes)\n"
-            else:
-                lines[import_line_idx] = old_line.rstrip() + ", require_scopes\n"
-            print("  ✓ Added require_scopes to existing import")
-            modified = True
-        else:
-            # No auth import exists, add new import after last 'from nextcloud_mcp_server' import
-            last_nc_import_idx = None
-            for i, line in enumerate(lines):
-                if line.startswith("from nextcloud_mcp_server"):
-                    last_nc_import_idx = i
-
-            if last_nc_import_idx is not None:
-                lines.insert(
-                    last_nc_import_idx + 1,
-                    "from nextcloud_mcp_server.auth import require_scopes\n",
-                )
-                print(
-                    "  ✓ Added new import: from nextcloud_mcp_server.auth import require_scopes"
-                )
-                modified = True
-            else:
-                print("  ⚠️  Could not find place to add require_scopes import")
-
-    # Write changes
-    if modified and not dry_run:
-        with open(file_path, "w") as f:
-            f.writelines(lines)
-        print(f"  💾 Saved changes to {file_path.name}")
-    elif dry_run and decorators_added > 0:
-        print(f"  🔍 DRY RUN - would add {decorators_added} decorators")
-
-    return decorators_added
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Add @require_scopes decorators to MCP tools"
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",
-        help="Show what would be changed without modifying files",
-    )
-    parser.add_argument(
-        "--file",
-        type=Path,
-        help="Process a single file instead of all server modules",
-    )
-    args = parser.parse_args()
-
-    server_dir = Path(__file__).parent.parent / "nextcloud_mcp_server" / "server"
-
-    if args.file:
-        files = [args.file]
-    else:
-        files = sorted(server_dir.glob("*.py"))
-        files = [f for f in files if f.name != "__init__.py"]
-
-    print("🔍 Scanning for tools needing scope decorators...")
-    print(
-        f"   {'DRY RUN MODE - No changes will be made' if args.dry_run else 'LIVE MODE - Files will be modified'}"
-    )
-
-    total_added = 0
-    for file_path in files:
-        file_path = file_path.resolve()  # Convert to absolute path
-        try:
-            display_path = file_path.relative_to(Path.cwd())
-        except ValueError:
-            display_path = file_path.name
-        print(f"\n📝 {display_path}")
-        added = process_file(file_path, dry_run=args.dry_run)
-        total_added += added
-
-    print(f"\n{'📊 Summary (dry run)' if args.dry_run else '✅ Complete'}")
-    print(f"   Total decorators added: {total_added}")
-
-    if args.dry_run and total_added > 0:
-        print("\n💡 Run without --dry-run to apply changes")
-
-
-if __name__ == "__main__":
-    main()

scripts/test_separate_clients.sh

@@ -1,90 +0,0 @@
-#!/bin/bash
-set -e
-
-echo "=== Testing Separate Clients Architecture ==="
-echo ""
-
-# Check both clients exist in Keycloak
-echo "1. Verifying Keycloak clients..."
-docker compose exec -T app curl -s http://keycloak:8080/realms/nextcloud-mcp/.well-known/openid-configuration > /dev/null && echo "✓ Keycloak realm available"
-
-# Check user_oidc provider configuration
-echo ""
-echo "2. Checking user_oidc provider..."
-PROVIDER_INFO=$(docker compose exec -T app php occ user_oidc:provider keycloak)
-echo "$PROVIDER_INFO" | grep -q "nextcloud" && echo "✓ user_oidc configured with 'nextcloud' client"
-
-# Get token from nextcloud-mcp-server client
-echo ""
-echo "3. Getting token from 'nextcloud-mcp-server' client..."
-TOKEN=$(curl -s -X POST "http://localhost:8888/realms/nextcloud-mcp/protocol/openid-connect/token" \
-  -d "grant_type=password" \
-  -d "client_id=nextcloud-mcp-server" \
-  -d "client_secret=mcp-secret-change-in-production" \
-  -d "username=admin" \
-  -d "password=admin" \
-  -d "scope=openid profile email offline_access" | jq -r '.access_token')
-
-if [ "$TOKEN" = "null" ] || [ -z "$TOKEN" ]; then
-    echo "✗ Failed to get token"
-    exit 1
-fi
-
-echo "✓ Got token from nextcloud-mcp-server client"
-
-# Check token claims
-echo ""
-echo "4. Inspecting token claims..."
-CLAIMS=$(echo "$TOKEN" | cut -d'.' -f2 | base64 -d 2>/dev/null | jq '{aud, azp, iss, preferred_username}')
-echo "$CLAIMS"
-
-AUD=$(echo "$CLAIMS" | jq -r '.aud')
-AZP=$(echo "$CLAIMS" | jq -r '.azp')
-
-echo ""
-echo "Architecture validation:"
-if [ "$AUD" = "nextcloud" ]; then
-    echo "  ✓ aud='nextcloud' - Token intended for Nextcloud resource server"
-else
-    echo "  ✗ FAILED: aud='$AUD', expected 'nextcloud'"
-    exit 1
-fi
-
-if [ "$AZP" = "nextcloud-mcp-server" ]; then
-    echo "  ✓ azp='nextcloud-mcp-server' - Token requested by MCP Server client"
-else
-    echo "  ✗ FAILED: azp='$AZP', expected 'nextcloud-mcp-server'"
-    exit 1
-fi
-
-# Test with Nextcloud API
-echo ""
-echo "5. Testing token with Nextcloud API..."
-HTTP_CODE=$(curl -s -w "%{http_code}" -o /tmp/nc_response.json \
-    -H "Authorization: Bearer $TOKEN" \
-    "http://localhost:8080/ocs/v2.php/cloud/capabilities?format=json")
-
-echo "HTTP Status: $HTTP_CODE"
-
-if [ "$HTTP_CODE" = "200" ]; then
-    echo "✓ Token validated successfully!"
-    echo ""
-    echo "===================================================================="
-    echo "SUCCESS: Separate Clients Architecture Working!"
-    echo "===================================================================="
-    echo ""
-    echo "Summary:"
-    echo "  - MCP Server client: 'nextcloud-mcp-server' (requests tokens)"
-    echo "  - Resource server: 'nextcloud' (validates tokens via user_oidc)"
-    echo "  - Token audience: 'nextcloud' (proper resource targeting)"
-    echo "  - Token azp: 'nextcloud-mcp-server' (identifies requester)"
-    echo ""
-    echo "This architecture supports:"
-    echo "  - Future multi-resource tokens: aud=['nextcloud', 'other-service']"
-    echo "  - Clear separation of OAuth client vs resource server"
-    echo "  - RFC 8707 Resource Indicators compliance"
-else
-    echo "✗ Token validation failed"
-    cat /tmp/nc_response.json
-    exit 1
-fi

tests/conftest.py

@@ -9,6 +9,7 @@ import pytest
 from httpx import HTTPStatusError
 from mcp import ClientSession
 from mcp.client.session import RequestContext
+from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
 from mcp.types import ElicitRequestParams, ElicitResult, ErrorData
 
@@ -165,6 +166,51 @@ async def create_mcp_client_session(
     logger.debug(f"{client_name} client session cleaned up successfully")
 
 
+async def create_mcp_client_session_sse(
+    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 using SSE transport.
+
+    Similar to create_mcp_client_session but uses SSE transport instead of streamable-http.
+    Uses native async context managers to ensure correct LIFO cleanup order.
+
+    Args:
+        url: MCP server URL (e.g., "http://localhost:8000/sse")
+        token: Optional OAuth access token for Bearer authentication
+        client_name: Client name for logging (e.g., "Basic MCP (SSE)")
+        elicitation_callback: Optional callback for handling elicitation requests
+
+    Yields:
+        Initialized MCP ClientSession
+
+    Note:
+        SSE transport is being deprecated in favor of streamable-http.
+        This function exists for compatibility testing only.
+    """
+    logger.info(f"Creating SSE client for {client_name}")
+
+    # Prepare headers with OAuth token if provided
+    headers = {"Authorization": f"Bearer {token}"} if token else None
+
+    # Use native async with - Python ensures LIFO cleanup
+    # Cleanup order will be: ClientSession.__aexit__ -> sse_client.__aexit__
+    # Note: sse_client yields only (read_stream, write_stream), not 3 values like streamablehttp_client
+    async with sse_client(url, headers=headers) as (read_stream, write_stream):
+        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
+
+    # Cleanup happens automatically in LIFO order - no exception suppression needed
+    logger.debug(f"{client_name} client session cleaned up successfully")
+
+
 @pytest.fixture(scope="session")
 async def nc_client(anyio_backend) -> AsyncGenerator[NextcloudClient, Any]:
     """
@@ -203,12 +249,14 @@ async def nc_client(anyio_backend) -> AsyncGenerator[NextcloudClient, Any]:
 @pytest.fixture(scope="session")
 async def nc_mcp_client(anyio_backend) -> AsyncGenerator[ClientSession, Any]:
     """
-    Fixture to create an MCP client session for integration tests using streamable-http.
+    Fixture to create an MCP client session for integration tests using SSE transport.
 
     Uses anyio pytest plugin for proper async fixture handling.
+
+    Note: SSE transport is being deprecated. This fixture uses SSE for compatibility testing.
     """
-    async for session in create_mcp_client_session(
+    async for session in create_mcp_client_session_sse(
-        url="http://localhost:8000/mcp", client_name="Basic MCP"
+        url="http://localhost:8000/sse", client_name="Basic MCP (SSE)"
     ):
         yield session
 

uv.lock

@@ -1053,7 +1053,7 @@ wheels = [
 
 [[package]]
 name = "nextcloud-mcp-server"
-version = "0.34.0"
+version = "0.34.2"
 source = { editable = "." }
 dependencies = [
     { name = "aiosqlite" },