Add OAuth server implementation and Docker Compose configuration
- 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.
This commit is contained in:
@@ -0,0 +1,580 @@
|
||||
"""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,
|
||||
)
|
||||
Reference in New Issue
Block a user