admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into feature/introduce_files_parsing_with_unstructured_service_for_webdav_files_retrieval

Browse Source
This commit is contained in:
yuisheaven
2025-10-23 21:30:09 +02:00
committed by GitHub
parent 98627593d5 8942f3119c
commit 29df645d53
45 changed files with 5287 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/app.py
+334 -77
View File
@@ -11,9 +11,16 @@ from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import Context, FastMCP
from pydantic import AnyHttpUrl
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route
from nextcloud_mcp_server.auth import NextcloudTokenVerifier, load_or_register_client
from nextcloud_mcp_server.auth import (
InsufficientScopeError,
NextcloudTokenVerifier,
get_access_token_scopes,
has_required_scopes,
is_jwt_token,
)
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import LOGGING_CONFIG, setup_logging
from nextcloud_mcp_server.context import get_client as get_nextcloud_client
@@ -135,6 +142,95 @@ def is_oauth_mode() -> bool:
return True
async def load_oauth_client_credentials(
nextcloud_host: str, registration_endpoint: str | None
) -> tuple[str, str]:
"""
Load OAuth client credentials from environment, storage file, or dynamic registration.
This consolidates the client loading logic that was duplicated across multiple functions.
Args:
nextcloud_host: Nextcloud instance URL
registration_endpoint: Dynamic registration endpoint URL (or None if not available)
Returns:
Tuple of (client_id, client_secret)
Raises:
ValueError: If credentials cannot be obtained
"""
# Try environment variables first
client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID")
client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET")
if client_id and client_secret:
logger.info("Using pre-configured OAuth client credentials from environment")
return (client_id, client_secret)
# Try loading from storage file
storage_path = os.getenv(
"NEXTCLOUD_OIDC_CLIENT_STORAGE", ".nextcloud_oauth_client.json"
)
from pathlib import Path
from nextcloud_mcp_server.auth.client_registration import load_client_from_file
client_info = load_client_from_file(Path(storage_path))
if client_info:
logger.info(
f"Loaded OAuth client from storage: {client_info.client_id[:16]}..."
)
return (client_info.client_id, client_info.client_secret)
# Try dynamic registration if available
if registration_endpoint:
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"]
# Get scopes from environment or use defaults
scopes = os.getenv(
"NEXTCLOUD_OIDC_SCOPES", "openid profile email nc:read nc:write"
)
logger.info(f"Requesting OAuth scopes: {scopes}")
# Get token type from environment (Bearer or jwt)
# Note: Must be lowercase "jwt" to match OIDC app's check
token_type = os.getenv("NEXTCLOUD_OIDC_TOKEN_TYPE", "Bearer").lower()
# Special case: "bearer" should remain capitalized for compatibility
if token_type != "jwt":
token_type = "Bearer"
logger.info(f"Requesting token type: {token_type}")
# Load or register client
from nextcloud_mcp_server.auth.client_registration import (
load_or_register_client,
)
client_info = await load_or_register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
storage_path=storage_path,
client_name="Nextcloud MCP Server",
redirect_uris=redirect_uris,
scopes=scopes,
token_type=token_type,
)
logger.info(f"OAuth client ready: {client_info.client_id[:16]}...")
return (client_info.client_id, client_info.client_secret)
# No credentials available
raise ValueError(
"OAuth mode requires either:\n"
"1. NEXTCLOUD_OIDC_CLIENT_ID and NEXTCLOUD_OIDC_CLIENT_SECRET environment variables, OR\n"
"2. Pre-existing client credentials file at NEXTCLOUD_OIDC_CLIENT_STORAGE, OR\n"
"3. Dynamic client registration enabled on Nextcloud OIDC app"
)
@asynccontextmanager
async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
"""
@@ -187,45 +283,24 @@ async def app_lifespan_oauth(server: FastMCP) -> AsyncIterator[OAuthAppContext]:
# Extract endpoints
userinfo_uri = discovery["userinfo_endpoint"]
registration_endpoint = discovery.get("registration_endpoint")
introspection_uri = discovery.get("introspection_endpoint")
logger.info(f"Userinfo endpoint: {userinfo_uri}")
if introspection_uri:
logger.info(f"Introspection endpoint: {introspection_uri}")
# Handle client registration
client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID")
client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET")
storage_path = os.getenv(
"NEXTCLOUD_OIDC_CLIENT_STORAGE", ".nextcloud_oauth_client.json"
# Load OAuth client credentials
client_id, client_secret = await load_oauth_client_credentials(
nextcloud_host=nextcloud_host, registration_endpoint=registration_endpoint
)
if client_id and client_secret:
logger.info("Using pre-configured OAuth client credentials")
elif registration_endpoint:
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"]
# Load or register client
client_info = await load_or_register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
storage_path=storage_path,
client_name="Nextcloud MCP Server",
redirect_uris=redirect_uris,
)
logger.info(f"OAuth client ready: {client_info.client_id[:16]}...")
else:
raise ValueError(
"OAuth mode requires either:\n"
"1. NEXTCLOUD_OIDC_CLIENT_ID and NEXTCLOUD_OIDC_CLIENT_SECRET, OR\n"
"2. Dynamic client registration enabled on Nextcloud OIDC app"
)
# Create token verifier
# Create token verifier with introspection support
token_verifier = NextcloudTokenVerifier(
nextcloud_host=nextcloud_host, userinfo_uri=userinfo_uri
nextcloud_host=nextcloud_host,
userinfo_uri=userinfo_uri,
introspection_uri=introspection_uri,
client_id=client_id,
client_secret=client_secret,
)
logger.info("OAuth initialization complete")
@@ -278,59 +353,60 @@ async def setup_oauth_config():
# Extract endpoints
issuer = discovery["issuer"]
userinfo_uri = discovery["userinfo_endpoint"]
jwks_uri = discovery.get("jwks_uri")
introspection_uri = discovery.get("introspection_endpoint")
registration_endpoint = discovery.get("registration_endpoint")
# Allow override of public issuer URL for clients
# (useful when MCP server accesses Nextcloud via internal URL
# but needs to advertise a different URL to clients)
logger.info("OIDC endpoints discovered:")
logger.info(f" Issuer: {issuer}")
logger.info(f" Userinfo: {userinfo_uri}")
logger.info(f" JWKS: {jwks_uri}")
if introspection_uri:
logger.info(f" Introspection: {introspection_uri}")
# Allow override of public issuer URL for both client configuration and JWT validation
# When clients access Nextcloud via a public URL (e.g., http://127.0.0.1:8080),
# the OIDC app issues JWT tokens with that public URL in the 'iss' claim,
# even though the MCP server accesses Nextcloud via an internal URL (e.g., http://app).
# Therefore, we must validate JWT tokens against the public issuer, not the internal one.
public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
if public_issuer:
public_issuer = public_issuer.rstrip("/")
logger.info(f"Using public issuer URL for clients: {public_issuer}")
logger.info(
f"Using public issuer URL for clients and JWT validation: {public_issuer}"
)
# Use public issuer for both client configuration AND JWT validation
issuer = public_issuer
# Handle client registration
client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID")
client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET")
if client_id and client_secret:
logger.info("Using pre-configured OAuth client credentials")
elif registration_endpoint:
logger.info("Dynamic client registration available")
storage_path = os.getenv(
"NEXTCLOUD_OIDC_CLIENT_STORAGE", ".nextcloud_oauth_client.json"
)
mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
redirect_uris = [f"{mcp_server_url}/oauth/callback"]
# Load or register client
client_info = await load_or_register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
storage_path=storage_path,
client_name="Nextcloud MCP Server",
redirect_uris=redirect_uris,
)
logger.info(f"OAuth client ready: {client_info.client_id[:16]}...")
jwt_validation_issuer = public_issuer
else:
raise ValueError(
"OAuth mode requires either:\n"
"1. NEXTCLOUD_OIDC_CLIENT_ID and NEXTCLOUD_OIDC_CLIENT_SECRET, OR\n"
"2. Dynamic client registration enabled on Nextcloud OIDC app"
)
# Use discovered issuer for both
jwt_validation_issuer = issuer
# Create token verifier
# Load OAuth client credentials
client_id, client_secret = await load_oauth_client_credentials(
nextcloud_host=nextcloud_host, registration_endpoint=registration_endpoint
)
# Create token verifier with JWT support and introspection
token_verifier = NextcloudTokenVerifier(
nextcloud_host=nextcloud_host, userinfo_uri=userinfo_uri
nextcloud_host=nextcloud_host,
userinfo_uri=userinfo_uri,
jwks_uri=jwks_uri, # Enable JWT verification if available
issuer=jwt_validation_issuer, # Use original issuer for JWT validation
introspection_uri=introspection_uri, # Enable introspection for opaque tokens
client_id=client_id,
client_secret=client_secret,
)
# Create auth settings
mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
# Note: We don't set required_scopes here anymore.
# Scopes are now advertised via PRM endpoint and enforced per-tool.
# This allows dynamic tool filtering based on user's actual token scopes.
auth_settings = AuthSettings(
issuer_url=AnyHttpUrl(issuer),
resource_server_url=AnyHttpUrl(mcp_server_url),
required_scopes=["openid", "profile"],
)
logger.info("OAuth configuration complete")
@@ -393,6 +469,54 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
f"Unknown app: {app_name}. Available apps: {list(available_apps.keys())}"
)
# Override list_tools to filter based on user's token scopes (OAuth mode only)
if oauth_enabled:
original_list_tools = mcp._tool_manager.list_tools
def list_tools_filtered():
"""List tools filtered by user's token scopes (JWT tokens only)."""
# Get user's scopes from token using MCP SDK's contextvar
# This works for all request types including list_tools
user_scopes = get_access_token_scopes()
is_jwt = is_jwt_token()
logger.info(
f"🔍 list_tools called - Token type: {'JWT' if is_jwt else 'opaque/none'}, "
f"User scopes: {user_scopes}"
)
# Get all tools
all_tools = original_list_tools()
# Only filter for JWT tokens (opaque tokens show all tools)
# JWT tokens have scopes embedded, so we can reliably filter
# Opaque tokens may not have accurate scope information from introspection
if is_jwt and user_scopes:
allowed_tools = [
tool
for tool in all_tools
if has_required_scopes(tool.fn, user_scopes)
]
logger.info(
f"✂️ JWT scope filtering: {len(allowed_tools)}/{len(all_tools)} tools "
f"available for scopes: {user_scopes}"
)
else:
# Opaque token, BasicAuth mode, or no token - show all tools
allowed_tools = all_tools
reason = (
"opaque token (no filtering)"
if not is_jwt and user_scopes
else "no token/BasicAuth"
)
logger.info(f"📋 Showing all {len(all_tools)} tools ({reason})")
# Return the Tool objects directly (they're already in the correct format)
return allowed_tools
# Replace the tool manager's list_tools method
mcp._tool_manager.list_tools = list_tools_filtered
logger.info("Dynamic tool filtering enabled for OAuth mode (JWT tokens only)")
if transport == "sse":
mcp_app = mcp.sse_app()
lifespan = None
@@ -405,7 +529,71 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
await stack.enter_async_context(mcp.session_manager.run())
yield
app = Starlette(routes=[Mount("/", app=mcp_app)], lifespan=lifespan)
# Add Protected Resource Metadata (PRM) endpoint for OAuth mode
routes = []
if oauth_enabled:
def oauth_protected_resource_metadata(request):
"""RFC 8959 Protected Resource Metadata endpoint."""
mcp_server_url = os.getenv(
"NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000"
)
# Use PUBLIC_ISSUER_URL for authorization server since external clients
# (like Claude) need the publicly accessible URL, not internal Docker URLs
public_issuer_url = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
if not public_issuer_url:
# Fallback to NEXTCLOUD_HOST if PUBLIC_ISSUER_URL not set
public_issuer_url = os.getenv("NEXTCLOUD_HOST", "")
return JSONResponse(
{
"resource": mcp_server_url,
"scopes_supported": ["nc:read", "nc:write"],
"authorization_servers": [public_issuer_url],
"bearer_methods_supported": ["header"],
"resource_signing_alg_values_supported": ["RS256"],
}
)
routes.append(
Route(
"/.well-known/oauth-protected-resource",
oauth_protected_resource_metadata,
methods=["GET"],
)
)
logger.info("Protected Resource Metadata (PRM) endpoint enabled")
routes.append(Mount("/", app=mcp_app))
app = Starlette(routes=routes, lifespan=lifespan)
# Add exception handler for scope challenges (OAuth mode only)
if oauth_enabled:
@app.exception_handler(InsufficientScopeError)
async def handle_insufficient_scope(request, exc: InsufficientScopeError):
"""Return 403 with WWW-Authenticate header for scope challenges."""
resource_url = os.getenv(
"NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000"
)
scope_str = " ".join(exc.missing_scopes)
return JSONResponse(
status_code=403,
headers={
"WWW-Authenticate": (
f'Bearer error="insufficient_scope", '
f'scope="{scope_str}", '
f'resource_metadata="{resource_url}/.well-known/oauth-protected-resource"'
)
},
content={
"error": "insufficient_scope",
"scopes_required": exc.missing_scopes,
},
)
logger.info("WWW-Authenticate scope challenge handler enabled")
return app
@@ -471,6 +659,41 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
show_default=True,
help="MCP server URL for OAuth callbacks (can also use NEXTCLOUD_MCP_SERVER_URL env var)",
)
@click.option(
"--nextcloud-host",
envvar="NEXTCLOUD_HOST",
help="Nextcloud instance URL (can also use NEXTCLOUD_HOST env var)",
)
@click.option(
"--nextcloud-username",
envvar="NEXTCLOUD_USERNAME",
help="Nextcloud username for BasicAuth (can also use NEXTCLOUD_USERNAME env var)",
)
@click.option(
"--nextcloud-password",
envvar="NEXTCLOUD_PASSWORD",
help="Nextcloud password for BasicAuth (can also use NEXTCLOUD_PASSWORD env var)",
)
@click.option(
"--oauth-scopes",
envvar="NEXTCLOUD_OIDC_SCOPES",
default="openid profile email nc:read nc:write",
show_default=True,
help="OAuth scopes to request (can also use NEXTCLOUD_OIDC_SCOPES env var)",
)
@click.option(
"--oauth-token-type",
envvar="NEXTCLOUD_OIDC_TOKEN_TYPE",
default="bearer",
show_default=True,
type=click.Choice(["bearer", "jwt"], case_sensitive=False),
help="OAuth token type (can also use NEXTCLOUD_OIDC_TOKEN_TYPE env var)",
)
@click.option(
"--public-issuer-url",
envvar="NEXTCLOUD_PUBLIC_ISSUER_URL",
help="Public issuer URL for OAuth (can also use NEXTCLOUD_PUBLIC_ISSUER_URL env var)",
)
def run(
host: str,
port: int,
@@ -482,6 +705,12 @@ def run(
oauth_client_secret: str | None,
oauth_storage_path: str,
mcp_server_url: str,
nextcloud_host: str | None,
nextcloud_username: str | None,
nextcloud_password: str | None,
oauth_scopes: str,
oauth_token_type: str,
public_issuer_url: str | None,
):
"""
Run the Nextcloud MCP server.
@@ -493,24 +722,52 @@ def run(
\b
Examples:
# BasicAuth mode (legacy)
# BasicAuth mode with CLI options
$ nextcloud-mcp-server --nextcloud-host=https://cloud.example.com \\
--nextcloud-username=admin --nextcloud-password=secret
# BasicAuth mode with env vars (recommended for credentials)
$ export NEXTCLOUD_HOST=https://cloud.example.com
$ export NEXTCLOUD_USERNAME=admin
$ export NEXTCLOUD_PASSWORD=secret
$ nextcloud-mcp-server --host 0.0.0.0 --port 8000
# OAuth mode with auto-registration
$ nextcloud-mcp-server --oauth
$ nextcloud-mcp-server --nextcloud-host=https://cloud.example.com --oauth
# OAuth mode with pre-configured client
$ nextcloud-mcp-server --oauth --oauth-client-id=xxx --oauth-client-secret=yyy
$ nextcloud-mcp-server --nextcloud-host=https://cloud.example.com --oauth \\
--oauth-client-id=xxx --oauth-client-secret=yyy
# OAuth mode with custom scopes and JWT tokens
$ nextcloud-mcp-server --nextcloud-host=https://cloud.example.com --oauth \\
--oauth-scopes="openid nc:read" --oauth-token-type=jwt
# OAuth with public issuer URL (for Docker/proxy setups)
$ nextcloud-mcp-server --nextcloud-host=http://app --oauth \\
--public-issuer-url=http://localhost:8080
"""
# Set OAuth env vars from CLI options if provided
# Set env vars from CLI options if provided
if nextcloud_host:
os.environ["NEXTCLOUD_HOST"] = nextcloud_host
if nextcloud_username:
os.environ["NEXTCLOUD_USERNAME"] = nextcloud_username
if nextcloud_password:
os.environ["NEXTCLOUD_PASSWORD"] = nextcloud_password
if oauth_client_id:
os.environ["NEXTCLOUD_OIDC_CLIENT_ID"] = oauth_client_id
if oauth_client_secret:
os.environ["NEXTCLOUD_OIDC_CLIENT_SECRET"] = oauth_client_secret
if oauth_storage_path:
os.environ["NEXTCLOUD_OIDC_CLIENT_STORAGE"] = oauth_storage_path
if oauth_scopes:
os.environ["NEXTCLOUD_OIDC_SCOPES"] = oauth_scopes
if oauth_token_type:
os.environ["NEXTCLOUD_OIDC_TOKEN_TYPE"] = oauth_token_type
if mcp_server_url:
os.environ["NEXTCLOUD_MCP_SERVER_URL"] = mcp_server_url
if public_issuer_url:
os.environ["NEXTCLOUD_PUBLIC_ISSUER_URL"] = public_issuer_url
# Force OAuth mode if explicitly requested
if oauth is True:
nextcloud_mcp_server/auth/__init__.py
+18
View File
@@ -3,6 +3,16 @@
from .bearer_auth import BearerAuth
from .client_registration import load_or_register_client, register_client
from .context_helper import get_client_from_context
from .scope_authorization import (
InsufficientScopeError,
ScopeAuthorizationError,
check_scopes,
get_access_token_scopes,
get_required_scopes,
has_required_scopes,
is_jwt_token,
require_scopes,
)
from .token_verifier import NextcloudTokenVerifier
__all__ = [
@@ -11,4 +21,12 @@ __all__ = [
"register_client",
"load_or_register_client",
"get_client_from_context",
"require_scopes",
"ScopeAuthorizationError",
"InsufficientScopeError",
"check_scopes",
"get_access_token_scopes",
"get_required_scopes",
"has_required_scopes",
"is_jwt_token",
]
nextcloud_mcp_server/auth/client_registration.py
+9
View File
@@ -68,6 +68,7 @@ async def register_client(
client_name: str = "Nextcloud MCP Server",
redirect_uris: list[str] | None = None,
scopes: str = "openid profile email",
token_type: str = "Bearer",
) -> ClientInfo:
"""
Register a new OAuth client with Nextcloud OIDC using dynamic client registration.
@@ -78,6 +79,7 @@ async def register_client(
client_name: Name of the client application
redirect_uris: List of redirect URIs (default: http://localhost:8000/oauth/callback)
scopes: Space-separated list of scopes to request
token_type: Type of access tokens to issue (default: "Bearer", also supports "JWT")
Returns:
ClientInfo with registration details
@@ -96,6 +98,7 @@ async def register_client(
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"scope": scopes,
"token_type": token_type,
}
logger.info(f"Registering OAuth client with Nextcloud: {client_name}")
@@ -215,6 +218,8 @@ async def load_or_register_client(
storage_path: str | Path,
client_name: str = "Nextcloud MCP Server",
redirect_uris: list[str] | None = None,
scopes: str = "openid profile email",
token_type: str = "Bearer",
) -> ClientInfo:
"""
Load client from storage or register a new one if not found/expired.
@@ -231,6 +236,8 @@ async def load_or_register_client(
storage_path: Path to store client credentials
client_name: Name of the client application
redirect_uris: List of redirect URIs
scopes: Space-separated list of scopes to request (default: "openid profile email")
token_type: Type of access tokens to issue (default: "Bearer", also supports "JWT")
Returns:
ClientInfo with valid credentials
@@ -253,6 +260,8 @@ async def load_or_register_client(
registration_endpoint=registration_endpoint,
client_name=client_name,
redirect_uris=redirect_uris,
scopes=scopes,
token_type=token_type,
)
# Save to storage
nextcloud_mcp_server/auth/scope_authorization.py
+278
View File
@@ -0,0 +1,278 @@
"""Scope-based authorization for MCP tools."""
import logging
from functools import wraps
from typing import Callable
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 mcp.server.fastmcp.utilities.context_injection import find_context_parameter
logger = logging.getLogger(__name__)
class ScopeAuthorizationError(Exception):
"""Raised when a request lacks required scopes."""
pass
class InsufficientScopeError(ScopeAuthorizationError):
"""Raised when request lacks required scopes (enables step-up auth).
This exception triggers a 403 response with WWW-Authenticate header
containing the missing scopes, allowing clients to perform step-up
authorization to obtain additional permissions.
"""
def __init__(self, missing_scopes: list[str], message: str | None = None):
self.missing_scopes = missing_scopes
super().__init__(
message or f"Missing required scopes: {', '.join(missing_scopes)}"
)
def require_scopes(*required_scopes: str):
"""
Decorator to require specific OAuth scopes for MCP tool execution.
This decorator:
1. Stores scope requirements as function metadata (_required_scopes attribute)
2. Checks that the access token contains all required scopes before execution
3. Raises ScopeAuthorizationError if any required scope is missing
The stored metadata enables dynamic tool filtering - tools can be hidden from
users who lack the necessary scopes.
Args:
*required_scopes: Variable number of scope strings required (e.g., "nc:read", "nc:write")
Returns:
Decorated function that checks scopes before execution
Example:
```python
@mcp.tool()
@require_scopes("nc:read")
async def nc_notes_get_note(ctx: Context, note_id: int):
# This tool requires the nc:read scope
...
@mcp.tool()
@require_scopes("nc:write")
async def nc_notes_create_note(ctx: Context, ...):
# This tool requires the nc:write scope
...
```
Raises:
ScopeAuthorizationError: If required scopes are not present in the access token
"""
def decorator(func: Callable):
# Store scope requirements as function metadata for dynamic filtering
func._required_scopes = list(required_scopes) # type: ignore
# 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):
# Extract context from kwargs (where FastMCP injected it)
ctx: Context | None = (
kwargs.get(context_param_name) if context_param_name else None
)
if ctx is None:
# 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)"
)
return await func(*args, **kwargs)
# Check if we're in OAuth mode (access token available)
access_token: AccessToken | None = getattr(
ctx.request_context, "access_token", None
)
if access_token is None:
# 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)"
)
return await func(*args, **kwargs)
# Extract scopes from access token
token_scopes = set(access_token.scopes or [])
required_scopes_set = set(required_scopes)
# Check if all required scopes are present
missing_scopes = required_scopes_set - token_scopes
if missing_scopes:
error_msg = (
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'}"
)
logger.warning(error_msg)
raise InsufficientScopeError(list(missing_scopes), error_msg)
# All required scopes present - allow execution
logger.debug(
f"Scope authorization passed for {func.__name__}: {required_scopes}"
)
return await func(*args, **kwargs)
return wrapper
return decorator
def get_access_token_scopes(ctx: Context | None = None) -> set[str]:
"""
Extract scopes from the authenticated user's access token.
This function uses MCP SDK's contextvar to access the token, which works
across all request types including list_tools.
Args:
ctx: FastMCP context object (unused, kept for compatibility)
Returns:
Set of scope strings, empty set if no token or no scopes
"""
# Use MCP SDK's get_access_token() which uses contextvars
# This works for all request types, including list_tools
access_token: AccessToken | None = get_access_token()
if access_token is None:
logger.debug("No access token found in auth context (likely BasicAuth mode)")
return set()
scopes = set(access_token.scopes or [])
logger.info(f"✅ Extracted scopes from access token: {scopes}")
return scopes
def check_scopes(ctx: Context, *required_scopes: str) -> tuple[bool, set[str]]:
"""
Check if the request context has all required scopes.
Utility function for manual scope checking without decorator.
Args:
ctx: FastMCP context object
*required_scopes: Variable number of required scope strings
Returns:
Tuple of (has_all_scopes: bool, missing_scopes: set[str])
Example:
```python
async def my_tool(ctx: Context):
has_scopes, missing = check_scopes(ctx, "nc:read", "nc:write")
if not has_scopes:
# Handle missing scopes
...
```
"""
token_scopes = get_access_token_scopes(ctx)
# If no access token, assume BasicAuth mode (all operations allowed)
if not token_scopes and getattr(ctx.request_context, "access_token", None) is None:
return True, set()
required_scopes_set = set(required_scopes)
missing_scopes = required_scopes_set - token_scopes
return len(missing_scopes) == 0, missing_scopes
def get_required_scopes(func: Callable) -> list[str]:
"""
Extract required scopes from a function decorated with @require_scopes.
Args:
func: Function to check (may be decorated)
Returns:
List of required scope strings, empty list if no scopes required
Example:
```python
@require_scopes("nc:read", "nc:write")
async def my_tool():
pass
scopes = get_required_scopes(my_tool) # ["nc:read", "nc:write"]
```
"""
return getattr(func, "_required_scopes", [])
def is_jwt_token() -> bool:
"""
Check if the current access token is in JWT format.
JWT tokens have 3 parts separated by dots (header.payload.signature).
Opaque tokens are random strings without this structure.
Returns:
True if current token is JWT format, False if opaque or no token
"""
access_token: AccessToken | None = get_access_token()
if access_token is None:
logger.debug("No access token found - not JWT")
return False
# JWT tokens have exactly 2 dots (3 parts)
token_string = access_token.token
is_jwt = "." in token_string and token_string.count(".") == 2
logger.debug(f"Token format check: is_jwt={is_jwt}")
return is_jwt
def has_required_scopes(func: Callable, user_scopes: set[str]) -> bool:
"""
Check if a user has all scopes required by a function.
Used for dynamic tool filtering - determines if a tool should be visible
to a user based on their token scopes.
Args:
func: Function decorated with @require_scopes
user_scopes: Set of scopes the user possesses
Returns:
True if user has all required scopes (or no scopes required), False otherwise
Example:
```python
@require_scopes("nc:write")
async def create_note():
pass
user_scopes = {"nc:read", "nc:write"}
can_see = has_required_scopes(create_note, user_scopes) # True
limited_user_scopes = {"nc:read"}
can_see = has_required_scopes(create_note, limited_user_scopes) # False
```
"""
required = get_required_scopes(func)
# No scopes required → always allow
if not required:
return True
# Empty user_scopes but scopes required → deny
if not user_scopes:
return False
# Check if user has all required scopes
return set(required).issubset(user_scopes)
nextcloud_mcp_server/auth/token_verifier.py
+264 -13
View File
@@ -5,6 +5,8 @@ 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__)
@@ -12,22 +14,33 @@ logger = logging.getLogger(__name__)
class NextcloudTokenVerifier(TokenVerifier):
"""
Validates access tokens using Nextcloud OIDC userinfo endpoint.
Validates access tokens using JWT verification with JWKS or userinfo endpoint fallback.
This verifier:
1. Calls the userinfo endpoint with the bearer token
2. Caches successful responses to avoid repeated API calls
3. Extracts username from the 'sub' or 'preferred_username' claim
4. Optionally supports JWT validation for performance (future enhancement)
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
The userinfo endpoint validates the token and returns user claims if valid,
or returns HTTP 400/401 if the token is invalid or expired.
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,
):
"""
@@ -36,26 +49,52 @@ class NextcloudTokenVerifier(TokenVerifier):
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 requests
# 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 by calling the userinfo endpoint.
Verify a bearer token using JWT verification, introspection, or userinfo endpoint.
This method:
1. Checks the cache first for recent validations
2. Calls the userinfo endpoint if not cached
3. Returns AccessToken with username stored in metadata
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
@@ -69,13 +108,225 @@ class NextcloudTokenVerifier(TokenVerifier):
logger.debug("Token found in cache")
return cached
# Validate via userinfo endpoint
# 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
payload = jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
issuer=self.issuer,
options={
"verify_signature": True,
"verify_exp": True,
"verify_iat": True,
"verify_iss": True if self.issuer else False,
"verify_aud": False, # Skip audience validation for Bearer tokens
},
)
logger.debug(f"JWT verified successfully for user: {payload.get('sub')}")
logger.debug(f"Full JWT payload: {payload}")
# Extract username (sub claim)
username = payload.get("sub")
if not username:
logger.error("No 'sub' 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.info(f"Token introspection failed: HTTP {response.status_code}")
return None
else:
logger.warning(
f"Unexpected response from introspection: {response.status_code}"
)
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.
nextcloud_mcp_server/server/calendar.py
+17
View File
@@ -4,6 +4,7 @@ from typing import Optional
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.calendar import (
Calendar,
@@ -18,6 +19,7 @@ logger = logging.getLogger(__name__)
def configure_calendar_tools(mcp: FastMCP):
# Calendar tools
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_list_calendars(ctx: Context) -> ListCalendarsResponse:
"""List all available calendars for the user"""
client = get_client(ctx)
@@ -27,6 +29,7 @@ def configure_calendar_tools(mcp: FastMCP):
return ListCalendarsResponse(calendars=calendars, total_count=len(calendars))
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_create_event(
calendar_name: str,
title: str,
@@ -102,6 +105,7 @@ def configure_calendar_tools(mcp: FastMCP):
return await client.calendar.create_event(calendar_name, event_data)
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_list_events(
calendar_name: str,
ctx: Context,
@@ -203,6 +207,7 @@ def configure_calendar_tools(mcp: FastMCP):
return events
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_get_event(
calendar_name: str,
event_uid: str,
@@ -214,6 +219,7 @@ def configure_calendar_tools(mcp: FastMCP):
return event_data
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_update_event(
calendar_name: str,
event_uid: str,
@@ -286,6 +292,7 @@ def configure_calendar_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_delete_event(
calendar_name: str,
event_uid: str,
@@ -296,6 +303,7 @@ def configure_calendar_tools(mcp: FastMCP):
return await client.calendar.delete_event(calendar_name, event_uid)
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_create_meeting(
title: str,
date: str,
@@ -361,6 +369,7 @@ def configure_calendar_tools(mcp: FastMCP):
return await client.calendar.create_event(calendar_name, event_data)
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_get_upcoming_events(
ctx: Context,
calendar_name: str = "", # Empty = all calendars
@@ -410,6 +419,7 @@ def configure_calendar_tools(mcp: FastMCP):
return all_events[:limit]
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_find_availability(
duration_minutes: int,
ctx: Context,
@@ -489,6 +499,7 @@ def configure_calendar_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_bulk_operations(
operation: str, # "update", "delete", "move"
ctx: Context,
@@ -737,6 +748,7 @@ def configure_calendar_tools(mcp: FastMCP):
}
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_manage_calendar(
action: str, # "create", "delete", "update", "list"
ctx: Context,
@@ -805,6 +817,7 @@ def configure_calendar_tools(mcp: FastMCP):
# ============= Todo/Task Tools =============
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_list_todos(
calendar_name: str,
ctx: Context,
@@ -849,6 +862,7 @@ def configure_calendar_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_create_todo(
calendar_name: str,
summary: str,
@@ -891,6 +905,7 @@ def configure_calendar_tools(mcp: FastMCP):
return await client.calendar.create_todo(calendar_name, todo_data)
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_update_todo(
calendar_name: str,
todo_uid: str,
@@ -950,6 +965,7 @@ def configure_calendar_tools(mcp: FastMCP):
return await client.calendar.update_todo(calendar_name, todo_uid, todo_data)
@mcp.tool()
@require_scopes("nc:write")
async def nc_calendar_delete_todo(
calendar_name: str,
todo_uid: str,
@@ -969,6 +985,7 @@ def configure_calendar_tools(mcp: FastMCP):
return await client.calendar.delete_todo(calendar_name, todo_uid)
@mcp.tool()
@require_scopes("nc:read")
async def nc_calendar_search_todos(
ctx: Context,
status: Optional[str] = None,
nextcloud_mcp_server/server/contacts.py
+8
View File
@@ -2,6 +2,7 @@ import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
logger = logging.getLogger(__name__)
@@ -10,18 +11,21 @@ logger = logging.getLogger(__name__)
def configure_contacts_tools(mcp: FastMCP):
# Contacts tools
@mcp.tool()
@require_scopes("nc:read")
async def nc_contacts_list_addressbooks(ctx: Context):
"""List all addressbooks for the user."""
client = get_client(ctx)
return await client.contacts.list_addressbooks()
@mcp.tool()
@require_scopes("nc:read")
async def nc_contacts_list_contacts(ctx: Context, *, addressbook: str):
"""List all contacts in the specified addressbook."""
client = get_client(ctx)
return await client.contacts.list_contacts(addressbook=addressbook)
@mcp.tool()
@require_scopes("nc:write")
async def nc_contacts_create_addressbook(
ctx: Context, *, name: str, display_name: str
):
@@ -37,12 +41,14 @@ def configure_contacts_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_contacts_delete_addressbook(ctx: Context, *, name: str):
"""Delete an addressbook."""
client = get_client(ctx)
return await client.contacts.delete_addressbook(name=name)
@mcp.tool()
@require_scopes("nc:write")
async def nc_contacts_create_contact(
ctx: Context, *, addressbook: str, uid: str, contact_data: dict
):
@@ -59,12 +65,14 @@ def configure_contacts_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_contacts_delete_contact(ctx: Context, *, addressbook: str, uid: str):
"""Delete a contact."""
client = get_client(ctx)
return await client.contacts.delete_contact(addressbook=addressbook, uid=uid)
@mcp.tool()
@require_scopes("nc:write")
async def nc_contacts_update_contact(
ctx: Context, *, addressbook: str, uid: str, contact_data: dict, etag: str = ""
):
nextcloud_mcp_server/server/cookbook.py
+14
View File
@@ -5,6 +5,7 @@ from mcp.server.fastmcp import Context, FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.cookbook import (
Category,
@@ -70,6 +71,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_cookbook_import_recipe(url: str, ctx: Context) -> ImportRecipeResponse:
"""Import a recipe from a URL using schema.org metadata.
@@ -126,6 +128,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_list_recipes(ctx: Context) -> ListRecipesResponse:
"""Get all recipes in the database"""
client = get_client(ctx)
@@ -150,6 +153,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_get_recipe(recipe_id: int, ctx: Context) -> Recipe:
"""Get a specific recipe by its ID"""
client = get_client(ctx)
@@ -174,6 +178,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_cookbook_create_recipe(
name: str,
description: str | None = None,
@@ -252,6 +257,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_cookbook_update_recipe(
recipe_id: int,
name: str | None = None,
@@ -340,6 +346,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_cookbook_delete_recipe(
recipe_id: int, ctx: Context
) -> DeleteRecipeResponse:
@@ -374,6 +381,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_search_recipes(
query: str, ctx: Context
) -> SearchRecipesResponse:
@@ -409,6 +417,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_list_categories(ctx: Context) -> ListCategoriesResponse:
"""Get all known categories.
@@ -435,6 +444,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_get_recipes_in_category(
category: str, ctx: Context
) -> ListRecipesResponse:
@@ -470,6 +480,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_list_keywords(ctx: Context) -> ListKeywordsResponse:
"""Get all known keywords/tags"""
client = get_client(ctx)
@@ -494,6 +505,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_cookbook_get_recipes_with_keywords(
keywords: list[str], ctx: Context
) -> ListRecipesResponse:
@@ -527,6 +539,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_cookbook_set_config(
folder: str | None = None,
update_interval: int | None = None,
@@ -569,6 +582,7 @@ def configure_cookbook_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_cookbook_reindex(ctx: Context) -> ReindexResponse:
"""Trigger a rescan of all recipes into the caching database.
nextcloud_mcp_server/server/deck.py
+26
View File
@@ -3,6 +3,7 @@ from typing import Optional
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.deck import (
CardOperationResponse,
@@ -116,6 +117,7 @@ def configure_deck_tools(mcp: FastMCP):
# Read Tools (converted from resources)
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_boards(ctx: Context) -> list[DeckBoard]:
"""Get all Nextcloud Deck boards"""
client = get_client(ctx)
@@ -123,6 +125,7 @@ def configure_deck_tools(mcp: FastMCP):
return boards
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_board(ctx: Context, board_id: int) -> DeckBoard:
"""Get details of a specific Nextcloud Deck board"""
client = get_client(ctx)
@@ -130,6 +133,7 @@ def configure_deck_tools(mcp: FastMCP):
return board
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_stacks(ctx: Context, board_id: int) -> list[DeckStack]:
"""Get all stacks in a Nextcloud Deck board"""
client = get_client(ctx)
@@ -137,6 +141,7 @@ def configure_deck_tools(mcp: FastMCP):
return stacks
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_stack(ctx: Context, board_id: int, stack_id: int) -> DeckStack:
"""Get details of a specific Nextcloud Deck stack"""
client = get_client(ctx)
@@ -144,6 +149,7 @@ def configure_deck_tools(mcp: FastMCP):
return stack
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_cards(
ctx: Context, board_id: int, stack_id: int
) -> list[DeckCard]:
@@ -155,6 +161,7 @@ def configure_deck_tools(mcp: FastMCP):
return []
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_card(
ctx: Context, board_id: int, stack_id: int, card_id: int
) -> DeckCard:
@@ -164,6 +171,7 @@ def configure_deck_tools(mcp: FastMCP):
return card
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_labels(ctx: Context, board_id: int) -> list[DeckLabel]:
"""Get all labels in a Nextcloud Deck board"""
client = get_client(ctx)
@@ -171,6 +179,7 @@ def configure_deck_tools(mcp: FastMCP):
return board.labels
@mcp.tool()
@require_scopes("nc:read")
async def deck_get_label(ctx: Context, board_id: int, label_id: int) -> DeckLabel:
"""Get details of a specific Nextcloud Deck label"""
client = get_client(ctx)
@@ -180,6 +189,7 @@ def configure_deck_tools(mcp: FastMCP):
# Create/Update/Delete Tools
@mcp.tool()
@require_scopes("nc:write")
async def deck_create_board(
ctx: Context, title: str, color: str
) -> CreateBoardResponse:
@@ -196,6 +206,7 @@ def configure_deck_tools(mcp: FastMCP):
# Stack Tools
@mcp.tool()
@require_scopes("nc:write")
async def deck_create_stack(
ctx: Context, board_id: int, title: str, order: int
) -> CreateStackResponse:
@@ -211,6 +222,7 @@ def configure_deck_tools(mcp: FastMCP):
return CreateStackResponse(id=stack.id, title=stack.title, order=stack.order)
@mcp.tool()
@require_scopes("nc:write")
async def deck_update_stack(
ctx: Context,
board_id: int,
@@ -236,6 +248,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_delete_stack(
ctx: Context, board_id: int, stack_id: int
) -> StackOperationResponse:
@@ -256,6 +269,7 @@ def configure_deck_tools(mcp: FastMCP):
# Card Tools
@mcp.tool()
@require_scopes("nc:write")
async def deck_create_card(
ctx: Context,
board_id: int,
@@ -289,6 +303,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_update_card(
ctx: Context,
board_id: int,
@@ -341,6 +356,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_delete_card(
ctx: Context, board_id: int, stack_id: int, card_id: int
) -> CardOperationResponse:
@@ -362,6 +378,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_archive_card(
ctx: Context, board_id: int, stack_id: int, card_id: int
) -> CardOperationResponse:
@@ -383,6 +400,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_unarchive_card(
ctx: Context, board_id: int, stack_id: int, card_id: int
) -> CardOperationResponse:
@@ -404,6 +422,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_reorder_card(
ctx: Context,
board_id: int,
@@ -435,6 +454,7 @@ def configure_deck_tools(mcp: FastMCP):
# Label Tools
@mcp.tool()
@require_scopes("nc:write")
async def deck_create_label(
ctx: Context, board_id: int, title: str, color: str
) -> CreateLabelResponse:
@@ -450,6 +470,7 @@ def configure_deck_tools(mcp: FastMCP):
return CreateLabelResponse(id=label.id, title=label.title, color=label.color)
@mcp.tool()
@require_scopes("nc:write")
async def deck_update_label(
ctx: Context,
board_id: int,
@@ -475,6 +496,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_delete_label(
ctx: Context, board_id: int, label_id: int
) -> LabelOperationResponse:
@@ -495,6 +517,7 @@ def configure_deck_tools(mcp: FastMCP):
# Card-Label Assignment Tools
@mcp.tool()
@require_scopes("nc:write")
async def deck_assign_label_to_card(
ctx: Context, board_id: int, stack_id: int, card_id: int, label_id: int
) -> CardOperationResponse:
@@ -517,6 +540,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_remove_label_from_card(
ctx: Context, board_id: int, stack_id: int, card_id: int, label_id: int
) -> CardOperationResponse:
@@ -540,6 +564,7 @@ def configure_deck_tools(mcp: FastMCP):
# Card-User Assignment Tools
@mcp.tool()
@require_scopes("nc:write")
async def deck_assign_user_to_card(
ctx: Context, board_id: int, stack_id: int, card_id: int, user_id: str
) -> CardOperationResponse:
@@ -562,6 +587,7 @@ def configure_deck_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def deck_unassign_user_from_card(
ctx: Context, board_id: int, stack_id: int, card_id: int, user_id: str
) -> CardOperationResponse:
nextcloud_mcp_server/server/notes.py
+12 -4
View File
@@ -5,6 +5,7 @@ from mcp.server.fastmcp import Context, FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.notes import (
AppendContentResponse,
@@ -84,10 +85,11 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_notes_create_note(
title: str, content: str, category: str, ctx: Context
) -> CreateNoteResponse:
"""Create a new note"""
"""Create a new note (requires nc:write scope)"""
client = get_client(ctx)
try:
note_data = await client.notes.create_note(
@@ -129,6 +131,7 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_notes_update_note(
note_id: int,
etag: str,
@@ -137,7 +140,7 @@ def configure_notes_tools(mcp: FastMCP):
category: str | None,
ctx: Context,
) -> UpdateNoteResponse:
"""Update an existing note's title, content, or category.
"""Update an existing note's title, content, or category (requires nc:write scope).
REQUIRED: etag parameter must be provided to prevent overwriting concurrent changes.
Get the current ETag by first retrieving the note using nc_notes_get_note tool.
@@ -193,6 +196,7 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_notes_append_content(
note_id: int, content: str, ctx: Context
) -> AppendContentResponse:
@@ -242,8 +246,9 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_notes_search_notes(query: str, ctx: Context) -> SearchNotesResponse:
"""Search notes by title or content, returning only id, title, and category."""
"""Search notes by title or content, returning only id, title, and category (requires nc:read scope)."""
client = get_client(ctx)
try:
search_results_raw = await client.notes_search_notes(query=query)
@@ -287,8 +292,9 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_notes_get_note(note_id: int, ctx: Context) -> Note:
"""Get a specific note by its ID"""
"""Get a specific note by its ID (requires nc:read scope)"""
client = get_client(ctx)
try:
note_data = await client.notes.get_note(note_id)
@@ -315,6 +321,7 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_notes_get_attachment(
note_id: int, attachment_filename: str, ctx: Context
) -> dict[str, str]:
@@ -360,6 +367,7 @@ def configure_notes_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_notes_delete_note(note_id: int, ctx: Context) -> DeleteNoteResponse:
"""Delete a note permanently"""
logger.info("Deleting note %s", note_id)
nextcloud_mcp_server/server/sharing.py
+6
View File
@@ -4,6 +4,7 @@ import json
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
@@ -15,6 +16,7 @@ def configure_sharing_tools(mcp: FastMCP):
"""
@mcp.tool()
@require_scopes("nc:write")
async def nc_share_create(
path: str,
share_with: str,
@@ -53,6 +55,7 @@ def configure_sharing_tools(mcp: FastMCP):
return json.dumps(share_data, indent=2)
@mcp.tool()
@require_scopes("nc:write")
async def nc_share_delete(share_id: int, ctx: Context) -> str:
"""Delete a share by its ID.
@@ -71,6 +74,7 @@ def configure_sharing_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_share_get(share_id: int, ctx: Context) -> str:
"""Get information about a specific share.
@@ -88,6 +92,7 @@ def configure_sharing_tools(mcp: FastMCP):
return json.dumps(share_data, indent=2)
@mcp.tool()
@require_scopes("nc:write")
async def nc_share_list(
ctx: Context, path: str | None = None, shared_with_me: bool = False
) -> str:
@@ -108,6 +113,7 @@ def configure_sharing_tools(mcp: FastMCP):
return json.dumps(shares, indent=2)
@mcp.tool()
@require_scopes("nc:write")
async def nc_share_update(share_id: int, permissions: int, ctx: Context) -> str:
"""Update the permissions of an existing share.
nextcloud_mcp_server/server/tables.py
+7
View File
@@ -2,6 +2,7 @@ import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
logger = logging.getLogger(__name__)
@@ -10,18 +11,21 @@ logger = logging.getLogger(__name__)
def configure_tables_tools(mcp: FastMCP):
# Tables tools
@mcp.tool()
@require_scopes("nc:read")
async def nc_tables_list_tables(ctx: Context):
"""List all tables available to the user"""
client = get_client(ctx)
return await client.tables.list_tables()
@mcp.tool()
@require_scopes("nc:read")
async def nc_tables_get_schema(table_id: int, ctx: Context):
"""Get the schema/structure of a specific table including columns and views"""
client = get_client(ctx)
return await client.tables.get_table_schema(table_id)
@mcp.tool()
@require_scopes("nc:read")
async def nc_tables_read_table(
table_id: int,
ctx: Context,
@@ -33,6 +37,7 @@ def configure_tables_tools(mcp: FastMCP):
return await client.tables.get_table_rows(table_id, limit, offset)
@mcp.tool()
@require_scopes("nc:write")
async def nc_tables_insert_row(table_id: int, data: dict, ctx: Context):
"""Insert a new row into a table.
@@ -42,6 +47,7 @@ def configure_tables_tools(mcp: FastMCP):
return await client.tables.create_row(table_id, data)
@mcp.tool()
@require_scopes("nc:write")
async def nc_tables_update_row(row_id: int, data: dict, ctx: Context):
"""Update an existing row in a table.
@@ -51,6 +57,7 @@ def configure_tables_tools(mcp: FastMCP):
return await client.tables.update_row(row_id, data)
@mcp.tool()
@require_scopes("nc:write")
async def nc_tables_delete_row(row_id: int, ctx: Context):
"""Delete a row from a table"""
client = get_client(ctx)
nextcloud_mcp_server/server/webdav.py
+12
View File
@@ -8,6 +8,7 @@ from nextcloud_mcp_server.utils.document_parser import (
parse_document,
)
from nextcloud_mcp_server.config import is_unstructured_parsing_enabled
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models import FileInfo, SearchFilesResponse
@@ -17,6 +18,7 @@ logger = logging.getLogger(__name__)
def configure_webdav_tools(mcp: FastMCP):
# WebDAV file system tools
@mcp.tool()
@require_scopes("nc:read")
async def nc_webdav_list_directory(ctx: Context, path: str = ""):
"""List files and directories in the specified NextCloud path.
@@ -30,6 +32,7 @@ def configure_webdav_tools(mcp: FastMCP):
return await client.webdav.list_directory(path)
@mcp.tool()
@require_scopes("nc:read")
async def nc_webdav_read_file(path: str, ctx: Context):
"""Read the content of a file from NextCloud.
@@ -105,6 +108,7 @@ def configure_webdav_tools(mcp: FastMCP):
}
@mcp.tool()
@require_scopes("nc:write")
async def nc_webdav_write_file(
path: str, content: str, ctx: Context, content_type: str | None = None
):
@@ -132,6 +136,7 @@ def configure_webdav_tools(mcp: FastMCP):
return await client.webdav.write_file(path, content_bytes, content_type)
@mcp.tool()
@require_scopes("nc:write")
async def nc_webdav_create_directory(path: str, ctx: Context):
"""Create a directory in NextCloud.
@@ -145,6 +150,7 @@ def configure_webdav_tools(mcp: FastMCP):
return await client.webdav.create_directory(path)
@mcp.tool()
@require_scopes("nc:write")
async def nc_webdav_delete_resource(path: str, ctx: Context):
"""Delete a file or directory in NextCloud.
@@ -158,6 +164,7 @@ def configure_webdav_tools(mcp: FastMCP):
return await client.webdav.delete_resource(path)
@mcp.tool()
@require_scopes("nc:write")
async def nc_webdav_move_resource(
source_path: str, destination_path: str, ctx: Context, overwrite: bool = False
):
@@ -177,6 +184,7 @@ def configure_webdav_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:write")
async def nc_webdav_copy_resource(
source_path: str, destination_path: str, ctx: Context, overwrite: bool = False
):
@@ -196,6 +204,7 @@ def configure_webdav_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_webdav_search_files(
ctx: Context,
scope: str = "",
@@ -311,6 +320,7 @@ def configure_webdav_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_webdav_find_by_name(
pattern: str, ctx: Context, scope: str = "", limit: int | None = None
) -> SearchFilesResponse:
@@ -337,6 +347,7 @@ def configure_webdav_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_webdav_find_by_type(
mime_type: str, ctx: Context, scope: str = "", limit: int | None = None
) -> SearchFilesResponse:
@@ -363,6 +374,7 @@ def configure_webdav_tools(mcp: FastMCP):
)
@mcp.tool()
@require_scopes("nc:read")
async def nc_webdav_list_favorites(
ctx: Context, scope: str = "", limit: int | None = None
) -> SearchFilesResponse: