ryan
91b7f312aa
- Introduced a new `compose.yaml` file to define the Docker service for the OAuth server. - Created `oauth.py` to implement a single-user OAuth authorization server with PKCE, SQLite persistence, and token management. - Added necessary database schema for OAuth clients, flows, codes, tokens, and login failures. - Implemented login routes with CSRF protection and error handling. - Updated `requirements.txt` to include dependencies for the OAuth server.
581 lines
24 KiB
Python
581 lines
24 KiB
Python
"""WebDAV-backed MCP server, scoped to one configured vault directory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import xml.etree.ElementTree as ET
|
|
from collections import deque
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from urllib.parse import quote, unquote, urljoin, urlsplit
|
|
|
|
import httpx
|
|
from mcp.server import MCPServer
|
|
from mcp.server.auth.middleware.auth_context import get_access_token
|
|
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions
|
|
from mcp.server.auth.routes import create_protected_resource_routes
|
|
from mcp.server.transport_security import RequestBodyLimitMiddleware, TransportSecuritySettings
|
|
from mcp.types import ToolAnnotations
|
|
from pydantic import AnyHttpUrl
|
|
from starlette.applications import Starlette
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
from starlette.routing import Mount, Route
|
|
|
|
from oauth import LocalOAuthProvider, make_login_routes
|
|
|
|
|
|
DAV_NS = "{DAV:}"
|
|
MAX_FILE_BYTES = 256 * 1024
|
|
MAX_PROPFIND_BYTES = 2 * 1024 * 1024
|
|
MAX_SEARCH_DIRECTORIES = 500
|
|
MAX_SEARCH_FILES = 1500
|
|
HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
|
|
|
|
|
|
class WebDAVError(Exception):
|
|
pass
|
|
|
|
|
|
def _secret(name: str, *, required: bool = True) -> str:
|
|
file_name = os.getenv(f"{name}_FILE")
|
|
if file_name:
|
|
try:
|
|
value = Path(file_name).read_bytes().rstrip(b"\r\n").decode("utf-8")
|
|
except OSError as exc:
|
|
raise RuntimeError(f"Cannot read secret file configured by {name}_FILE") from exc
|
|
else:
|
|
value = os.getenv(name, "")
|
|
if required and not value:
|
|
raise RuntimeError(f"Set {name} or {name}_FILE before starting the service")
|
|
return value
|
|
|
|
|
|
def _https_origin(name: str, default: str) -> str:
|
|
value = os.getenv(name, default).strip().rstrip("/")
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme != "https" and parsed.hostname not in {"localhost", "127.0.0.1", "::1"}:
|
|
raise RuntimeError(f"{name} must use HTTPS")
|
|
if not parsed.scheme or not parsed.netloc or parsed.username or parsed.password or parsed.query or parsed.fragment:
|
|
raise RuntimeError(f"{name} must be an absolute URL without credentials, query, or fragment")
|
|
return value
|
|
|
|
|
|
WEBDAV_URL = _https_origin("WEBDAV_URL", "https://dav.ryankegel.com")
|
|
WEBDAV_USERNAME = os.getenv("WEBDAV_USERNAME", "").strip()
|
|
WEBDAV_PASSWORD = _secret("WEBDAV_PASSWORD")
|
|
if not WEBDAV_USERNAME:
|
|
raise RuntimeError("Set WEBDAV_USERNAME before starting the service")
|
|
|
|
WEBDAV_ROOT = os.getenv("WEBDAV_ROOT", "/Notes/Personal")
|
|
if not WEBDAV_ROOT.startswith("/"):
|
|
raise RuntimeError("WEBDAV_ROOT must be an absolute WebDAV path, for example /Notes/Personal")
|
|
_root_raw = WEBDAV_ROOT.strip("/")
|
|
_root_parts = _root_raw.split("/") if _root_raw else []
|
|
if not _root_parts or any(part in {"", ".", ".."} for part in _root_parts):
|
|
raise RuntimeError("WEBDAV_ROOT must be a non-root path without dot segments")
|
|
for _part in _root_parts:
|
|
_decoded_part = _part
|
|
for _ in range(4):
|
|
_decoded_part = unquote(_decoded_part)
|
|
if _decoded_part in {".", ".."} or "/" in _decoded_part or "\\" in _decoded_part:
|
|
raise RuntimeError("WEBDAV_ROOT cannot contain encoded path separators or dot segments")
|
|
if _decoded_part == _part:
|
|
break
|
|
|
|
MCP_PUBLIC_URL = _https_origin("MCP_PUBLIC_URL", "https://mcp.example.com")
|
|
if urlsplit(MCP_PUBLIC_URL).path not in {"", "/"}:
|
|
raise RuntimeError("MCP_PUBLIC_URL must be the HTTPS origin only, without a path")
|
|
MCP_RESOURCE_URL = f"{MCP_PUBLIC_URL}/mcp"
|
|
MCP_AUTH_USERNAME = os.getenv("MCP_AUTH_USERNAME", "").strip()
|
|
MCP_AUTH_PASSWORD = _secret("MCP_AUTH_PASSWORD")
|
|
if not MCP_AUTH_USERNAME:
|
|
raise RuntimeError("Set MCP_AUTH_USERNAME before starting the service")
|
|
if len(MCP_AUTH_PASSWORD.encode("utf-8")) < 16:
|
|
raise RuntimeError("MCP_AUTH_PASSWORD must contain at least 16 bytes")
|
|
|
|
MCP_ALLOWED_HOSTS = [item.strip() for item in os.getenv("MCP_ALLOWED_HOSTS", "").split(",") if item.strip()]
|
|
if not MCP_ALLOWED_HOSTS:
|
|
raise RuntimeError("Set MCP_ALLOWED_HOSTS to the public MCP host name")
|
|
MCP_ALLOWED_ORIGINS = [item.strip() for item in os.getenv("MCP_ALLOWED_ORIGINS", "").split(",") if item.strip()]
|
|
READ_ONLY = os.getenv("MCP_READ_ONLY", "true").strip().lower() not in {"0", "false", "no"}
|
|
DATABASE_PATH = os.getenv("OAUTH_DATABASE_PATH", "/data/oauth.sqlite3")
|
|
OAUTH_SCOPES = ["notes:read"] + ([] if READ_ONLY else ["notes:write"])
|
|
|
|
_root_encoded = "/".join(quote(part, safe="-._~") for part in _root_parts)
|
|
_WEBDAV_ROOT_URL = f"{WEBDAV_URL.rstrip('/')}/{_root_encoded}"
|
|
_DAV_ORIGIN = urlsplit(WEBDAV_URL)
|
|
_ROOT_PATH_UNQUOTED = unquote(urlsplit(_WEBDAV_ROOT_URL).path).rstrip("/")
|
|
|
|
|
|
def _clean_relative_path(path: str, *, allow_empty: bool = True) -> list[str]:
|
|
if not isinstance(path, str):
|
|
raise WebDAVError("Path must be text")
|
|
if len(path) > 2048:
|
|
raise WebDAVError("Path is too long")
|
|
if path.startswith("/") or "\\" in path or "\x00" in path:
|
|
raise WebDAVError("Use a relative path inside the configured vault")
|
|
raw = path[:-1] if path.endswith("/") else path
|
|
if not raw:
|
|
if allow_empty:
|
|
return []
|
|
raise WebDAVError("A file path is required")
|
|
parts = raw.split("/")
|
|
if any(not part for part in parts):
|
|
raise WebDAVError("Path cannot contain empty segments")
|
|
for part in parts:
|
|
decoded = part
|
|
for _ in range(4):
|
|
decoded = unquote(decoded)
|
|
if decoded in {".", ".."} or "/" in decoded or "\\" in decoded:
|
|
raise WebDAVError("Path cannot escape the configured vault")
|
|
if decoded == part:
|
|
break
|
|
if any(ord(char) < 32 or ord(char) == 127 for char in decoded):
|
|
raise WebDAVError("Path cannot contain control characters")
|
|
return parts
|
|
|
|
|
|
def _path_url(path: str, *, collection: bool = False) -> str:
|
|
parts = _clean_relative_path(path)
|
|
encoded = "/".join(quote(part, safe="-._~") for part in parts)
|
|
url = _WEBDAV_ROOT_URL + (f"/{encoded}" if encoded else "")
|
|
if collection:
|
|
return url.rstrip("/") + "/"
|
|
return url
|
|
|
|
|
|
def _same_origin(url: str) -> bool:
|
|
parsed = urlsplit(url)
|
|
try:
|
|
base_port = _DAV_ORIGIN.port or (443 if _DAV_ORIGIN.scheme.lower() == "https" else 80)
|
|
parsed_port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
|
|
except ValueError:
|
|
return False
|
|
return (
|
|
parsed.scheme.lower() == _DAV_ORIGIN.scheme.lower()
|
|
and (parsed.hostname or "").lower() == (_DAV_ORIGIN.hostname or "").lower()
|
|
and parsed_port == base_port
|
|
)
|
|
|
|
|
|
def _href_relative(href: str) -> str | None:
|
|
absolute = urljoin(WEBDAV_URL.rstrip("/") + "/", href)
|
|
parsed = urlsplit(absolute)
|
|
if not _same_origin(absolute) or parsed.query or parsed.fragment:
|
|
return None
|
|
candidate = unquote(parsed.path).rstrip("/")
|
|
if candidate == _ROOT_PATH_UNQUOTED:
|
|
return ""
|
|
prefix = _ROOT_PATH_UNQUOTED + "/"
|
|
if not candidate.startswith(prefix):
|
|
return None
|
|
relative = candidate[len(prefix):]
|
|
try:
|
|
return "/".join(_clean_relative_path(relative, allow_empty=False))
|
|
except WebDAVError:
|
|
return None
|
|
|
|
|
|
def _dav_client() -> httpx.AsyncClient:
|
|
return httpx.AsyncClient(
|
|
auth=httpx.BasicAuth(WEBDAV_USERNAME, WEBDAV_PASSWORD),
|
|
timeout=HTTP_TIMEOUT,
|
|
follow_redirects=False,
|
|
headers={"User-Agent": "dav-mcp/1.0"},
|
|
)
|
|
|
|
|
|
async def _limited_response_body(response: httpx.Response, max_bytes: int) -> bytes:
|
|
result = bytearray()
|
|
async for chunk in response.aiter_bytes():
|
|
if len(result) + len(chunk) > max_bytes:
|
|
raise WebDAVError(f"Response exceeds the {max_bytes}-byte size limit")
|
|
result.extend(chunk)
|
|
return bytes(result)
|
|
|
|
|
|
async def _propfind(client: httpx.AsyncClient, path: str, *, depth: str = "1") -> tuple[bytes, int]:
|
|
body = (
|
|
'<?xml version="1.0" encoding="utf-8"?>'
|
|
'<d:propfind xmlns:d="DAV:"><d:prop><d:resourcetype/>'
|
|
'<d:getcontentlength/><d:getlastmodified/><d:getetag/></d:prop></d:propfind>'
|
|
)
|
|
try:
|
|
async with client.stream(
|
|
"PROPFIND", _path_url(path, collection=True),
|
|
headers={"Depth": depth, "Content-Type": "application/xml; charset=utf-8"},
|
|
content=body,
|
|
) as response:
|
|
if response.status_code == 404:
|
|
raise WebDAVError(f"Folder not found inside the vault: {path or '/'}")
|
|
if response.status_code not in {200, 207}:
|
|
raise WebDAVError(f"WebDAV PROPFIND failed with HTTP {response.status_code}")
|
|
return await _limited_response_body(response, MAX_PROPFIND_BYTES), response.status_code
|
|
except httpx.HTTPError as exc:
|
|
raise WebDAVError(f"WebDAV request failed: {type(exc).__name__}") from exc
|
|
|
|
|
|
def _parse_propfind(xml_body: bytes) -> list[dict[str, object]]:
|
|
try:
|
|
root = ET.fromstring(xml_body)
|
|
except ET.ParseError as exc:
|
|
raise WebDAVError("WebDAV returned invalid directory XML") from exc
|
|
items: list[dict[str, object]] = []
|
|
for response in root.findall(f"{DAV_NS}response"):
|
|
href_node = response.find(f"{DAV_NS}href")
|
|
if href_node is None or not href_node.text:
|
|
continue
|
|
rel_path = _href_relative(href_node.text)
|
|
if rel_path is None:
|
|
continue
|
|
props = None
|
|
for propstat in response.findall(f"{DAV_NS}propstat"):
|
|
status_node = propstat.find(f"{DAV_NS}status")
|
|
if status_node is not None and status_node.text and " 200 " not in status_node.text:
|
|
continue
|
|
props = propstat.find(f"{DAV_NS}prop")
|
|
if props is not None:
|
|
break
|
|
if props is None:
|
|
continue
|
|
resource_type = props.find(f"{DAV_NS}resourcetype")
|
|
is_directory = resource_type is not None and resource_type.find(f"{DAV_NS}collection") is not None
|
|
length_node = props.find(f"{DAV_NS}getcontentlength")
|
|
modified_node = props.find(f"{DAV_NS}getlastmodified")
|
|
etag_node = props.find(f"{DAV_NS}getetag")
|
|
try:
|
|
size = int(length_node.text) if length_node is not None and length_node.text else None
|
|
except ValueError:
|
|
size = None
|
|
items.append(
|
|
{
|
|
"path": rel_path,
|
|
"name": rel_path.rsplit("/", 1)[-1] if rel_path else "/",
|
|
"is_directory": is_directory,
|
|
"size": size,
|
|
"last_modified": modified_node.text if modified_node is not None else None,
|
|
"etag": etag_node.text if etag_node is not None else None,
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
async def _list_directory(client: httpx.AsyncClient, path: str) -> list[dict[str, object]]:
|
|
base = "/".join(_clean_relative_path(path))
|
|
xml_body, _ = await _propfind(client, base, depth="1")
|
|
all_items = _parse_propfind(xml_body)
|
|
target = next((item for item in all_items if item["path"] == base), None)
|
|
if target is not None and not target["is_directory"]:
|
|
raise WebDAVError(f"Not a folder: {base or '/'}")
|
|
children = []
|
|
for item in all_items:
|
|
item_path = str(item["path"])
|
|
if item_path == base:
|
|
continue
|
|
parent = item_path.rsplit("/", 1)[0] if "/" in item_path else ""
|
|
if parent == base:
|
|
children.append(item)
|
|
children.sort(key=lambda item: (not bool(item["is_directory"]), str(item["name"]).casefold()))
|
|
return children
|
|
|
|
|
|
async def _walk_directories(client: httpx.AsyncClient, start: str):
|
|
queue = deque([(start, 0)])
|
|
visited: set[str] = set()
|
|
while queue and len(visited) < MAX_SEARCH_DIRECTORIES:
|
|
current, depth = queue.popleft()
|
|
if current in visited:
|
|
continue
|
|
visited.add(current)
|
|
entries = await _list_directory(client, current)
|
|
yield current, entries
|
|
if depth < 32:
|
|
for entry in entries:
|
|
if entry["is_directory"]:
|
|
queue.append((str(entry["path"]), depth + 1))
|
|
|
|
|
|
def _write_scope_ok() -> bool:
|
|
access = get_access_token()
|
|
return bool(access and "notes:write" in access.scopes)
|
|
|
|
|
|
oauth_provider = LocalOAuthProvider(
|
|
issuer_url=MCP_PUBLIC_URL,
|
|
resource_url=MCP_RESOURCE_URL,
|
|
database_path=DATABASE_PATH,
|
|
username=MCP_AUTH_USERNAME,
|
|
password=MCP_AUTH_PASSWORD,
|
|
writable=not READ_ONLY,
|
|
)
|
|
|
|
mcp = MCPServer(
|
|
"webdav-personal-notes",
|
|
title="Personal Obsidian Vault",
|
|
description="Browse, search, and read the WebDAV vault at /Notes/Personal.",
|
|
instructions=(
|
|
"All paths are relative to the configured /Notes/Personal vault. "
|
|
"Use list_directory, search_files, or search_notes to discover notes before reading. "
|
|
"The server never exposes paths outside the configured vault."
|
|
),
|
|
auth_server_provider=oauth_provider,
|
|
auth=AuthSettings(
|
|
issuer_url=AnyHttpUrl(MCP_PUBLIC_URL),
|
|
resource_server_url=AnyHttpUrl(MCP_RESOURCE_URL),
|
|
required_scopes=["notes:read"],
|
|
validate_token_resource=True,
|
|
client_registration_options=ClientRegistrationOptions(
|
|
enabled=True,
|
|
valid_scopes=OAUTH_SCOPES,
|
|
default_scopes=OAUTH_SCOPES,
|
|
),
|
|
revocation_options=RevocationOptions(enabled=True),
|
|
),
|
|
)
|
|
|
|
|
|
@mcp.tool(
|
|
description="List files and subfolders immediately inside a vault folder. Use an empty path for the vault root.",
|
|
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True),
|
|
)
|
|
async def list_directory(path: str = "") -> str:
|
|
"""List direct children of a relative vault folder."""
|
|
try:
|
|
async with _dav_client() as client:
|
|
entries = await _list_directory(client, path)
|
|
return json.dumps(entries, ensure_ascii=False, indent=2)
|
|
except (WebDAVError, httpx.HTTPError) as exc:
|
|
return f"Error: {exc}"
|
|
|
|
|
|
@mcp.tool(
|
|
description="Read a UTF-8 text or Markdown file from the vault. Returns the file text and ETag; files are limited to 256 KiB.",
|
|
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True),
|
|
)
|
|
async def read_file(path: str) -> str:
|
|
"""Read a text note by its relative path."""
|
|
try:
|
|
parts = _clean_relative_path(path, allow_empty=False)
|
|
relative = "/".join(parts)
|
|
async with _dav_client() as client:
|
|
async with client.stream("GET", _path_url(relative)) as response:
|
|
if response.status_code == 404:
|
|
raise WebDAVError(f"File not found inside the vault: {relative}")
|
|
if response.status_code >= 300:
|
|
raise WebDAVError(f"WebDAV GET failed with HTTP {response.status_code}")
|
|
data = await _limited_response_body(response, MAX_FILE_BYTES)
|
|
etag = response.headers.get("ETag")
|
|
text = data.decode("utf-8-sig", errors="replace")
|
|
return json.dumps({"path": relative, "etag": etag, "content": text}, ensure_ascii=False)
|
|
except (WebDAVError, httpx.HTTPError) as exc:
|
|
return f"Error: {exc}"
|
|
|
|
|
|
@mcp.tool(
|
|
description="Find files and folders by a case-insensitive path or filename substring. Searches recursively from the supplied folder.",
|
|
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True),
|
|
)
|
|
async def search_files(query: str, path: str = "", limit: int = 100) -> str:
|
|
"""Search file and folder paths recursively."""
|
|
try:
|
|
_clean_relative_path(path)
|
|
if not query or len(query) > 256:
|
|
raise WebDAVError("Search query must contain 1 to 256 characters")
|
|
if not 1 <= limit <= 200:
|
|
raise WebDAVError("limit must be between 1 and 200")
|
|
needle = query.casefold()
|
|
matches: list[dict[str, object]] = []
|
|
dirs_visited = 0
|
|
truncated = False
|
|
async with _dav_client() as client:
|
|
async for _, entries in _walk_directories(client, path):
|
|
dirs_visited += 1
|
|
for entry in entries:
|
|
if needle in str(entry["path"]).casefold():
|
|
matches.append(entry)
|
|
if len(matches) >= limit:
|
|
truncated = True
|
|
break
|
|
if truncated:
|
|
break
|
|
if dirs_visited >= MAX_SEARCH_DIRECTORIES:
|
|
truncated = True
|
|
return json.dumps({"matches": matches, "truncated": truncated}, ensure_ascii=False, indent=2)
|
|
except (WebDAVError, httpx.HTTPError) as exc:
|
|
return f"Error: {exc}"
|
|
|
|
|
|
@mcp.tool(
|
|
description="Search note text recursively. Searches Markdown and text files only; each file is limited to 256 KiB and the crawl is capped.",
|
|
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True),
|
|
)
|
|
async def search_notes(query: str, path: str = "", limit: int = 20) -> str:
|
|
"""Search Markdown and text file contents for a phrase."""
|
|
try:
|
|
_clean_relative_path(path)
|
|
if not query or len(query) > 256:
|
|
raise WebDAVError("Search query must contain 1 to 256 characters")
|
|
if not 1 <= limit <= 100:
|
|
raise WebDAVError("limit must be between 1 and 100")
|
|
needle = query.casefold()
|
|
results: list[dict[str, str]] = []
|
|
scanned_files = 0
|
|
dirs_visited = 0
|
|
truncated = False
|
|
async with _dav_client() as client:
|
|
async for _, entries in _walk_directories(client, path):
|
|
dirs_visited += 1
|
|
for entry in entries:
|
|
if entry["is_directory"]:
|
|
continue
|
|
file_path = str(entry["path"])
|
|
if not file_path.casefold().endswith((".md", ".markdown", ".txt")):
|
|
continue
|
|
scanned_files += 1
|
|
if scanned_files > MAX_SEARCH_FILES:
|
|
truncated = True
|
|
break
|
|
try:
|
|
async with client.stream("GET", _path_url(file_path)) as response:
|
|
if response.status_code == 404:
|
|
continue
|
|
if response.status_code >= 300:
|
|
raise WebDAVError(f"WebDAV GET failed with HTTP {response.status_code}")
|
|
data = await _limited_response_body(response, MAX_FILE_BYTES)
|
|
except httpx.HTTPError as exc:
|
|
raise WebDAVError(f"WebDAV request failed: {type(exc).__name__}") from exc
|
|
except WebDAVError as exc:
|
|
if "size limit" in str(exc):
|
|
truncated = True
|
|
continue
|
|
raise
|
|
content = data.decode("utf-8-sig", errors="replace")
|
|
match_at = content.casefold().find(needle)
|
|
if match_at >= 0:
|
|
start = max(0, match_at - 180)
|
|
end = min(len(content), match_at + len(query) + 180)
|
|
snippet = content[start:end].replace("\r", " ").replace("\n", " ")
|
|
results.append({"path": file_path, "snippet": snippet})
|
|
if len(results) >= limit:
|
|
truncated = True
|
|
break
|
|
if truncated:
|
|
break
|
|
if dirs_visited >= MAX_SEARCH_DIRECTORIES or scanned_files >= MAX_SEARCH_FILES:
|
|
truncated = True
|
|
return json.dumps({"matches": results, "scanned_files": scanned_files, "truncated": truncated}, ensure_ascii=False, indent=2)
|
|
except (WebDAVError, httpx.HTTPError) as exc:
|
|
return f"Error: {exc}"
|
|
|
|
|
|
if not READ_ONLY:
|
|
|
|
@mcp.tool(
|
|
description=(
|
|
"Create or replace a UTF-8 text file in the vault. New files are create-only by default. "
|
|
"To replace a note, set overwrite=true and pass the ETag returned by read_file as expected_etag."
|
|
),
|
|
annotations=ToolAnnotations(
|
|
read_only_hint=False, destructive_hint=True, idempotent_hint=False, open_world_hint=True
|
|
),
|
|
)
|
|
async def write_file(
|
|
path: str,
|
|
content: str,
|
|
overwrite: bool = False,
|
|
expected_etag: str | None = None,
|
|
) -> str:
|
|
"""Create a note, or replace it only when an ETag confirms the version read."""
|
|
if not _write_scope_ok():
|
|
return "Error: This OAuth connection lacks notes:write. Reconnect and authorize write access."
|
|
try:
|
|
parts = _clean_relative_path(path, allow_empty=False)
|
|
relative = "/".join(parts)
|
|
if path.endswith("/"):
|
|
raise WebDAVError("write_file requires a file path, not a folder path")
|
|
if overwrite and not expected_etag:
|
|
raise WebDAVError("expected_etag is required when overwrite=true; read the file first")
|
|
if not overwrite and expected_etag:
|
|
raise WebDAVError("expected_etag is only used when overwrite=true")
|
|
data = content.encode("utf-8")
|
|
if len(data) > MAX_FILE_BYTES:
|
|
raise WebDAVError(f"File exceeds the {MAX_FILE_BYTES}-byte write limit")
|
|
headers = {"Content-Type": "text/markdown; charset=utf-8"}
|
|
headers["If-Match" if overwrite else "If-None-Match"] = expected_etag if overwrite else "*"
|
|
async with _dav_client() as client:
|
|
async with client.stream("PUT", _path_url(relative), headers=headers, content=data) as response:
|
|
if response.status_code == 412:
|
|
raise WebDAVError("The file changed or already exists. Read it again and retry deliberately.")
|
|
if response.status_code not in {200, 201, 204}:
|
|
raise WebDAVError(f"WebDAV PUT failed with HTTP {response.status_code}")
|
|
etag = response.headers.get("ETag")
|
|
return json.dumps({"saved": relative, "etag": etag, "bytes_written": len(data)}, ensure_ascii=False)
|
|
except (WebDAVError, httpx.HTTPError) as exc:
|
|
return f"Error: {exc}"
|
|
|
|
@mcp.tool(
|
|
description="Create one new folder inside the vault. The parent folder must already exist.",
|
|
annotations=ToolAnnotations(
|
|
read_only_hint=False, destructive_hint=False, idempotent_hint=True, open_world_hint=True
|
|
),
|
|
)
|
|
async def create_directory(path: str) -> str:
|
|
"""Create a single directory using WebDAV MKCOL."""
|
|
if not _write_scope_ok():
|
|
return "Error: This OAuth connection lacks notes:write. Reconnect and authorize write access."
|
|
try:
|
|
parts = _clean_relative_path(path, allow_empty=False)
|
|
relative = "/".join(parts)
|
|
async with _dav_client() as client:
|
|
async with client.stream("MKCOL", _path_url(relative, collection=True)) as response:
|
|
if response.status_code == 405:
|
|
raise WebDAVError("That folder already exists")
|
|
if response.status_code not in {200, 201, 204}:
|
|
raise WebDAVError(f"WebDAV MKCOL failed with HTTP {response.status_code}")
|
|
return f"Created folder: {relative}"
|
|
except (WebDAVError, httpx.HTTPError) as exc:
|
|
return f"Error: {exc}"
|
|
|
|
|
|
async def health(_: Request) -> JSONResponse:
|
|
return JSONResponse({"status": "ok"})
|
|
|
|
|
|
transport_security = TransportSecuritySettings(
|
|
allowed_hosts=MCP_ALLOWED_HOSTS,
|
|
allowed_origins=MCP_ALLOWED_ORIGINS,
|
|
)
|
|
mcp_http_app = mcp.streamable_http_app(transport_security=transport_security)
|
|
_resource_metadata_path = "/.well-known/oauth-protected-resource/mcp"
|
|
mcp_http_app.router.routes = [
|
|
route for route in mcp_http_app.router.routes
|
|
if getattr(route, "path", None) != _resource_metadata_path
|
|
]
|
|
mcp_http_app.router.routes.extend(
|
|
create_protected_resource_routes(
|
|
resource_url=AnyHttpUrl(MCP_RESOURCE_URL),
|
|
authorization_servers=[AnyHttpUrl(MCP_PUBLIC_URL)],
|
|
scopes_supported=OAUTH_SCOPES,
|
|
resource_name="Personal Obsidian Vault",
|
|
)
|
|
)
|
|
for _route in make_login_routes(oauth_provider):
|
|
if getattr(_route, "path", None) == "/login/callback":
|
|
_route.app = RequestBodyLimitMiddleware(_route.app, 16 * 1024)
|
|
mcp_http_app.router.routes.append(_route)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: Starlette):
|
|
await oauth_provider.initialize()
|
|
async with mcp.session_manager.run():
|
|
yield
|
|
|
|
|
|
app = Starlette(
|
|
routes=[Route("/healthz", health, methods=["GET"]), Mount("/", app=mcp_http_app)],
|
|
lifespan=lifespan,
|
|
)
|