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

Merge remote-tracking branch 'origin/master' into feature/user-api

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-14 23:43:03 +02:00
parent 961f23b5ea 8652684466
commit 898c2e72ae
79 changed files with 7312 additions and 1003 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/app.py
+494 -29
View File
@@ -1,62 +1,373 @@
import click
import logging
import uvicorn
import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from contextlib import AsyncExitStack, asynccontextmanager
from dataclasses import dataclass
import click
import uvicorn
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 mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.config import setup_logging
from nextcloud_mcp_server.auth import NextcloudTokenVerifier, load_or_register_client
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import setup_logging
from nextcloud_mcp_server.context import get_client as get_nextcloud_client
from nextcloud_mcp_server.server import (
configure_calendar_tools,
configure_contacts_tools,
configure_deck_tools,
configure_notes_tools,
configure_tables_tools,
configure_webdav_tools,
configure_deck_tools,
)
logger = logging.getLogger(__name__)
def validate_pkce_support(discovery: dict, discovery_url: str) -> None:
"""
Validate that the OIDC provider properly advertises PKCE support.
According to RFC 8414, if code_challenge_methods_supported is absent,
it means the authorization server does not support PKCE.
MCP clients require PKCE with S256 and will refuse to connect if this
field is missing or doesn't include S256.
"""
code_challenge_methods = discovery.get("code_challenge_methods_supported")
if code_challenge_methods is None:
click.echo("=" * 80, err=True)
click.echo(
"ERROR: OIDC CONFIGURATION ERROR - Missing PKCE Support Advertisement",
err=True,
)
click.echo("=" * 80, err=True)
click.echo(f"Discovery URL: {discovery_url}", err=True)
click.echo("", err=True)
click.echo(
"The OIDC discovery document is missing 'code_challenge_methods_supported'.",
err=True,
)
click.echo(
"According to RFC 8414, this means the server does NOT support PKCE.",
err=True,
)
click.echo("", err=True)
click.echo("⚠️ MCP clients (like Claude Code) WILL REJECT this provider!")
click.echo("", err=True)
click.echo("How to fix:", err=True)
click.echo(
" 1. Ensure PKCE is enabled in Nextcloud OIDC app settings", err=True
)
click.echo(
" 2. Update the OIDC app to advertise PKCE support in discovery", err=True
)
click.echo(" 3. See: RFC 8414 Section 2 (Authorization Server Metadata)")
click.echo("=" * 80, err=True)
click.echo("", err=True)
return
if "S256" not in code_challenge_methods:
click.echo("=" * 80, err=True)
click.echo(
"WARNING: OIDC CONFIGURATION WARNING - S256 Challenge Method Not Advertised",
err=True,
)
click.echo("=" * 80, err=True)
click.echo(f"Discovery URL: {discovery_url}", err=True)
click.echo(f"Advertised methods: {code_challenge_methods}", err=True)
click.echo("", err=True)
click.echo("MCP specification requires S256 code challenge method.", err=True)
click.echo("Some clients may reject this provider.", err=True)
click.echo("=" * 80, err=True)
click.echo("", err=True)
return
click.echo(f"✓ PKCE support validated: {code_challenge_methods}")
@dataclass
class AppContext:
"""Application context for BasicAuth mode."""
client: NextcloudClient
@dataclass
class OAuthAppContext:
"""Application context for OAuth mode."""
nextcloud_host: str
token_verifier: NextcloudTokenVerifier
def is_oauth_mode() -> bool:
"""
Determine if OAuth mode should be used.
OAuth mode is enabled when:
- NEXTCLOUD_USERNAME and NEXTCLOUD_PASSWORD are NOT set
- Or explicitly enabled via configuration
Returns:
True if OAuth mode, False if BasicAuth mode
"""
username = os.getenv("NEXTCLOUD_USERNAME")
password = os.getenv("NEXTCLOUD_PASSWORD")
# If both username and password are set, use BasicAuth
if username and password:
logger.info(
"BasicAuth mode detected (NEXTCLOUD_USERNAME and NEXTCLOUD_PASSWORD set)"
)
return False
logger.info("OAuth mode detected (NEXTCLOUD_USERNAME/PASSWORD not set)")
return True
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle with type-safe context"""
# Initialize on startup
logging.info("Creating Nextcloud client")
async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
"""
Manage application lifecycle for BasicAuth mode.
Creates a single Nextcloud client with basic authentication
that is shared across all requests.
"""
logger.info("Starting MCP server in BasicAuth mode")
logger.info("Creating Nextcloud client with BasicAuth")
client = NextcloudClient.from_env()
logging.info("Client initialization wait complete.")
logger.info("Client initialization complete")
try:
yield AppContext(client=client)
finally:
# Cleanup on shutdown
logger.info("Shutting down BasicAuth mode")
await client.close()
@asynccontextmanager
async def app_lifespan_oauth(server: FastMCP) -> AsyncIterator[OAuthAppContext]:
"""
Manage application lifecycle for OAuth mode.
Initializes OAuth client registration and token verifier.
Does NOT create a Nextcloud client - clients are created per-request.
"""
logger.info("Starting MCP server in OAuth mode")
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
raise ValueError("NEXTCLOUD_HOST environment variable is required")
nextcloud_host = nextcloud_host.rstrip("/")
# Get OAuth discovery endpoint
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
try:
# Fetch OIDC discovery
import httpx
async with httpx.AsyncClient() as client:
response = await client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
logger.info(f"OIDC discovery successful: {discovery_url}")
# Extract endpoints
userinfo_uri = discovery["userinfo_endpoint"]
registration_endpoint = discovery.get("registration_endpoint")
logger.info(f"Userinfo endpoint: {userinfo_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"
)
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
token_verifier = NextcloudTokenVerifier(
nextcloud_host=nextcloud_host, userinfo_uri=userinfo_uri
)
logger.info("OAuth initialization complete")
try:
yield OAuthAppContext(
nextcloud_host=nextcloud_host, token_verifier=token_verifier
)
finally:
logger.info("Shutting down OAuth mode")
await token_verifier.close()
except Exception as e:
logger.error(f"Failed to initialize OAuth mode: {e}")
raise
async def setup_oauth_config():
"""
Setup OAuth configuration by performing OIDC discovery and client registration.
This is done synchronously before FastMCP initialization because FastMCP
requires token_verifier at construction time.
Returns:
Tuple of (nextcloud_host, token_verifier, auth_settings)
"""
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
raise ValueError(
"NEXTCLOUD_HOST environment variable is required for OAuth mode"
)
nextcloud_host = nextcloud_host.rstrip("/")
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
logger.info(f"Performing OIDC discovery: {discovery_url}")
# Fetch OIDC discovery
import httpx
async with httpx.AsyncClient() as client:
response = await client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
logger.info("OIDC discovery successful")
# Validate PKCE support
validate_pkce_support(discovery, discovery_url)
# Extract endpoints
issuer = discovery["issuer"]
userinfo_uri = discovery["userinfo_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)
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}")
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]}...")
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
token_verifier = NextcloudTokenVerifier(
nextcloud_host=nextcloud_host, userinfo_uri=userinfo_uri
)
# Create auth settings
mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
auth_settings = AuthSettings(
issuer_url=AnyHttpUrl(issuer),
resource_server_url=AnyHttpUrl(mcp_server_url),
required_scopes=["openid", "profile"],
)
logger.info("OAuth configuration complete")
return nextcloud_host, token_verifier, auth_settings
def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
setup_logging()
# Create an MCP server
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan)
# Determine authentication mode
oauth_enabled = is_oauth_mode()
if oauth_enabled:
logger.info("Configuring MCP server for OAuth mode")
# Asynchronously get the OAuth configuration
import asyncio
nextcloud_host, token_verifier, auth_settings = asyncio.run(
setup_oauth_config()
)
mcp = FastMCP(
"Nextcloud MCP",
lifespan=app_lifespan_oauth,
token_verifier=token_verifier,
auth=auth_settings,
)
else:
logger.info("Configuring MCP server for BasicAuth mode")
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan_basic)
@mcp.resource("nc://capabilities")
async def nc_get_capabilities():
"""Get the Nextcloud Host capabilities"""
ctx: Context = (
mcp.get_context()
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
client: NextcloudClient = ctx.request_context.lifespan_context.client
ctx: Context = mcp.get_context()
client = get_nextcloud_client(ctx)
return await client.capabilities()
# Define available apps and their configuration functions
@@ -83,24 +394,49 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
f"Unknown app: {app_name}. Available apps: {list(available_apps.keys())}"
)
mcp_app = mcp.sse_app() if transport == "sse" else mcp.streamable_http_app()
if transport == "sse":
mcp_app = mcp.sse_app()
lifespan = None
elif transport in ("http", "streamable-http"):
mcp_app = mcp.streamable_http_app()
app = Starlette(routes=[Mount("/", app=mcp_app)])
@asynccontextmanager
async def lifespan(app: Starlette):
async with AsyncExitStack() as stack:
await stack.enter_async_context(mcp.session_manager.run())
yield
app = Starlette(routes=[Mount("/", app=mcp_app)], lifespan=lifespan)
return app
@click.command()
@click.option("--host", "-h", default="127.0.0.1")
@click.option("--port", "-p", type=int, default=8000)
@click.option("--workers", "-w", type=int, default=None)
@click.option("--reload", "-r", is_flag=True)
@click.option(
"--log-level",
type=click.Choice(["critical", "error", "warning", "info", "debug", "trace"]),
"--host", "-h", default="127.0.0.1", show_default=True, help="Server host"
)
@click.option(
"--transport", "-t", default="sse", type=click.Choice(["sse", "streamable-http"])
"--port", "-p", type=int, default=8000, show_default=True, help="Server port"
)
@click.option(
"--workers", "-w", type=int, default=None, help="Number of worker processes"
)
@click.option("--reload", "-r", is_flag=True, help="Enable auto-reload")
@click.option(
"--log-level",
"-l",
default="info",
show_default=True,
type=click.Choice(["critical", "error", "warning", "info", "debug", "trace"]),
help="Logging level",
)
@click.option(
"--transport",
"-t",
default="sse",
show_default=True,
type=click.Choice(["sse", "streamable-http", "http"]),
help="MCP transport protocol",
)
@click.option(
"--enable-app",
@@ -109,6 +445,35 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
type=click.Choice(["notes", "tables", "webdav", "calendar", "contacts", "deck"]),
help="Enable specific Nextcloud app APIs. Can be specified multiple times. If not specified, all apps are enabled.",
)
@click.option(
"--oauth/--no-oauth",
default=None,
help="Force OAuth mode (if enabled) or BasicAuth mode (if disabled). By default, auto-detected based on environment variables.",
)
@click.option(
"--oauth-client-id",
envvar="NEXTCLOUD_OIDC_CLIENT_ID",
help="OAuth client ID (can also use NEXTCLOUD_OIDC_CLIENT_ID env var)",
)
@click.option(
"--oauth-client-secret",
envvar="NEXTCLOUD_OIDC_CLIENT_SECRET",
help="OAuth client secret (can also use NEXTCLOUD_OIDC_CLIENT_SECRET env var)",
)
@click.option(
"--oauth-storage-path",
envvar="NEXTCLOUD_OIDC_CLIENT_STORAGE",
default=".nextcloud_oauth_client.json",
show_default=True,
help="Path to store OAuth client credentials (can also use NEXTCLOUD_OIDC_CLIENT_STORAGE env var)",
)
@click.option(
"--mcp-server-url",
envvar="NEXTCLOUD_MCP_SERVER_URL",
default="http://localhost:8000",
show_default=True,
help="MCP server URL for OAuth callbacks (can also use NEXTCLOUD_MCP_SERVER_URL env var)",
)
def run(
host: str,
port: int,
@@ -117,7 +482,107 @@ def run(
log_level: str,
transport: str,
enable_app: tuple[str, ...],
oauth: bool | None,
oauth_client_id: str | None,
oauth_client_secret: str | None,
oauth_storage_path: str,
mcp_server_url: str,
):
"""
Run the Nextcloud MCP server.
\b
Authentication Modes:
- BasicAuth: Set NEXTCLOUD_USERNAME and NEXTCLOUD_PASSWORD
- OAuth: Leave USERNAME/PASSWORD unset (requires OIDC app enabled)
\b
Examples:
# BasicAuth mode (legacy)
$ nextcloud-mcp-server --host 0.0.0.0 --port 8000
# OAuth mode with auto-registration
$ nextcloud-mcp-server --oauth
# OAuth mode with pre-configured client
$ nextcloud-mcp-server --oauth --oauth-client-id=xxx --oauth-client-secret=yyy
"""
# Set OAuth env vars from CLI options if provided
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 mcp_server_url:
os.environ["NEXTCLOUD_MCP_SERVER_URL"] = mcp_server_url
# Force OAuth mode if explicitly requested
if oauth is True:
# Clear username/password to force OAuth mode
if "NEXTCLOUD_USERNAME" in os.environ:
click.echo(
"Warning: --oauth flag set, ignoring NEXTCLOUD_USERNAME", err=True
)
del os.environ["NEXTCLOUD_USERNAME"]
if "NEXTCLOUD_PASSWORD" in os.environ:
click.echo(
"Warning: --oauth flag set, ignoring NEXTCLOUD_PASSWORD", err=True
)
del os.environ["NEXTCLOUD_PASSWORD"]
# Validate OAuth configuration
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
raise click.ClickException(
"OAuth mode requires NEXTCLOUD_HOST environment variable to be set"
)
# Check if we have client credentials OR if dynamic registration is possible
has_client_creds = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID") and os.getenv(
"NEXTCLOUD_OIDC_CLIENT_SECRET"
)
if not has_client_creds:
# No client credentials - will attempt dynamic registration
# Show helpful message before server starts
click.echo("", err=True)
click.echo("OAuth Configuration:", err=True)
click.echo(" Mode: Dynamic Client Registration", err=True)
click.echo(" Host: " + nextcloud_host, err=True)
click.echo(
" Storage: "
+ os.getenv(
"NEXTCLOUD_OIDC_CLIENT_STORAGE", ".nextcloud_oauth_client.json"
),
err=True,
)
click.echo("", err=True)
click.echo(
"Note: Make sure 'Dynamic Client Registration' is enabled", err=True
)
click.echo(" in your Nextcloud OIDC app settings.", err=True)
click.echo("", err=True)
else:
click.echo("", err=True)
click.echo("OAuth Configuration:", err=True)
click.echo(" Mode: Pre-configured Client", err=True)
click.echo(" Host: " + nextcloud_host, err=True)
click.echo(
" Client ID: "
+ os.getenv("NEXTCLOUD_OIDC_CLIENT_ID", "")[:16]
+ "...",
err=True,
)
click.echo("", err=True)
elif oauth is False:
# Force BasicAuth mode - verify credentials exist
if not os.getenv("NEXTCLOUD_USERNAME") or not os.getenv("NEXTCLOUD_PASSWORD"):
raise click.ClickException(
"--no-oauth flag set but NEXTCLOUD_USERNAME or NEXTCLOUD_PASSWORD not set"
)
enabled_apps = list(enable_app) if enable_app else None
if reload or workers:
nextcloud_mcp_server/auth/__init__.py
+14
View File
@@ -0,0 +1,14 @@
"""OAuth authentication components for Nextcloud MCP server."""
from .bearer_auth import BearerAuth
from .client_registration import load_or_register_client, register_client
from .context_helper import get_client_from_context
from .token_verifier import NextcloudTokenVerifier
__all__ = [
"BearerAuth",
"NextcloudTokenVerifier",
"register_client",
"load_or_register_client",
"get_client_from_context",
]
nextcloud_mcp_server/auth/bearer_auth.py
+34
View File
@@ -0,0 +1,34 @@
"""Bearer token authentication for httpx."""
from httpx import Auth, Request
class BearerAuth(Auth):
"""
Bearer token authentication flow for httpx.
This auth class adds the Authorization: Bearer <token> header
to all outgoing requests.
"""
def __init__(self, token: str):
"""
Initialize bearer authentication.
Args:
token: The bearer token to use for authentication
"""
self.token = token
def auth_flow(self, request: Request):
"""
Add Authorization header to the request.
Args:
request: The outgoing HTTP request
Yields:
The modified request with Authorization header
"""
request.headers["Authorization"] = f"Bearer {self.token}"
yield request
nextcloud_mcp_server/auth/client_registration.py
+260
View File
@@ -0,0 +1,260 @@
"""Dynamic client registration for Nextcloud OIDC."""
import json
import logging
import os
import time
from pathlib import Path
from typing import Any
import httpx
logger = logging.getLogger(__name__)
class ClientInfo:
"""Client registration information."""
def __init__(
self,
client_id: str,
client_secret: str,
client_id_issued_at: int,
client_secret_expires_at: int,
redirect_uris: list[str],
):
self.client_id = client_id
self.client_secret = client_secret
self.client_id_issued_at = client_id_issued_at
self.client_secret_expires_at = client_secret_expires_at
self.redirect_uris = redirect_uris
@property
def is_expired(self) -> bool:
"""Check if the client has expired."""
return time.time() >= self.client_secret_expires_at
@property
def expires_soon(self) -> bool:
"""Check if client expires within 5 minutes."""
return time.time() >= (self.client_secret_expires_at - 300)
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for storage."""
return {
"client_id": self.client_id,
"client_secret": self.client_secret,
"client_id_issued_at": self.client_id_issued_at,
"client_secret_expires_at": self.client_secret_expires_at,
"redirect_uris": self.redirect_uris,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ClientInfo":
"""Create from dictionary."""
return cls(
client_id=data["client_id"],
client_secret=data["client_secret"],
client_id_issued_at=data["client_id_issued_at"],
client_secret_expires_at=data["client_secret_expires_at"],
redirect_uris=data["redirect_uris"],
)
async def register_client(
nextcloud_url: str,
registration_endpoint: str,
client_name: str = "Nextcloud MCP Server",
redirect_uris: list[str] | None = None,
scopes: str = "openid profile email",
) -> ClientInfo:
"""
Register a new OAuth client with Nextcloud OIDC using dynamic client registration.
Args:
nextcloud_url: Base URL of the Nextcloud instance
registration_endpoint: Full URL to the registration endpoint
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
Returns:
ClientInfo with registration details
Raises:
httpx.HTTPStatusError: If registration fails
ValueError: If response is invalid
"""
if redirect_uris is None:
redirect_uris = ["http://localhost:8000/oauth/callback"]
client_metadata = {
"client_name": client_name,
"redirect_uris": redirect_uris,
"token_endpoint_auth_method": "client_secret_post",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"scope": scopes,
}
logger.info(f"Registering OAuth client with Nextcloud: {client_name}")
logger.debug(f"Registration endpoint: {registration_endpoint}")
async with httpx.AsyncClient(timeout=30.0) as client:
try:
response = await client.post(
registration_endpoint,
json=client_metadata,
headers={"Content-Type": "application/json"},
)
response.raise_for_status()
client_info = response.json()
logger.info(
f"Successfully registered client: {client_info.get('client_id')}"
)
logger.info(
f"Client expires at: {client_info.get('client_secret_expires_at')} "
f"(in {client_info.get('client_secret_expires_at', 0) - int(time.time())} seconds)"
)
return ClientInfo(
client_id=client_info["client_id"],
client_secret=client_info["client_secret"],
client_id_issued_at=client_info.get(
"client_id_issued_at", int(time.time())
),
client_secret_expires_at=client_info.get(
"client_secret_expires_at", int(time.time()) + 3600
),
redirect_uris=client_info.get("redirect_uris", redirect_uris),
)
except httpx.HTTPStatusError as e:
logger.error(f"Failed to register client: HTTP {e.response.status_code}")
logger.error(f"Response: {e.response.text}")
raise
except KeyError as e:
logger.error(f"Invalid response from registration endpoint: missing {e}")
raise ValueError(f"Invalid registration response: missing {e}")
def load_client_from_file(storage_path: Path) -> ClientInfo | None:
"""
Load client credentials from storage file.
Args:
storage_path: Path to the JSON file containing client credentials
Returns:
ClientInfo if file exists and is valid, None otherwise
"""
if not storage_path.exists():
logger.debug(f"Client storage file not found: {storage_path}")
return None
try:
with open(storage_path, "r") as f:
data = json.load(f)
client_info = ClientInfo.from_dict(data)
if client_info.is_expired:
logger.warning(
f"Stored client has expired (expired at {client_info.client_secret_expires_at})"
)
return None
logger.info(f"Loaded client from storage: {client_info.client_id[:16]}...")
if client_info.expires_soon:
logger.warning("Client expires soon (within 5 minutes)")
return client_info
except (json.JSONDecodeError, KeyError, ValueError) as e:
logger.error(f"Failed to load client from file: {e}")
return None
def save_client_to_file(client_info: ClientInfo, storage_path: Path):
"""
Save client credentials to storage file.
Args:
client_info: Client information to save
storage_path: Path to save the JSON file
Raises:
OSError: If file cannot be written
"""
try:
# Create directory if it doesn't exist
storage_path.parent.mkdir(parents=True, exist_ok=True)
# Write client info
with open(storage_path, "w") as f:
json.dump(client_info.to_dict(), f, indent=2)
# Set restrictive permissions (owner read/write only)
os.chmod(storage_path, 0o600)
logger.info(f"Saved client credentials to {storage_path}")
except OSError as e:
logger.error(f"Failed to save client credentials: {e}")
raise
async def load_or_register_client(
nextcloud_url: str,
registration_endpoint: str,
storage_path: str | Path,
client_name: str = "Nextcloud MCP Server",
redirect_uris: list[str] | None = None,
force_register: bool = True,
) -> ClientInfo:
"""
Load client from storage or register a new one if not found/expired.
This function:
1. Checks for existing client credentials in storage
2. Validates the credentials are not expired
3. Registers a new client if needed
4. Saves the new client credentials
Args:
nextcloud_url: Base URL of the Nextcloud instance
registration_endpoint: Full URL to the registration endpoint
storage_path: Path to store client credentials
client_name: Name of the client application
redirect_uris: List of redirect URIs
force_register: Force registration even if valid credentials exist
Returns:
ClientInfo with valid credentials
Raises:
httpx.HTTPStatusError: If registration fails
ValueError: If response is invalid
"""
storage_path = Path(storage_path)
# Try to load existing client unless forced to register
if not force_register:
client_info = load_client_from_file(storage_path)
if client_info:
return client_info
# Register new client
logger.info("Registering new OAuth client...")
client_info = await register_client(
nextcloud_url=nextcloud_url,
registration_endpoint=registration_endpoint,
client_name=client_name,
redirect_uris=redirect_uris,
)
# Save to storage
save_client_to_file(client_info, storage_path)
return client_info
nextcloud_mcp_server/auth/context_helper.py
+65
View File
@@ -0,0 +1,65 @@
"""Helper functions for extracting OAuth context from MCP requests."""
import logging
from mcp.server.auth.provider import AccessToken
from mcp.server.fastmcp import Context
from ..client import NextcloudClient
logger = logging.getLogger(__name__)
def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
"""
Extract authenticated user context from MCP request and create NextcloudClient.
This function retrieves the OAuth access token from the MCP context,
extracts the username from the token's resource field (where we stored it
during token verification), and creates a NextcloudClient with bearer auth.
Args:
ctx: MCP request context containing session info
base_url: Nextcloud base URL
Returns:
NextcloudClient configured with bearer token auth
Raises:
AttributeError: If context doesn't contain expected OAuth session data
ValueError: If username cannot be extracted from token
"""
try:
# In Starlette with FastMCP OAuth, the authenticated user info is stored in request.user
# The FastMCP auth middleware sets request.user to an AuthenticatedUser object
# which contains the access_token
if hasattr(ctx.request_context.request, "user") and hasattr(
ctx.request_context.request.user, "access_token"
):
access_token: AccessToken = ctx.request_context.request.user.access_token
logger.debug("Retrieved access token from request.user for OAuth request")
else:
logger.error(
"OAuth authentication failed: No access token found in request"
)
raise AttributeError("No access token found in OAuth request context")
# Extract username from resource field (RFC 8707)
# We stored the username here during token verification
username = access_token.resource
if not username:
logger.error("No username found in access token resource field")
raise ValueError("Username not available in OAuth token context")
logger.debug(f"Creating OAuth NextcloudClient for user: {username}")
# Create client with bearer token
return NextcloudClient.from_token(
base_url=base_url, token=access_token.token, username=username
)
except AttributeError as e:
logger.error(f"Failed to extract OAuth context: {e}")
logger.error("This may indicate the server is not running in OAuth mode")
raise
nextcloud_mcp_server/auth/token_verifier.py
+207
View File
@@ -0,0 +1,207 @@
"""Token verification using Nextcloud OIDC userinfo endpoint."""
import logging
import time
from typing import Any
import httpx
from mcp.server.auth.provider import AccessToken, TokenVerifier
logger = logging.getLogger(__name__)
class NextcloudTokenVerifier(TokenVerifier):
"""
Validates access tokens using Nextcloud OIDC userinfo endpoint.
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)
The userinfo endpoint validates the token and returns user claims if valid,
or returns HTTP 400/401 if the token is invalid or expired.
"""
def __init__(
self,
nextcloud_host: str,
userinfo_uri: str,
cache_ttl: int = 3600,
):
"""
Initialize the token verifier.
Args:
nextcloud_host: Base URL of the Nextcloud instance (e.g., https://cloud.example.com)
userinfo_uri: Full URL to the userinfo endpoint
cache_ttl: Time-to-live for cached tokens in seconds (default: 3600)
"""
self.nextcloud_host = nextcloud_host.rstrip("/")
self.userinfo_uri = userinfo_uri
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
self._client = httpx.AsyncClient(timeout=10.0)
async def verify_token(self, token: str) -> AccessToken | None:
"""
Verify a bearer token by calling the 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
Args:
token: The bearer token to verify
Returns:
AccessToken if valid, None if invalid or expired
"""
# Check cache first
cached = self._get_cached_token(token)
if cached:
logger.debug("Token found in cache")
return cached
# Validate via userinfo endpoint
try:
return await self._verify_via_userinfo(token)
except Exception as e:
logger.warning(f"Token verification failed: {e}")
return None
async def _verify_via_userinfo(self, token: str) -> AccessToken | None:
"""
Validate token by calling the userinfo endpoint.
Args:
token: The bearer token to verify
Returns:
AccessToken if valid, None otherwise
"""
try:
response = await self._client.get(
self.userinfo_uri, headers={"Authorization": f"Bearer {token}"}
)
if response.status_code == 200:
userinfo = response.json()
logger.debug(
f"Token validated successfully for user: {userinfo.get('sub')}"
)
# Cache the result
expiry = time.time() + self.cache_ttl
self._token_cache[token] = (userinfo, expiry)
# Create AccessToken with username in resource field (workaround for MCP SDK)
username = userinfo.get("sub") or userinfo.get("preferred_username")
if not username:
logger.error("No username found in userinfo response")
return None
return AccessToken(
token=token,
client_id="", # Not available from userinfo
scopes=self._extract_scopes(userinfo),
expires_at=int(expiry),
resource=username, # Store username in resource field (RFC 8707)
)
elif response.status_code in (400, 401, 403):
logger.info(f"Token validation failed: HTTP {response.status_code}")
return None
else:
logger.warning(
f"Unexpected response from userinfo: {response.status_code}"
)
return None
except httpx.TimeoutException:
logger.error("Timeout while validating token via userinfo endpoint")
return None
except httpx.RequestError as e:
logger.error(f"Network error while validating token: {e}")
return None
except Exception as e:
logger.error(f"Unexpected error during token validation: {e}")
return None
def _get_cached_token(self, token: str) -> AccessToken | None:
"""
Retrieve a token from cache if not expired.
Args:
token: The bearer token to look up
Returns:
AccessToken if cached and valid, None otherwise
"""
if token not in self._token_cache:
return None
userinfo, expiry = self._token_cache[token]
# Check if expired
if time.time() >= expiry:
logger.debug("Cached token expired, removing from cache")
del self._token_cache[token]
return None
# Return cached AccessToken
username = userinfo.get("sub") or userinfo.get("preferred_username")
return AccessToken(
token=token,
client_id="",
scopes=self._extract_scopes(userinfo),
expires_at=int(expiry),
resource=username,
)
def _extract_scopes(self, userinfo: dict[str, Any]) -> list[str]:
"""
Extract scopes from userinfo response.
Since the userinfo response doesn't include the original scopes,
we infer them from the claims present in the response.
Args:
userinfo: The userinfo response dictionary
Returns:
List of inferred scopes
"""
scopes = ["openid"] # Always present
if "email" in userinfo:
scopes.append("email")
if any(
key in userinfo for key in ["name", "given_name", "family_name", "picture"]
):
scopes.append("profile")
if "roles" in userinfo:
scopes.append("roles")
if "groups" in userinfo:
scopes.append("groups")
return scopes
def clear_cache(self):
"""Clear the token cache."""
self._token_cache.clear()
logger.debug("Token cache cleared")
async def close(self):
"""Cleanup resources."""
await self._client.aclose()
logger.debug("Token verifier closed")
nextcloud_mcp_server/client/__init__.py
+19 -2
View File
@@ -2,13 +2,13 @@ import logging
import os
from httpx import (
AsyncBaseTransport,
AsyncClient,
AsyncHTTPTransport,
Auth,
BasicAuth,
Request,
Response,
AsyncBaseTransport,
AsyncHTTPTransport,
)
from ..controllers.notes_search import NotesSearchController
@@ -87,6 +87,23 @@ class NextcloudClient:
# Pass username to constructor
return cls(base_url=host, username=username, auth=BasicAuth(username, password))
@classmethod
def from_token(cls, base_url: str, token: str, username: str):
"""Create NextcloudClient with OAuth bearer token.
Args:
base_url: Nextcloud base URL
token: OAuth access token
username: Nextcloud username
Returns:
NextcloudClient configured with bearer token authentication
"""
from ..auth import BearerAuth
logger.info(f"Creating NC Client for user '{username}' using OAuth token")
return cls(base_url=base_url, username=username, auth=BearerAuth(token))
async def capabilities(self):
response = await self._client.get(
"/ocs/v2.php/cloud/capabilities",
nextcloud_mcp_server/client/base.py
+4 -4
View File
@@ -1,11 +1,11 @@
"""Base client for Nextcloud operations with shared authentication."""
import logging
from abc import ABC
from functools import wraps
import time
from httpx import HTTPStatusError, codes, RequestError, AsyncClient
from abc import ABC
from functools import wraps
from httpx import AsyncClient, HTTPStatusError, RequestError, codes
logger = logging.getLogger(__name__)
nextcloud_mcp_server/client/contacts.py
+3 -1
View File
@@ -1,10 +1,12 @@
"""CardDAV client for NextCloud contacts operations."""
import logging
from .base import BaseNextcloudClient
import xml.etree.ElementTree as ET
from pythonvCard4.vcard import Contact
from .base import BaseNextcloudClient
logger = logging.getLogger(__name__)
nextcloud_mcp_server/client/deck.py
+6 -6
View File
@@ -1,16 +1,16 @@
from typing import List, Optional, Dict, Any
from typing import Any, Dict, List, Optional
from nextcloud_mcp_server.client.base import BaseNextcloudClient
from nextcloud_mcp_server.models.deck import (
DeckBoard,
DeckStack,
DeckCard,
DeckLabel,
DeckACL,
DeckAttachment,
DeckBoard,
DeckCard,
DeckComment,
DeckSession,
DeckConfig,
DeckLabel,
DeckSession,
DeckStack,
)
nextcloud_mcp_server/context.py
+51
View File
@@ -0,0 +1,51 @@
"""Helper functions for accessing context in MCP tools."""
from mcp.server.fastmcp import Context
from nextcloud_mcp_server.client import NextcloudClient
def get_client(ctx: Context) -> NextcloudClient:
"""
Get the appropriate Nextcloud client based on authentication mode.
In BasicAuth mode, returns the shared client from lifespan context.
In OAuth mode, creates a new client per-request using the OAuth context.
This function automatically detects the authentication mode by checking
the type of the lifespan context.
Args:
ctx: MCP request context
Returns:
NextcloudClient configured for the current authentication mode
Raises:
AttributeError: If context doesn't contain expected data
Example:
```python
@mcp.tool()
async def my_tool(ctx: Context):
client = get_client(ctx)
return await client.capabilities()
```
"""
lifespan_ctx = ctx.request_context.lifespan_context
# Try BasicAuth mode first (has 'client' attribute)
if hasattr(lifespan_ctx, "client"):
return lifespan_ctx.client
# OAuth mode (has 'nextcloud_host' attribute)
if hasattr(lifespan_ctx, "nextcloud_host"):
from nextcloud_mcp_server.auth import get_client_from_context
return get_client_from_context(ctx, lifespan_ctx.nextcloud_host)
# Unknown context type
raise AttributeError(
f"Lifespan context does not have 'client' or 'nextcloud_host' attribute. "
f"Type: {type(lifespan_ctx)}"
)
nextcloud_mcp_server/models/__init__.py
+36 -40
View File
@@ -1,41 +1,25 @@
"""Pydantic models for structured MCP server responses."""
# Base models
from .base import (
BaseResponse,
IdResponse,
StatusResponse,
)
# Notes models
from .notes import (
Note,
NoteSearchResult,
NotesSettings,
CreateNoteResponse,
UpdateNoteResponse,
DeleteNoteResponse,
AppendContentResponse,
SearchNotesResponse,
)
from .base import BaseResponse, IdResponse, StatusResponse
# Calendar models
from .calendar import (
AvailabilitySlot,
BulkOperationResponse,
BulkOperationResult,
Calendar,
CalendarEvent,
CalendarEventSummary,
CreateEventResponse,
UpdateEventResponse,
DeleteEventResponse,
ListEventsResponse,
ListCalendarsResponse,
AvailabilitySlot,
FindAvailabilityResponse,
BulkOperationResult,
BulkOperationResponse,
CreateMeetingResponse,
UpcomingEventsResponse,
DeleteEventResponse,
FindAvailabilityResponse,
ListCalendarsResponse,
ListEventsResponse,
ManageCalendarResponse,
UpcomingEventsResponse,
UpdateEventResponse,
)
# Contacts models
@@ -43,38 +27,50 @@ from .contacts import (
AddressBook,
Contact,
ContactField,
CreateAddressBookResponse,
CreateContactResponse,
DeleteAddressBookResponse,
DeleteContactResponse,
ListAddressBooksResponse,
ListContactsResponse,
CreateContactResponse,
UpdateContactResponse,
DeleteContactResponse,
CreateAddressBookResponse,
DeleteAddressBookResponse,
)
# Notes models
from .notes import (
AppendContentResponse,
CreateNoteResponse,
DeleteNoteResponse,
Note,
NoteSearchResult,
NotesSettings,
SearchNotesResponse,
UpdateNoteResponse,
)
# Tables models
from .tables import (
CreateRowResponse,
DeleteRowResponse,
GetSchemaResponse,
ListTablesResponse,
ReadTableResponse,
Table,
TableColumn,
TableRow,
TableView,
TableSchema,
ListTablesResponse,
GetSchemaResponse,
ReadTableResponse,
CreateRowResponse,
TableView,
UpdateRowResponse,
DeleteRowResponse,
)
# WebDAV models
from .webdav import (
FileInfo,
DirectoryListing,
ReadFileResponse,
WriteFileResponse,
CreateDirectoryResponse,
DeleteResourceResponse,
DirectoryListing,
FileInfo,
ReadFileResponse,
WriteFileResponse,
)
__all__ = [
nextcloud_mcp_server/models/deck.py
+1 -1
View File
@@ -1,5 +1,5 @@
from datetime import datetime
from typing import List, Optional, Dict, Any, Union
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, Field, field_validator
nextcloud_mcp_server/server/__init__.py
+2 -2
View File
@@ -1,9 +1,9 @@
from .calendar import configure_calendar_tools
from .contacts import configure_contacts_tools
from .deck import configure_deck_tools
from .notes import configure_notes_tools
from .tables import configure_tables_tools
from .webdav import configure_webdav_tools
from .contacts import configure_contacts_tools
from .deck import configure_deck_tools
__all__ = [
"configure_calendar_tools",
nextcloud_mcp_server/server/calendar.py
+13 -16
View File
@@ -4,11 +4,8 @@ from typing import Optional
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.models.calendar import (
Calendar,
ListCalendarsResponse,
)
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.calendar import Calendar, ListCalendarsResponse
logger = logging.getLogger(__name__)
@@ -18,7 +15,7 @@ def configure_calendar_tools(mcp: FastMCP):
@mcp.tool()
async def nc_calendar_list_calendars(ctx: Context) -> ListCalendarsResponse:
"""List all available calendars for the user"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
calendars_data = await client.calendar.list_calendars()
calendars = [Calendar(**cal_data) for cal_data in calendars_data]
@@ -74,7 +71,7 @@ def configure_calendar_tools(mcp: FastMCP):
Returns:
Dict with event creation result
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
event_data = {
"title": title,
@@ -133,7 +130,7 @@ def configure_calendar_tools(mcp: FastMCP):
Returns:
List of events matching the filters
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
# Convert YYYY-MM-DD format dates to datetime objects
start_datetime = None
@@ -207,7 +204,7 @@ def configure_calendar_tools(mcp: FastMCP):
ctx: Context,
):
"""Get detailed information about a specific event"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
event_data, etag = await client.calendar.get_event(calendar_name, event_uid)
return event_data
@@ -240,7 +237,7 @@ def configure_calendar_tools(mcp: FastMCP):
etag: str = "",
):
"""Update any aspect of an existing event"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
# Build update data with only non-None values
event_data = {}
@@ -290,7 +287,7 @@ def configure_calendar_tools(mcp: FastMCP):
ctx: Context,
):
"""Delete a calendar event"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.calendar.delete_event(calendar_name, event_uid)
@mcp.tool()
@@ -332,7 +329,7 @@ def configure_calendar_tools(mcp: FastMCP):
Returns:
Dict with meeting creation result
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
# Combine date and time for start_datetime
start_datetime = f"{date}T{time}:00"
@@ -366,7 +363,7 @@ def configure_calendar_tools(mcp: FastMCP):
limit: int = 10,
):
"""Get upcoming events in next N days"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
now = dt.datetime.now()
end_datetime = now + dt.timedelta(days=days_ahead)
@@ -435,7 +432,7 @@ def configure_calendar_tools(mcp: FastMCP):
Returns:
List of available time slots with start/end times and duration
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
# Parse attendees
attendee_list = []
@@ -536,7 +533,7 @@ def configure_calendar_tools(mcp: FastMCP):
Returns:
Summary of operation results including counts and details
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
if operation not in ["update", "delete", "move"]:
raise ValueError("Operation must be 'update', 'delete', or 'move'")
@@ -758,7 +755,7 @@ def configure_calendar_tools(mcp: FastMCP):
Returns:
Result of the calendar management operation
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
if action == "list":
return await client.calendar.list_calendars()
nextcloud_mcp_server/server/contacts.py
+8 -8
View File
@@ -2,7 +2,7 @@ import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.context import get_client
logger = logging.getLogger(__name__)
@@ -12,13 +12,13 @@ def configure_contacts_tools(mcp: FastMCP):
@mcp.tool()
async def nc_contacts_list_addressbooks(ctx: Context):
"""List all addressbooks for the user."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.list_addressbooks()
@mcp.tool()
async def nc_contacts_list_contacts(ctx: Context, *, addressbook: str):
"""List all contacts in the specified addressbook."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.list_contacts(addressbook=addressbook)
@mcp.tool()
@@ -31,7 +31,7 @@ def configure_contacts_tools(mcp: FastMCP):
name: The name of the addressbook.
display_name: The display name of the addressbook.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.create_addressbook(
name=name, display_name=display_name
)
@@ -39,7 +39,7 @@ def configure_contacts_tools(mcp: FastMCP):
@mcp.tool()
async def nc_contacts_delete_addressbook(ctx: Context, *, name: str):
"""Delete an addressbook."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.delete_addressbook(name=name)
@mcp.tool()
@@ -53,7 +53,7 @@ def configure_contacts_tools(mcp: FastMCP):
uid: The unique ID for the contact.
contact_data: A dictionary with the contact's details, e.g. {"fn": "John Doe", "email": "john.doe@example.com"}.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.create_contact(
addressbook=addressbook, uid=uid, contact_data=contact_data
)
@@ -61,7 +61,7 @@ def configure_contacts_tools(mcp: FastMCP):
@mcp.tool()
async def nc_contacts_delete_contact(ctx: Context, *, addressbook: str, uid: str):
"""Delete a contact."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.delete_contact(addressbook=addressbook, uid=uid)
@mcp.tool()
@@ -76,7 +76,7 @@ def configure_contacts_tools(mcp: FastMCP):
contact_data: A dictionary with the contact's updated details, e.g. {"fn": "Jane Doe", "email": "jane.doe@example.com"}.
etag: Optional ETag for optimistic concurrency control.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.contacts.update_contact(
addressbook=addressbook, uid=uid, contact_data=contact_data, etag=etag
)
nextcloud_mcp_server/server/deck.py
+121 -31
View File
@@ -3,15 +3,19 @@ from typing import Optional
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.deck import (
CreateBoardResponse,
CreateStackResponse,
StackOperationResponse,
CreateCardResponse,
CardOperationResponse,
CreateBoardResponse,
CreateCardResponse,
CreateLabelResponse,
CreateStackResponse,
DeckBoard,
DeckCard,
DeckLabel,
DeckStack,
LabelOperationResponse,
StackOperationResponse,
)
logger = logging.getLogger(__name__)
@@ -25,7 +29,8 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_boards_resource():
"""List all Nextcloud Deck boards"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning("This message is deprecated, use the deck_get_board instead")
client = get_client(ctx)
boards = await client.deck.get_boards()
return [board.model_dump() for board in boards]
@@ -33,7 +38,10 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_board_resource(board_id: int):
"""Get details of a specific Nextcloud Deck board"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_board tool instead"
)
client = get_client(ctx)
board = await client.deck.get_board(board_id)
return board.model_dump()
@@ -41,7 +49,10 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_stacks_resource(board_id: int):
"""List all stacks in a Nextcloud Deck board"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_stacks tool instead"
)
client = get_client(ctx)
stacks = await client.deck.get_stacks(board_id)
return [stack.model_dump() for stack in stacks]
@@ -49,7 +60,10 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_stack_resource(board_id: int, stack_id: int):
"""Get details of a specific Nextcloud Deck stack"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_stack tool instead"
)
client = get_client(ctx)
stack = await client.deck.get_stack(board_id, stack_id)
return stack.model_dump()
@@ -57,7 +71,10 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_cards_resource(board_id: int, stack_id: int):
"""List all cards in a Nextcloud Deck stack"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_cards tool instead"
)
client = get_client(ctx)
stack = await client.deck.get_stack(board_id, stack_id)
if stack.cards:
return [card.model_dump() for card in stack.cards]
@@ -67,7 +84,10 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_card_resource(board_id: int, stack_id: int, card_id: int):
"""Get details of a specific Nextcloud Deck card"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_card tool instead"
)
client = get_client(ctx)
card = await client.deck.get_card(board_id, stack_id, card_id)
return card.model_dump()
@@ -75,7 +95,10 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_labels_resource(board_id: int):
"""List all labels in a Nextcloud Deck board"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_labels tool instead"
)
client = get_client(ctx)
board = await client.deck.get_board(board_id)
return [label.model_dump() for label in board.labels]
@@ -83,11 +106,78 @@ def configure_deck_tools(mcp: FastMCP):
async def deck_label_resource(board_id: int, label_id: int):
"""Get details of a specific Nextcloud Deck label"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
await ctx.warning(
"This resource is deprecated, use the deck_get_label tool instead"
)
client = get_client(ctx)
label = await client.deck.get_label(board_id, label_id)
return label.model_dump()
# Tools
# Read Tools (converted from resources)
@mcp.tool()
async def deck_get_boards(ctx: Context) -> list[DeckBoard]:
"""Get all Nextcloud Deck boards"""
client = get_client(ctx)
boards = await client.deck.get_boards()
return boards
@mcp.tool()
async def deck_get_board(ctx: Context, board_id: int) -> DeckBoard:
"""Get details of a specific Nextcloud Deck board"""
client = get_client(ctx)
board = await client.deck.get_board(board_id)
return board
@mcp.tool()
async def deck_get_stacks(ctx: Context, board_id: int) -> list[DeckStack]:
"""Get all stacks in a Nextcloud Deck board"""
client = get_client(ctx)
stacks = await client.deck.get_stacks(board_id)
return stacks
@mcp.tool()
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)
stack = await client.deck.get_stack(board_id, stack_id)
return stack
@mcp.tool()
async def deck_get_cards(
ctx: Context, board_id: int, stack_id: int
) -> list[DeckCard]:
"""Get all cards in a Nextcloud Deck stack"""
client = get_client(ctx)
stack = await client.deck.get_stack(board_id, stack_id)
if stack.cards:
return stack.cards
return []
@mcp.tool()
async def deck_get_card(
ctx: Context, board_id: int, stack_id: int, card_id: int
) -> DeckCard:
"""Get details of a specific Nextcloud Deck card"""
client = get_client(ctx)
card = await client.deck.get_card(board_id, stack_id, card_id)
return card
@mcp.tool()
async def deck_get_labels(ctx: Context, board_id: int) -> list[DeckLabel]:
"""Get all labels in a Nextcloud Deck board"""
client = get_client(ctx)
board = await client.deck.get_board(board_id)
return board.labels
@mcp.tool()
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)
label = await client.deck.get_label(board_id, label_id)
return label
# Create/Update/Delete Tools
@mcp.tool()
async def deck_create_board(
@@ -99,7 +189,7 @@ def configure_deck_tools(mcp: FastMCP):
title: The title of the new board
color: The hexadecimal color of the new board (e.g. FF0000)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
board = await client.deck.create_board(title, color)
return CreateBoardResponse(id=board.id, title=board.title, color=board.color)
@@ -116,7 +206,7 @@ def configure_deck_tools(mcp: FastMCP):
title: The title of the new stack
order: Order for sorting the stacks
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
stack = await client.deck.create_stack(board_id, title, order)
return CreateStackResponse(id=stack.id, title=stack.title, order=stack.order)
@@ -136,7 +226,7 @@ def configure_deck_tools(mcp: FastMCP):
title: New title for the stack
order: New order for the stack
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.update_stack(board_id, stack_id, title, order)
return StackOperationResponse(
success=True,
@@ -155,7 +245,7 @@ def configure_deck_tools(mcp: FastMCP):
board_id: The ID of the board
stack_id: The ID of the stack
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.delete_stack(board_id, stack_id)
return StackOperationResponse(
success=True,
@@ -187,7 +277,7 @@ def configure_deck_tools(mcp: FastMCP):
description: Description of the card
duedate: Due date of the card (ISO-8601 format)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
card = await client.deck.create_card(
board_id, stack_id, title, type, order, description, duedate
)
@@ -228,7 +318,7 @@ def configure_deck_tools(mcp: FastMCP):
archived: Whether the card should be archived
done: Completion date for the card (ISO-8601 format)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.update_card(
board_id,
stack_id,
@@ -261,7 +351,7 @@ def configure_deck_tools(mcp: FastMCP):
stack_id: The ID of the stack
card_id: The ID of the card
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.delete_card(board_id, stack_id, card_id)
return CardOperationResponse(
success=True,
@@ -282,7 +372,7 @@ def configure_deck_tools(mcp: FastMCP):
stack_id: The ID of the stack
card_id: The ID of the card
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.archive_card(board_id, stack_id, card_id)
return CardOperationResponse(
success=True,
@@ -303,7 +393,7 @@ def configure_deck_tools(mcp: FastMCP):
stack_id: The ID of the stack
card_id: The ID of the card
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.unarchive_card(board_id, stack_id, card_id)
return CardOperationResponse(
success=True,
@@ -331,7 +421,7 @@ def configure_deck_tools(mcp: FastMCP):
order: New position in the target stack
target_stack_id: The ID of the target stack
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.reorder_card(
board_id, stack_id, card_id, order, target_stack_id
)
@@ -355,7 +445,7 @@ def configure_deck_tools(mcp: FastMCP):
title: The title of the new label
color: The color of the new label (hex format without #)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
label = await client.deck.create_label(board_id, title, color)
return CreateLabelResponse(id=label.id, title=label.title, color=label.color)
@@ -375,7 +465,7 @@ def configure_deck_tools(mcp: FastMCP):
title: New title for the label
color: New color for the label (hex format without #)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.update_label(board_id, label_id, title, color)
return LabelOperationResponse(
success=True,
@@ -394,7 +484,7 @@ def configure_deck_tools(mcp: FastMCP):
board_id: The ID of the board
label_id: The ID of the label
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.delete_label(board_id, label_id)
return LabelOperationResponse(
success=True,
@@ -416,7 +506,7 @@ def configure_deck_tools(mcp: FastMCP):
card_id: The ID of the card
label_id: The ID of the label to assign
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.assign_label_to_card(board_id, stack_id, card_id, label_id)
return CardOperationResponse(
success=True,
@@ -438,7 +528,7 @@ def configure_deck_tools(mcp: FastMCP):
card_id: The ID of the card
label_id: The ID of the label to remove
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.remove_label_from_card(board_id, stack_id, card_id, label_id)
return CardOperationResponse(
success=True,
@@ -461,7 +551,7 @@ def configure_deck_tools(mcp: FastMCP):
card_id: The ID of the card
user_id: The user ID to assign
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.assign_user_to_card(board_id, stack_id, card_id, user_id)
return CardOperationResponse(
success=True,
@@ -483,7 +573,7 @@ def configure_deck_tools(mcp: FastMCP):
card_id: The ID of the card
user_id: The user ID to unassign
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
await client.deck.unassign_user_from_card(board_id, stack_id, card_id, user_id)
return CardOperationResponse(
success=True,
nextcloud_mcp_server/server/notes.py
+80 -20
View File
@@ -1,20 +1,20 @@
import logging
from httpx import HTTPStatusError
from mcp.server.fastmcp import Context, FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.notes import (
Note,
NotesSettings,
CreateNoteResponse,
UpdateNoteResponse,
DeleteNoteResponse,
AppendContentResponse,
SearchNotesResponse,
CreateNoteResponse,
DeleteNoteResponse,
Note,
NoteSearchResult,
NotesSettings,
SearchNotesResponse,
UpdateNoteResponse,
)
logger = logging.getLogger(__name__)
@@ -27,15 +27,15 @@ def configure_notes_tools(mcp: FastMCP):
ctx: Context = (
mcp.get_context()
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
settings_data = await client.notes.get_settings()
return NotesSettings(**settings_data)
@mcp.resource("nc://Notes/{note_id}/attachments/{attachment_filename}")
async def nc_notes_get_attachment(note_id: int, attachment_filename: str):
async def nc_notes_get_attachment_resource(note_id: int, attachment_filename: str):
"""Get a specific attachment from a note"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
# Assuming a method get_note_attachment exists in the client
# This method should return the raw content and determine the mime type
content, mime_type = await client.webdav.get_note_attachment(
@@ -53,11 +53,11 @@ def configure_notes_tools(mcp: FastMCP):
}
@mcp.resource("nc://Notes/{note_id}")
async def nc_get_note(note_id: int):
async def nc_get_note_resource(note_id: int):
"""Get user note using note id"""
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
try:
note_data = await client.notes.get_note(note_id)
return Note(**note_data)
@@ -81,7 +81,7 @@ def configure_notes_tools(mcp: FastMCP):
title: str, content: str, category: str, ctx: Context
) -> CreateNoteResponse:
"""Create a new note"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
try:
note_data = await client.notes.create_note(
title=title,
@@ -129,11 +129,11 @@ def configure_notes_tools(mcp: FastMCP):
"""Update an existing note's title, content, or category.
REQUIRED: etag parameter must be provided to prevent overwriting concurrent changes.
Get the current ETag by first retrieving the note using nc://Notes/{note_id} resource.
Get the current ETag by first retrieving the note using nc_notes_get_note tool.
If the note has been modified by someone else since you retrieved it,
the update will fail with a 412 error."""
logger.info("Updating note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
try:
note_data = await client.notes.update(
note_id=note_id,
@@ -183,7 +183,7 @@ def configure_notes_tools(mcp: FastMCP):
between the note and what will be appended."""
logger.info("Appending content to note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
try:
note_data = await client.notes.append_content(
note_id=note_id, content=content
@@ -220,7 +220,7 @@ def configure_notes_tools(mcp: FastMCP):
@mcp.tool()
async def nc_notes_search_notes(query: str, ctx: Context) -> SearchNotesResponse:
"""Search notes by title or content, returning only id, title, and category."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
try:
search_results_raw = await client.notes_search_notes(query=query)
@@ -258,11 +258,71 @@ def configure_notes_tools(mcp: FastMCP):
)
)
@mcp.tool()
async def nc_notes_get_note(note_id: int, ctx: Context) -> Note:
"""Get a specific note by its ID"""
client = get_client(ctx)
try:
note_data = await client.notes.get_note(note_id)
return Note(**note_data)
except HTTPStatusError as e:
if e.response.status_code == 404:
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
elif e.response.status_code == 403:
raise McpError(
ErrorData(code=-1, message=f"Access denied to note {note_id}")
)
else:
raise McpError(
ErrorData(
code=-1,
message=f"Failed to retrieve note {note_id}: {e.response.reason_phrase}",
)
)
@mcp.tool()
async def nc_notes_get_attachment(
note_id: int, attachment_filename: str, ctx: Context
) -> dict[str, str]:
"""Get a specific attachment from a note"""
client = get_client(ctx)
try:
content, mime_type = await client.webdav.get_note_attachment(
note_id=note_id, filename=attachment_filename
)
return {
"uri": f"nc://Notes/{note_id}/attachments/{attachment_filename}",
"mimeType": mime_type,
"data": content,
}
except HTTPStatusError as e:
if e.response.status_code == 404:
raise McpError(
ErrorData(
code=-1,
message=f"Attachment {attachment_filename} not found for note {note_id}",
)
)
elif e.response.status_code == 403:
raise McpError(
ErrorData(
code=-1,
message=f"Access denied to attachment {attachment_filename} for note {note_id}",
)
)
else:
raise McpError(
ErrorData(
code=-1,
message=f"Failed to retrieve attachment: {e.response.reason_phrase}",
)
)
@mcp.tool()
async def nc_notes_delete_note(note_id: int, ctx: Context) -> DeleteNoteResponse:
"""Delete a note permanently"""
logger.info("Deleting note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
try:
await client.notes.delete_note(note_id)
return DeleteNoteResponse(
nextcloud_mcp_server/server/tables.py
+7 -7
View File
@@ -2,7 +2,7 @@ import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.context import get_client
logger = logging.getLogger(__name__)
@@ -12,13 +12,13 @@ def configure_tables_tools(mcp: FastMCP):
@mcp.tool()
async def nc_tables_list_tables(ctx: Context):
"""List all tables available to the user"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.tables.list_tables()
@mcp.tool()
async def nc_tables_get_schema(table_id: int, ctx: Context):
"""Get the schema/structure of a specific table including columns and views"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.tables.get_table_schema(table_id)
@mcp.tool()
@@ -29,7 +29,7 @@ def configure_tables_tools(mcp: FastMCP):
offset: int | None = None,
):
"""Read rows from a table with optional pagination"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.tables.get_table_rows(table_id, limit, offset)
@mcp.tool()
@@ -38,7 +38,7 @@ def configure_tables_tools(mcp: FastMCP):
Data should be a dictionary mapping column IDs to values, e.g. {1: "text", 2: 42}
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.tables.create_row(table_id, data)
@mcp.tool()
@@ -47,11 +47,11 @@ def configure_tables_tools(mcp: FastMCP):
Data should be a dictionary mapping column IDs to new values, e.g. {1: "new text", 2: 99}
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.tables.update_row(row_id, data)
@mcp.tool()
async def nc_tables_delete_row(row_id: int, ctx: Context):
"""Delete a row from a table"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.tables.delete_row(row_id)
nextcloud_mcp_server/server/webdav.py
+8 -8
View File
@@ -2,7 +2,7 @@ import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.context import get_client
logger = logging.getLogger(__name__)
@@ -26,7 +26,7 @@ def configure_webdav_tools(mcp: FastMCP):
# List a specific folder
await nc_webdav_list_directory("Documents/Projects")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.webdav.list_directory(path)
@mcp.tool()
@@ -49,7 +49,7 @@ def configure_webdav_tools(mcp: FastMCP):
result = await nc_webdav_read_file("Images/photo.jpg")
logger.info(result['encoding']) # 'base64'
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
content, content_type = await client.webdav.read_file(path)
# For text files, decode content for easier viewing
@@ -97,7 +97,7 @@ def configure_webdav_tools(mcp: FastMCP):
# Write binary data (base64 encoded)
await nc_webdav_write_file("files/data.bin", base64_content, "application/octet-stream;base64")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
# Handle base64 encoded content
if content_type and "base64" in content_type.lower():
@@ -127,7 +127,7 @@ def configure_webdav_tools(mcp: FastMCP):
# Create nested directories (parent must exist)
await nc_webdav_create_directory("Projects/MyApp/docs")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.webdav.create_directory(path)
@mcp.tool()
@@ -147,7 +147,7 @@ def configure_webdav_tools(mcp: FastMCP):
# Delete a directory (will delete all contents)
await nc_webdav_delete_resource("temp_folder")
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.webdav.delete_resource(path)
@mcp.tool()
@@ -177,7 +177,7 @@ def configure_webdav_tools(mcp: FastMCP):
# Move and overwrite if destination exists
await nc_webdav_move_resource("document.txt", "Archive/document.txt", overwrite=True)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.webdav.move_resource(
source_path, destination_path, overwrite
)
@@ -209,7 +209,7 @@ def configure_webdav_tools(mcp: FastMCP):
# Copy and overwrite if destination exists
await nc_webdav_copy_resource("document.txt", "Backup/document.txt", overwrite=True)
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
client = get_client(ctx)
return await client.webdav.copy_resource(
source_path, destination_path, overwrite
)