1032 lines
42 KiB
Python
1032 lines
42 KiB
Python
"""MCP server exposing safe wiki source-file resources."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import hmac
|
|
import os
|
|
import sys
|
|
from email import policy
|
|
from email.parser import BytesParser
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from obsidian_wiki.config import config_value, resolve_vault_path
|
|
from obsidian_wiki.file_assets import clean_config_value, db_path_for_vault, registry_path_for_vault
|
|
from obsidian_wiki.mcp_ingest import (
|
|
DEFAULT_INGEST_MAX_CHUNK_BYTES,
|
|
DEFAULT_INGEST_MAX_FILE_BYTES,
|
|
DEFAULT_INGEST_MAX_FILES_PER_JOB,
|
|
DEFAULT_INGEST_OPENCODE_MODEL,
|
|
DEFAULT_INGEST_OPENCODE_VARIANT,
|
|
DEFAULT_UPLOAD_TICKET_SECONDS,
|
|
DEFAULT_OPENCODE_PATH,
|
|
MAX_UPLOAD_TICKET_SECONDS,
|
|
IngestError,
|
|
IngestManager,
|
|
parse_bool,
|
|
)
|
|
from obsidian_wiki.mcp_query import (
|
|
DEFAULT_QUERY_MAX_CHARS,
|
|
DEFAULT_QUERY_OPENCODE_MODEL,
|
|
DEFAULT_QUERY_OPENCODE_VARIANT,
|
|
QueryError,
|
|
QueryManager,
|
|
)
|
|
from obsidian_wiki.mcp_assets import (
|
|
DEFAULT_MAX_FILE_BYTES,
|
|
DEFAULT_PROFILE_NAME,
|
|
AssetAccessError,
|
|
AssetNotFoundError,
|
|
AssetTooLargeError,
|
|
get_file_asset_metadata as core_get_file_asset_metadata,
|
|
load_registry_by_file_id,
|
|
mcp_runtime_options,
|
|
read_file_asset_resource,
|
|
search_file_assets as core_search_file_assets,
|
|
)
|
|
|
|
DEFAULT_HTTP_HOST = "127.0.0.1"
|
|
DEFAULT_HTTP_PORT = 8765
|
|
DEFAULT_HTTP_PATH = "/mcp"
|
|
DEFAULT_INGEST_UPLOAD_PATH = "/mcp-ingest/uploads"
|
|
DEFAULT_TOKEN_ENV = "OBSIDIAN_MCP_TOKEN"
|
|
MCP_SERVER_NAME = "wiki-file-assets"
|
|
TOOL_FILE_ASSETS = "file-assets"
|
|
TOOL_INGEST = "ingest"
|
|
TOOL_QUERY = "query"
|
|
DEFAULT_ENABLED_TOOLS = {TOOL_FILE_ASSETS}
|
|
AVAILABLE_TOOLS = {TOOL_FILE_ASSETS, TOOL_INGEST, TOOL_QUERY}
|
|
|
|
|
|
def text_note_file_name(title: str | None) -> str:
|
|
text = str(title or "").strip().lower()
|
|
slug = []
|
|
last_dash = False
|
|
for char in text:
|
|
if char.isascii() and char.isalnum():
|
|
slug.append(char)
|
|
last_dash = False
|
|
elif char in {"-", "_"}:
|
|
slug.append(char)
|
|
last_dash = False
|
|
elif not last_dash:
|
|
slug.append("-")
|
|
last_dash = True
|
|
name = "".join(slug).strip("-_")[:80].strip("-_")
|
|
return (name or "thought") + ".md"
|
|
|
|
|
|
def text_note_payload(text: str, title: str | None = None) -> bytes:
|
|
body = str(text).replace("\x00", "").strip()
|
|
if not body:
|
|
raise IngestError("text must not be empty")
|
|
heading = " ".join(str(title or "").replace("\x00", "").split())
|
|
if heading:
|
|
content = f"# {heading}\n\n{body}\n"
|
|
else:
|
|
content = body + "\n"
|
|
return content.encode("utf-8")
|
|
|
|
|
|
def normalize_tool_name(name: str) -> str:
|
|
normalized = name.strip().lower().replace("_", "-")
|
|
aliases = {
|
|
"file": TOOL_FILE_ASSETS,
|
|
"files": TOOL_FILE_ASSETS,
|
|
"file-assets": TOOL_FILE_ASSETS,
|
|
"source-file": TOOL_FILE_ASSETS,
|
|
"source-files": TOOL_FILE_ASSETS,
|
|
"ingest": TOOL_INGEST,
|
|
"upload": TOOL_INGEST,
|
|
"uploads": TOOL_INGEST,
|
|
"opencode-ingest": TOOL_INGEST,
|
|
"query": TOOL_QUERY,
|
|
"wiki-query": TOOL_QUERY,
|
|
"ask": TOOL_QUERY,
|
|
"question": TOOL_QUERY,
|
|
"questions": TOOL_QUERY,
|
|
}
|
|
if normalized not in aliases:
|
|
raise ValueError(f"unknown MCP tool group: {name}")
|
|
return aliases[normalized]
|
|
|
|
|
|
def parse_enabled_tools(raw: str | None) -> set[str]:
|
|
if raw is None or not raw.strip():
|
|
return set(DEFAULT_ENABLED_TOOLS)
|
|
if raw.strip().lower() in {"none", "off", "disabled"}:
|
|
return set()
|
|
return {normalize_tool_name(part) for part in raw.split(",") if part.strip()}
|
|
|
|
|
|
def enabled_tool_names() -> set[str]:
|
|
configured = runtime_config_value("OBSIDIAN_MCP_ENABLED_TOOLS", "")
|
|
if configured:
|
|
return parse_enabled_tools(configured)
|
|
enabled = set(DEFAULT_ENABLED_TOOLS)
|
|
if parse_bool(runtime_config_value("OBSIDIAN_MCP_ENABLE_INGEST", "0")):
|
|
enabled.add(TOOL_INGEST)
|
|
return enabled
|
|
|
|
|
|
def bearer_token(authorization: str) -> str:
|
|
scheme, _, token = authorization.partition(" ")
|
|
if scheme.lower() != "bearer" or not token:
|
|
return ""
|
|
return token.strip()
|
|
|
|
|
|
class BearerTokenAuthMiddleware:
|
|
def __init__(self, app: Any, token: str, *, bypass_paths: set[str] | None = None) -> None:
|
|
if not token:
|
|
raise ValueError("HTTP MCP token must not be empty")
|
|
self.app = app
|
|
self.token = token
|
|
self.bypass_paths = bypass_paths or set()
|
|
|
|
async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
|
|
if scope.get("type") != "http":
|
|
await self.app(scope, receive, send)
|
|
return
|
|
if scope.get("path") in self.bypass_paths:
|
|
await self.app(scope, receive, send)
|
|
return
|
|
headers = {key.lower(): value for key, value in scope.get("headers", [])}
|
|
authorization = headers.get(b"authorization", b"").decode("latin1")
|
|
token = bearer_token(authorization)
|
|
if not hmac.compare_digest(token, self.token):
|
|
await send(
|
|
{
|
|
"type": "http.response.start",
|
|
"status": 401,
|
|
"headers": [
|
|
(b"content-type", b"application/json"),
|
|
(b"www-authenticate", b"Bearer"),
|
|
],
|
|
}
|
|
)
|
|
await send({"type": "http.response.body", "body": b'{"error":"unauthorized"}'})
|
|
return
|
|
await self.app(scope, receive, send)
|
|
|
|
|
|
def create_server(
|
|
*,
|
|
vault_path: str | Path,
|
|
db_path: str | Path | None = None,
|
|
profile_name: str = DEFAULT_PROFILE_NAME,
|
|
max_file_bytes: int = DEFAULT_MAX_FILE_BYTES,
|
|
expose_paths: bool = False,
|
|
host: str = DEFAULT_HTTP_HOST,
|
|
port: int = DEFAULT_HTTP_PORT,
|
|
streamable_http_path: str = DEFAULT_HTTP_PATH,
|
|
allowed_hosts: list[str] | None = None,
|
|
allowed_origins: list[str] | None = None,
|
|
enable_file_assets: bool = True,
|
|
enable_ingest: bool = False,
|
|
enable_query: bool = False,
|
|
project_root: str | Path | None = None,
|
|
ingest_runtime_root: str | Path | None = None,
|
|
query_runtime_root: str | Path | None = None,
|
|
opencode_path: str | Path = DEFAULT_OPENCODE_PATH,
|
|
ingest_opencode_model: str | None = DEFAULT_INGEST_OPENCODE_MODEL,
|
|
ingest_opencode_variant: str | None = DEFAULT_INGEST_OPENCODE_VARIANT,
|
|
query_opencode_model: str | None = DEFAULT_QUERY_OPENCODE_MODEL,
|
|
query_opencode_variant: str | None = DEFAULT_QUERY_OPENCODE_VARIANT,
|
|
ingest_max_file_bytes: int = DEFAULT_INGEST_MAX_FILE_BYTES,
|
|
ingest_max_chunk_bytes: int = DEFAULT_INGEST_MAX_CHUNK_BYTES,
|
|
ingest_max_files_per_job: int = DEFAULT_INGEST_MAX_FILES_PER_JOB,
|
|
query_max_chars: int = DEFAULT_QUERY_MAX_CHARS,
|
|
ingest_manager: IngestManager | None = None,
|
|
query_manager: QueryManager | None = None,
|
|
):
|
|
if sys.version_info < (3, 10):
|
|
raise RuntimeError("obsidian-wiki MCP server requires Python 3.10 or newer")
|
|
try:
|
|
from mcp.server.fastmcp import FastMCP
|
|
try:
|
|
from mcp.server.fastmcp.server import TransportSecuritySettings
|
|
except ImportError:
|
|
TransportSecuritySettings = None
|
|
except ImportError as exc:
|
|
raise RuntimeError('MCP support is not installed. Run: pip install "obsidian-wiki[mcp]"') from exc
|
|
|
|
vault = Path(clean_config_value(str(vault_path))).expanduser()
|
|
db = _resolve_db_path(vault, db_path)
|
|
project = Path(project_root).expanduser().resolve() if project_root is not None else Path.cwd().resolve()
|
|
ingest_runtime = (
|
|
Path(ingest_runtime_root).expanduser().resolve()
|
|
if ingest_runtime_root is not None
|
|
else project / ".runtime" / "mcp-ingest"
|
|
)
|
|
query_runtime = (
|
|
Path(query_runtime_root).expanduser().resolve()
|
|
if query_runtime_root is not None
|
|
else project / ".runtime" / "mcp-query"
|
|
)
|
|
transport_security = None
|
|
if TransportSecuritySettings is not None and (allowed_hosts is not None or allowed_origins is not None):
|
|
transport_security = TransportSecuritySettings(
|
|
allowed_hosts=allowed_hosts or [],
|
|
allowed_origins=allowed_origins or [],
|
|
)
|
|
server = FastMCP(
|
|
MCP_SERVER_NAME,
|
|
host=host,
|
|
port=port,
|
|
streamable_http_path=streamable_http_path,
|
|
transport_security=transport_security,
|
|
)
|
|
|
|
@server.tool()
|
|
def file_index_status() -> dict[str, Any]:
|
|
"""Show the configured vault, file index, registry, and MCP limits."""
|
|
registry = registry_path_for_vault(vault)
|
|
return {
|
|
"vault_path": str(vault),
|
|
"db_path": str(db),
|
|
"db_exists": db.is_file(),
|
|
"registry_path": str(registry),
|
|
"registry_exists": registry.is_file(),
|
|
"source_registry_records": len(load_registry_by_file_id(vault)),
|
|
"profile_name": profile_name,
|
|
"max_file_bytes": max_file_bytes,
|
|
"expose_paths": expose_paths,
|
|
"enabled_tools": sorted(
|
|
([TOOL_FILE_ASSETS] if enable_file_assets else [])
|
|
+ ([TOOL_INGEST] if enable_ingest else [])
|
|
+ ([TOOL_QUERY] if enable_query else [])
|
|
),
|
|
"ingest_enabled": enable_ingest,
|
|
"query_enabled": enable_query,
|
|
"ingest_opencode_model": ingest_opencode_model or "",
|
|
"ingest_opencode_variant": ingest_opencode_variant or "",
|
|
"query_opencode_model": query_opencode_model or "",
|
|
"query_opencode_variant": query_opencode_variant or "",
|
|
}
|
|
|
|
if enable_ingest:
|
|
if ingest_manager is None:
|
|
ingest_manager = IngestManager(
|
|
vault_path=vault,
|
|
project_root=project,
|
|
runtime_root=ingest_runtime,
|
|
opencode_path=opencode_path,
|
|
opencode_model=ingest_opencode_model,
|
|
opencode_variant=ingest_opencode_variant,
|
|
max_file_bytes=ingest_max_file_bytes,
|
|
max_chunk_bytes=ingest_max_chunk_bytes,
|
|
max_files_per_job=ingest_max_files_per_job,
|
|
)
|
|
|
|
@server.tool()
|
|
def get_ingest_upload_instructions() -> dict[str, Any]:
|
|
"""Return preferred upload methods and limits for remote MCP ingest clients."""
|
|
return {
|
|
"preferred_method": "use ingest_text_note for short text; POST multipart/form-data files to the HTTP upload endpoint for files",
|
|
"http_upload": {
|
|
"method": "POST",
|
|
"path": DEFAULT_INGEST_UPLOAD_PATH,
|
|
"file_field": "files",
|
|
"auth": (
|
|
"use the same Authorization: Bearer <token> header as the MCP endpoint, "
|
|
"or call create_ingest_upload_ticket and send X-Obsidian-Wiki-Upload-Token"
|
|
),
|
|
"optional_fields": ["project", "doc_type_hint", "note", "start_job"],
|
|
"start_job_default": True,
|
|
"curl_example": (
|
|
"curl -H 'Authorization: Bearer <token>' "
|
|
"-F 'files=@drawing.pdf' -F 'project=bjd' "
|
|
"-F 'doc_type_hint=drawing' -F 'start_job=true' "
|
|
"http://<host>:<port>/mcp-ingest/uploads"
|
|
),
|
|
},
|
|
"upload_ticket_tool": {
|
|
"name": "create_ingest_upload_ticket",
|
|
"token_header": "X-Obsidian-Wiki-Upload-Token",
|
|
"best_for": "remote agents that can run curl but cannot read the main MCP bearer token",
|
|
"curl_example": (
|
|
"curl -H 'X-Obsidian-Wiki-Upload-Token: <upload-token>' "
|
|
"-F 'files=@drawing.pdf' http://<host>:<port>/mcp-ingest/uploads"
|
|
),
|
|
},
|
|
"direct_text_tool": {
|
|
"name": "ingest_text_note",
|
|
"best_for": "short remote-agent thoughts, notes, findings, or pasted text",
|
|
"start_job_default": True,
|
|
"notes": "wraps text as a temporary markdown upload, then uses the same fixed wiki-ingest job path",
|
|
},
|
|
"limits": {
|
|
"max_file_bytes": ingest_manager.max_file_bytes,
|
|
"max_chunk_bytes": ingest_manager.max_chunk_bytes,
|
|
"max_files_per_job": ingest_manager.max_files_per_job,
|
|
},
|
|
"staging": {
|
|
"runtime_upload_root": str(ingest_manager.upload_root),
|
|
"vault_path": str(ingest_manager.vault),
|
|
"note": "uploads are temporary staging files outside the vault; wiki-ingest decides final archive/index placement",
|
|
},
|
|
}
|
|
|
|
@server.tool()
|
|
def ingest_text_note(
|
|
text: str,
|
|
title: str | None = None,
|
|
project: str | None = None,
|
|
doc_type_hint: str | None = "thought",
|
|
note: str | None = None,
|
|
start_job: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Ingest a short text note by wrapping it as a temporary markdown upload."""
|
|
payload = text_note_payload(text, title=title)
|
|
upload = ingest_manager.create_completed_upload(
|
|
file_name=text_note_file_name(title),
|
|
payload=payload,
|
|
project=project,
|
|
doc_type_hint=doc_type_hint or "thought",
|
|
)
|
|
job = ingest_manager.start_ingest_job([upload["upload_id"]], note=note) if start_job else None
|
|
return {"uploads": [upload], "job": job, "text_bytes": len(payload)}
|
|
|
|
@server.tool()
|
|
def create_ingest_upload_ticket(
|
|
expires_seconds: int = DEFAULT_UPLOAD_TICKET_SECONDS,
|
|
max_files: int | None = None,
|
|
project: str | None = None,
|
|
doc_type_hint: str | None = None,
|
|
note: str | None = None,
|
|
start_job: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Create a short-lived HTTP upload ticket so remote agents can upload files without the main MCP token."""
|
|
ticket = ingest_manager.create_upload_ticket(
|
|
expires_seconds=expires_seconds,
|
|
max_files=max_files,
|
|
project=project,
|
|
doc_type_hint=doc_type_hint,
|
|
note=note,
|
|
start_job=start_job,
|
|
)
|
|
return {
|
|
**ticket,
|
|
"upload_path": DEFAULT_INGEST_UPLOAD_PATH,
|
|
"token_header": "X-Obsidian-Wiki-Upload-Token",
|
|
"expires_seconds_max": MAX_UPLOAD_TICKET_SECONDS,
|
|
"curl_example": (
|
|
"curl -H 'X-Obsidian-Wiki-Upload-Token: "
|
|
+ ticket["upload_token"]
|
|
+ "' -F 'files=@/path/to/file.pdf' http://<host>:<port>/mcp-ingest/uploads"
|
|
),
|
|
}
|
|
|
|
@server.tool()
|
|
def start_ingest_job(upload_ids: list[str], note: str | None = None) -> dict[str, Any]:
|
|
"""Start or queue a local OpenCode wiki-ingest job for completed uploads."""
|
|
return ingest_manager.start_ingest_job(upload_ids, note=note)
|
|
|
|
@server.tool()
|
|
def get_ingest_job_status(job_id: str) -> dict[str, Any]:
|
|
"""Return state, log tail, and exit status for an MCP ingest job."""
|
|
return ingest_manager.get_ingest_job_status(job_id)
|
|
|
|
@server.tool()
|
|
def get_ingest_job_result(job_id: str) -> dict[str, Any]:
|
|
"""Return the final or current summary for an MCP ingest job."""
|
|
return ingest_manager.get_ingest_job_result(job_id)
|
|
|
|
@server.tool()
|
|
def cancel_ingest_job(job_id: str) -> dict[str, Any]:
|
|
"""Cancel a queued or running MCP ingest job."""
|
|
return ingest_manager.cancel_ingest_job(job_id)
|
|
|
|
@server.tool()
|
|
def check_ingest_opencode_readiness(run_smoke_test: bool = False) -> dict[str, Any]:
|
|
"""Check whether local OpenCode can load the wiki-ingest skill for MCP ingest jobs."""
|
|
return ingest_manager.check_opencode_readiness(run_smoke_test=run_smoke_test)
|
|
|
|
if enable_query:
|
|
if query_manager is None:
|
|
query_manager = QueryManager(
|
|
vault_path=vault,
|
|
project_root=project,
|
|
runtime_root=query_runtime,
|
|
opencode_path=opencode_path,
|
|
opencode_model=query_opencode_model,
|
|
opencode_variant=query_opencode_variant,
|
|
max_query_chars=query_max_chars,
|
|
)
|
|
|
|
@server.tool()
|
|
def start_wiki_query(
|
|
query: str,
|
|
mode: str = "normal",
|
|
public_only: bool = False,
|
|
note: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Start or queue a local OpenCode wiki-query job."""
|
|
return query_manager.start_query_job(
|
|
query,
|
|
mode=mode,
|
|
public_only=public_only,
|
|
note=note,
|
|
)
|
|
|
|
@server.tool()
|
|
def get_wiki_query_status(job_id: str) -> dict[str, Any]:
|
|
"""Return state, log tail, and exit status for an MCP wiki query job."""
|
|
return query_manager.get_query_job_status(job_id)
|
|
|
|
@server.tool()
|
|
def get_wiki_query_result(job_id: str) -> dict[str, Any]:
|
|
"""Return the final or current answer for an MCP wiki query job."""
|
|
return query_manager.get_query_job_result(job_id)
|
|
|
|
@server.tool()
|
|
def cancel_wiki_query(job_id: str) -> dict[str, Any]:
|
|
"""Cancel a queued or running MCP wiki query job."""
|
|
return query_manager.cancel_query_job(job_id)
|
|
|
|
@server.tool()
|
|
def check_query_opencode_readiness(run_smoke_test: bool = False) -> dict[str, Any]:
|
|
"""Check whether local OpenCode can load the wiki-query skill for MCP query jobs."""
|
|
return query_manager.check_opencode_readiness(run_smoke_test=run_smoke_test)
|
|
|
|
if enable_file_assets:
|
|
|
|
@server.tool()
|
|
def search_file_assets(
|
|
query: str,
|
|
doc_type: str | None = None,
|
|
source_type: str | None = None,
|
|
product_line: str | None = None,
|
|
customer: str | None = None,
|
|
competitor: str | None = None,
|
|
limit: int = 10,
|
|
) -> list[dict[str, Any]]:
|
|
"""Search safe engineering source files such as drawings, 3D models, specs, and test reports."""
|
|
return _as_mcp_results(
|
|
core_search_file_assets(
|
|
vault,
|
|
query,
|
|
db_path=db,
|
|
limit=limit,
|
|
doc_type=doc_type,
|
|
source_type=source_type,
|
|
product_line=product_line,
|
|
customer=customer,
|
|
competitor=competitor,
|
|
profile_name=profile_name,
|
|
expose_paths=expose_paths,
|
|
)
|
|
)
|
|
|
|
@server.tool()
|
|
def list_file_assets(
|
|
doc_type: str | None = None,
|
|
source_type: str | None = None,
|
|
product_line: str | None = None,
|
|
customer: str | None = None,
|
|
competitor: str | None = None,
|
|
limit: int = 10,
|
|
) -> list[dict[str, Any]]:
|
|
"""List safe source files matching optional filters. Use search_file_assets when the user gives keywords such as product codes, sizes, or drawing names."""
|
|
return _as_mcp_results(
|
|
core_search_file_assets(
|
|
vault,
|
|
"",
|
|
db_path=db,
|
|
limit=limit,
|
|
doc_type=doc_type,
|
|
source_type=source_type,
|
|
product_line=product_line,
|
|
customer=customer,
|
|
competitor=competitor,
|
|
profile_name=profile_name,
|
|
expose_paths=expose_paths,
|
|
)
|
|
)
|
|
|
|
@server.tool()
|
|
def get_file_asset_metadata(file_id: str) -> dict[str, Any]:
|
|
"""Return safe metadata and the resource URI for one indexed source file."""
|
|
return _as_mcp_result(
|
|
core_get_file_asset_metadata(
|
|
vault,
|
|
file_id,
|
|
db_path=db,
|
|
profile_name=profile_name,
|
|
expose_paths=expose_paths,
|
|
)
|
|
)
|
|
|
|
@server.resource("asset://source/sha256/{digest}", mime_type="application/octet-stream")
|
|
def read_file_asset(digest: str) -> bytes:
|
|
"""Read one allowed source file as a binary MCP resource."""
|
|
resource = read_file_asset_resource(
|
|
vault,
|
|
f"asset://source/sha256/{digest}",
|
|
db_path=db,
|
|
profile_name=profile_name,
|
|
max_file_bytes=max_file_bytes,
|
|
)
|
|
return resource["bytes"]
|
|
|
|
@server.prompt(
|
|
"how_to_use_wiki_file_assets",
|
|
description="Explain how to search and retrieve original source files from the wiki-file-assets MCP server.",
|
|
)
|
|
def how_to_use_wiki_file_assets() -> str:
|
|
return (
|
|
"wiki-file-assets serves original source files archived from an Obsidian wiki vault. "
|
|
"Use search_file_assets for keyword searches such as product codes, drawing names, sizes, "
|
|
"or document types. Use list_file_assets to browse allowed engineering files with optional "
|
|
"filters. Use get_file_asset_metadata with a file_id from search/list results to get the "
|
|
"asset://source/sha256/... URI, then read that resource to fetch the original file bytes. "
|
|
"This MCP is intentionally limited to safe engineering files such as drawings, CAD/3D models, "
|
|
"specifications, product parameters, and test reports; it blocks raw orders, prices, email "
|
|
"correspondence, complaint evidence, claims, and personal data."
|
|
)
|
|
|
|
return server
|
|
|
|
|
|
def read_token_from_env(env_name: str = DEFAULT_TOKEN_ENV) -> str:
|
|
token = os.environ.get(env_name, "").strip()
|
|
if not token:
|
|
raise RuntimeError(f"{env_name} must be set for HTTP MCP transport")
|
|
return token
|
|
|
|
|
|
def runtime_config_value(key: str, default: str) -> str:
|
|
raw = os.environ.get(key)
|
|
if raw is None:
|
|
raw = config_value(key, default)
|
|
value = clean_config_value(raw).strip()
|
|
return value or default
|
|
|
|
|
|
def runtime_config_int(key: str, default: int) -> int:
|
|
raw = runtime_config_value(key, str(default))
|
|
try:
|
|
return int(raw)
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
class IngestUploadRouter:
|
|
def __init__(
|
|
self,
|
|
app: Any,
|
|
*,
|
|
ingest_manager: IngestManager | None = None,
|
|
upload_path: str = DEFAULT_INGEST_UPLOAD_PATH,
|
|
bearer_auth_token: str = "",
|
|
) -> None:
|
|
self.app = app
|
|
self.ingest_manager = ingest_manager
|
|
self.upload_path = upload_path
|
|
self.bearer_auth_token = bearer_auth_token
|
|
|
|
async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
|
|
if scope.get("type") != "http" or scope.get("path") != self.upload_path:
|
|
await self.app(scope, receive, send)
|
|
return
|
|
try:
|
|
upload_ticket = self._authorize_upload(scope)
|
|
except IngestError:
|
|
await _send_json(send, 401, {"error": "unauthorized"})
|
|
return
|
|
if scope.get("method", "").upper() != "POST":
|
|
await _send_json(send, 405, {"error": "method_not_allowed"})
|
|
return
|
|
if self.ingest_manager is None:
|
|
await _send_json(send, 404, {"error": "ingest_not_enabled"})
|
|
return
|
|
try:
|
|
data = await self._handle_upload(scope, receive, upload_ticket=upload_ticket)
|
|
except IngestError as exc:
|
|
await _send_json(send, 400, {"error": str(exc)})
|
|
return
|
|
await _send_json(send, 200, data)
|
|
|
|
def _authorize_upload(self, scope: dict[str, Any]) -> dict[str, Any] | None:
|
|
authorization = _header_value(scope, b"authorization")
|
|
token = bearer_token(authorization)
|
|
if self.bearer_auth_token and hmac.compare_digest(token, self.bearer_auth_token):
|
|
return None
|
|
if self.ingest_manager is None:
|
|
raise IngestError("unauthorized")
|
|
upload_token = _header_value(scope, b"x-obsidian-wiki-upload-token")
|
|
if not upload_token:
|
|
upload_token = _query_param(scope, "upload_token")
|
|
if not upload_token:
|
|
raise IngestError("unauthorized")
|
|
return self.ingest_manager.consume_upload_ticket(upload_token)
|
|
|
|
async def _handle_upload(
|
|
self,
|
|
scope: dict[str, Any],
|
|
receive: Any,
|
|
*,
|
|
upload_ticket: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
body = await _read_http_body(receive)
|
|
content_type = _header_value(scope, b"content-type")
|
|
fields, files = _parse_multipart_form(content_type, body)
|
|
if not files:
|
|
raise IngestError("multipart field 'files' must include at least one file")
|
|
manager = self.ingest_manager
|
|
if manager is None:
|
|
raise IngestError("ingest is not enabled")
|
|
max_files = int(upload_ticket.get("max_files", manager.max_files_per_job)) if upload_ticket else manager.max_files_per_job
|
|
if len(files) > max_files:
|
|
raise IngestError(f"too many files for one ingest job: {len(files)} > {max_files}")
|
|
project = _first_field(fields, "project") or (upload_ticket or {}).get("project")
|
|
doc_type_hint = _first_field(fields, "doc_type_hint") or (upload_ticket or {}).get("doc_type_hint")
|
|
note = _first_field(fields, "note") or (upload_ticket or {}).get("note")
|
|
start_default = bool((upload_ticket or {}).get("start_job", True))
|
|
start_job = parse_bool(_first_field(fields, "start_job"), default=start_default)
|
|
uploads = [
|
|
manager.create_completed_upload(
|
|
file_name=file["file_name"],
|
|
payload=file["payload"],
|
|
project=project,
|
|
doc_type_hint=doc_type_hint,
|
|
)
|
|
for file in files
|
|
]
|
|
job = manager.start_ingest_job([upload["upload_id"] for upload in uploads], note=note) if start_job else None
|
|
return {"uploads": uploads, "job": job}
|
|
|
|
|
|
def create_http_app(
|
|
server: Any,
|
|
token: str,
|
|
*,
|
|
ingest_manager: IngestManager | None = None,
|
|
upload_path: str = DEFAULT_INGEST_UPLOAD_PATH,
|
|
) -> Any:
|
|
app = server.streamable_http_app() if hasattr(server, "streamable_http_app") else server
|
|
app = IngestUploadRouter(
|
|
app,
|
|
ingest_manager=ingest_manager,
|
|
upload_path=upload_path,
|
|
bearer_auth_token=token,
|
|
)
|
|
return BearerTokenAuthMiddleware(app, token, bypass_paths={upload_path})
|
|
|
|
|
|
async def _read_http_body(receive: Any) -> bytes:
|
|
chunks: list[bytes] = []
|
|
while True:
|
|
message = await receive()
|
|
if message.get("type") == "http.disconnect":
|
|
break
|
|
if message.get("type") != "http.request":
|
|
continue
|
|
chunks.append(message.get("body", b""))
|
|
if not message.get("more_body", False):
|
|
break
|
|
return b"".join(chunks)
|
|
|
|
|
|
def _header_value(scope: dict[str, Any], name: bytes) -> str:
|
|
headers = {key.lower(): value for key, value in scope.get("headers", [])}
|
|
return headers.get(name.lower(), b"").decode("latin1")
|
|
|
|
|
|
def _query_param(scope: dict[str, Any], name: str) -> str:
|
|
from urllib.parse import parse_qs
|
|
|
|
raw = scope.get("query_string", b"")
|
|
query = raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else str(raw)
|
|
values = parse_qs(query).get(name, [])
|
|
return values[0] if values else ""
|
|
|
|
|
|
def _parse_multipart_form(content_type: str, body: bytes) -> tuple[dict[str, list[str]], list[dict[str, Any]]]:
|
|
if not content_type.lower().startswith("multipart/form-data"):
|
|
raise IngestError("content-type must be multipart/form-data")
|
|
message = BytesParser(policy=policy.default).parsebytes(
|
|
b"Content-Type: "
|
|
+ content_type.encode("latin1")
|
|
+ b"\r\nMIME-Version: 1.0\r\n\r\n"
|
|
+ body
|
|
)
|
|
if not message.is_multipart():
|
|
raise IngestError("multipart request body is invalid")
|
|
fields: dict[str, list[str]] = {}
|
|
files: list[dict[str, Any]] = []
|
|
for part in message.iter_parts():
|
|
if part.get_content_disposition() != "form-data":
|
|
continue
|
|
field_name = part.get_param("name", header="content-disposition")
|
|
if not field_name:
|
|
continue
|
|
payload = part.get_payload(decode=True)
|
|
if payload is None:
|
|
content = part.get_content()
|
|
payload = content.encode(part.get_content_charset() or "utf-8") if isinstance(content, str) else bytes(content)
|
|
file_name = part.get_filename()
|
|
if file_name is None:
|
|
fields.setdefault(field_name, []).append(payload.decode(part.get_content_charset() or "utf-8", errors="replace"))
|
|
continue
|
|
if field_name != "files":
|
|
continue
|
|
files.append({"file_name": file_name, "payload": payload})
|
|
return fields, files
|
|
|
|
|
|
def _first_field(fields: dict[str, list[str]], name: str) -> str | None:
|
|
values = fields.get(name) or []
|
|
if not values:
|
|
return None
|
|
return values[0]
|
|
|
|
|
|
async def _send_json(send: Any, status: int, payload: dict[str, Any]) -> None:
|
|
body = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
|
|
await send(
|
|
{
|
|
"type": "http.response.start",
|
|
"status": status,
|
|
"headers": [
|
|
(b"content-type", b"application/json"),
|
|
(b"content-length", str(len(body)).encode("ascii")),
|
|
],
|
|
}
|
|
)
|
|
await send({"type": "http.response.body", "body": body})
|
|
|
|
|
|
def run_http_server(
|
|
*,
|
|
vault_path: str | Path,
|
|
db_path: str | Path | None,
|
|
profile_name: str,
|
|
max_file_bytes: int,
|
|
expose_paths: bool,
|
|
host: str,
|
|
port: int,
|
|
token_env: str,
|
|
streamable_http_path: str,
|
|
allowed_hosts: list[str] | None,
|
|
allowed_origins: list[str] | None,
|
|
enable_file_assets: bool = True,
|
|
enable_ingest: bool = False,
|
|
enable_query: bool = False,
|
|
project_root: str | Path | None = None,
|
|
ingest_runtime_root: str | Path | None = None,
|
|
query_runtime_root: str | Path | None = None,
|
|
opencode_path: str | Path = DEFAULT_OPENCODE_PATH,
|
|
ingest_opencode_model: str | None = DEFAULT_INGEST_OPENCODE_MODEL,
|
|
ingest_opencode_variant: str | None = DEFAULT_INGEST_OPENCODE_VARIANT,
|
|
query_opencode_model: str | None = DEFAULT_QUERY_OPENCODE_MODEL,
|
|
query_opencode_variant: str | None = DEFAULT_QUERY_OPENCODE_VARIANT,
|
|
ingest_max_file_bytes: int = DEFAULT_INGEST_MAX_FILE_BYTES,
|
|
ingest_max_chunk_bytes: int = DEFAULT_INGEST_MAX_CHUNK_BYTES,
|
|
ingest_max_files_per_job: int = DEFAULT_INGEST_MAX_FILES_PER_JOB,
|
|
query_max_chars: int = DEFAULT_QUERY_MAX_CHARS,
|
|
) -> None:
|
|
token = read_token_from_env(token_env)
|
|
try:
|
|
import uvicorn
|
|
except ImportError as exc:
|
|
raise RuntimeError('HTTP MCP support requires uvicorn. Run: pip install "obsidian-wiki[mcp]"') from exc
|
|
|
|
ingest_manager = None
|
|
if enable_ingest:
|
|
project = Path(project_root).expanduser().resolve() if project_root is not None else Path.cwd().resolve()
|
|
ingest_runtime = (
|
|
Path(ingest_runtime_root).expanduser().resolve()
|
|
if ingest_runtime_root is not None
|
|
else project / ".runtime" / "mcp-ingest"
|
|
)
|
|
ingest_manager = IngestManager(
|
|
vault_path=vault_path,
|
|
project_root=project,
|
|
runtime_root=ingest_runtime,
|
|
opencode_path=opencode_path,
|
|
opencode_model=ingest_opencode_model,
|
|
opencode_variant=ingest_opencode_variant,
|
|
max_file_bytes=ingest_max_file_bytes,
|
|
max_chunk_bytes=ingest_max_chunk_bytes,
|
|
max_files_per_job=ingest_max_files_per_job,
|
|
)
|
|
|
|
query_manager = None
|
|
if enable_query:
|
|
project = Path(project_root).expanduser().resolve() if project_root is not None else Path.cwd().resolve()
|
|
query_runtime = (
|
|
Path(query_runtime_root).expanduser().resolve()
|
|
if query_runtime_root is not None
|
|
else project / ".runtime" / "mcp-query"
|
|
)
|
|
query_manager = QueryManager(
|
|
vault_path=vault_path,
|
|
project_root=project,
|
|
runtime_root=query_runtime,
|
|
opencode_path=opencode_path,
|
|
opencode_model=query_opencode_model,
|
|
opencode_variant=query_opencode_variant,
|
|
max_query_chars=query_max_chars,
|
|
)
|
|
|
|
server = create_server(
|
|
vault_path=vault_path,
|
|
db_path=db_path,
|
|
profile_name=profile_name,
|
|
max_file_bytes=max_file_bytes,
|
|
expose_paths=expose_paths,
|
|
host=host,
|
|
port=port,
|
|
streamable_http_path=streamable_http_path,
|
|
allowed_hosts=allowed_hosts,
|
|
allowed_origins=allowed_origins,
|
|
enable_file_assets=enable_file_assets,
|
|
enable_ingest=enable_ingest,
|
|
enable_query=enable_query,
|
|
project_root=project_root,
|
|
ingest_runtime_root=ingest_runtime_root,
|
|
query_runtime_root=query_runtime_root,
|
|
opencode_path=opencode_path,
|
|
ingest_opencode_model=ingest_opencode_model,
|
|
ingest_opencode_variant=ingest_opencode_variant,
|
|
query_opencode_model=query_opencode_model,
|
|
query_opencode_variant=query_opencode_variant,
|
|
ingest_max_file_bytes=ingest_max_file_bytes,
|
|
ingest_max_chunk_bytes=ingest_max_chunk_bytes,
|
|
ingest_max_files_per_job=ingest_max_files_per_job,
|
|
query_max_chars=query_max_chars,
|
|
ingest_manager=ingest_manager,
|
|
query_manager=query_manager,
|
|
)
|
|
app = create_http_app(server, token, ingest_manager=ingest_manager)
|
|
uvicorn.run(app, host=host, port=port, log_level="info")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="obsidian-wiki-mcp", description="Run the obsidian-wiki MCP server")
|
|
parser.add_argument("--vault", metavar="PATH", help="absolute path to the Obsidian vault")
|
|
parser.add_argument("--db", metavar="PATH", help="SQLite file asset index path")
|
|
parser.add_argument("--profile", default=None, help=f"bot profile name, default: {DEFAULT_PROFILE_NAME}")
|
|
parser.add_argument("--max-file-bytes", type=int, default=None, help="maximum resource size, 0 disables the limit")
|
|
parser.add_argument("--expose-paths", action="store_true", help="include archive/absolute paths in tool output")
|
|
parser.add_argument(
|
|
"--transport",
|
|
choices=["stdio", "streamable-http"],
|
|
default=None,
|
|
help="MCP transport. HTTP mode requires a bearer token.",
|
|
)
|
|
parser.add_argument("--host", default=None, help="HTTP bind host. Used only with --transport streamable-http.")
|
|
parser.add_argument("--port", default=None, type=int, help="HTTP port. Used only with --transport streamable-http.")
|
|
parser.add_argument("--path", default=None, help="Streamable HTTP MCP path.")
|
|
parser.add_argument("--token-env", default=None, help="Environment variable containing the HTTP bearer token.")
|
|
parser.add_argument(
|
|
"--allowed-host",
|
|
action="append",
|
|
dest="allowed_hosts",
|
|
help="Allowed HTTP Host pattern for DNS rebinding protection. Can be repeated.",
|
|
)
|
|
parser.add_argument(
|
|
"--allowed-origin",
|
|
action="append",
|
|
dest="allowed_origins",
|
|
help="Allowed Origin pattern for DNS rebinding protection. Can be repeated.",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
vault = resolve_vault_path(args.vault)
|
|
if not vault:
|
|
print("error: vault path is not configured; run `obsidian-wiki setup --vault /path/to/vault`", file=sys.stderr)
|
|
return 1
|
|
options = mcp_runtime_options(
|
|
profile_name=args.profile,
|
|
max_file_bytes=args.max_file_bytes,
|
|
expose_paths=args.expose_paths,
|
|
)
|
|
transport = args.transport or runtime_config_value("OBSIDIAN_MCP_TRANSPORT", "stdio")
|
|
host = args.host or runtime_config_value("OBSIDIAN_MCP_HOST", DEFAULT_HTTP_HOST)
|
|
port = args.port if args.port is not None else runtime_config_int("OBSIDIAN_MCP_PORT", DEFAULT_HTTP_PORT)
|
|
http_path = args.path or runtime_config_value("OBSIDIAN_MCP_PATH", DEFAULT_HTTP_PATH)
|
|
token_env = args.token_env or runtime_config_value("OBSIDIAN_MCP_TOKEN_ENV", DEFAULT_TOKEN_ENV)
|
|
enabled_tools = enabled_tool_names()
|
|
enable_file_assets = TOOL_FILE_ASSETS in enabled_tools
|
|
enable_ingest = TOOL_INGEST in enabled_tools
|
|
enable_query = TOOL_QUERY in enabled_tools
|
|
ingest_max_file_bytes = runtime_config_int("OBSIDIAN_MCP_INGEST_MAX_FILE_BYTES", DEFAULT_INGEST_MAX_FILE_BYTES)
|
|
ingest_max_chunk_bytes = runtime_config_int("OBSIDIAN_MCP_INGEST_MAX_CHUNK_BYTES", DEFAULT_INGEST_MAX_CHUNK_BYTES)
|
|
ingest_max_files_per_job = runtime_config_int(
|
|
"OBSIDIAN_MCP_INGEST_MAX_FILES_PER_JOB",
|
|
DEFAULT_INGEST_MAX_FILES_PER_JOB,
|
|
)
|
|
opencode_path = runtime_config_value("OBSIDIAN_MCP_OPENCODE_PATH", "")
|
|
if not opencode_path:
|
|
opencode_path = runtime_config_value("OBSIDIAN_MCP_INGEST_OPENCODE_PATH", DEFAULT_OPENCODE_PATH)
|
|
ingest_runtime_root = runtime_config_value(
|
|
"OBSIDIAN_MCP_INGEST_RUNTIME_ROOT",
|
|
str(Path.cwd() / ".runtime" / "mcp-ingest"),
|
|
)
|
|
query_runtime_root = runtime_config_value(
|
|
"OBSIDIAN_MCP_QUERY_RUNTIME_ROOT",
|
|
str(Path.cwd() / ".runtime" / "mcp-query"),
|
|
)
|
|
ingest_opencode_model = runtime_config_value("OBSIDIAN_MCP_INGEST_OPENCODE_MODEL", DEFAULT_INGEST_OPENCODE_MODEL)
|
|
ingest_opencode_variant = runtime_config_value("OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT", DEFAULT_INGEST_OPENCODE_VARIANT)
|
|
query_opencode_model = runtime_config_value("OBSIDIAN_MCP_QUERY_OPENCODE_MODEL", DEFAULT_QUERY_OPENCODE_MODEL)
|
|
query_opencode_variant = runtime_config_value("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT", DEFAULT_QUERY_OPENCODE_VARIANT)
|
|
query_max_chars = runtime_config_int("OBSIDIAN_MCP_QUERY_MAX_CHARS", DEFAULT_QUERY_MAX_CHARS)
|
|
try:
|
|
if transport == "stdio":
|
|
server = create_server(
|
|
vault_path=vault,
|
|
db_path=args.db,
|
|
profile_name=options["profile_name"],
|
|
max_file_bytes=options["max_file_bytes"],
|
|
expose_paths=options["expose_paths"],
|
|
enable_file_assets=enable_file_assets,
|
|
enable_ingest=enable_ingest,
|
|
enable_query=enable_query,
|
|
project_root=Path.cwd(),
|
|
ingest_runtime_root=ingest_runtime_root,
|
|
query_runtime_root=query_runtime_root,
|
|
opencode_path=opencode_path,
|
|
ingest_opencode_model=ingest_opencode_model,
|
|
ingest_opencode_variant=ingest_opencode_variant,
|
|
query_opencode_model=query_opencode_model,
|
|
query_opencode_variant=query_opencode_variant,
|
|
ingest_max_file_bytes=ingest_max_file_bytes,
|
|
ingest_max_chunk_bytes=ingest_max_chunk_bytes,
|
|
ingest_max_files_per_job=ingest_max_files_per_job,
|
|
query_max_chars=query_max_chars,
|
|
)
|
|
try:
|
|
server.run(transport="stdio")
|
|
except TypeError:
|
|
server.run()
|
|
return 0
|
|
if transport != "streamable-http":
|
|
raise RuntimeError(f"unsupported MCP transport: {transport}")
|
|
run_http_server(
|
|
vault_path=vault,
|
|
db_path=args.db,
|
|
profile_name=options["profile_name"],
|
|
max_file_bytes=options["max_file_bytes"],
|
|
expose_paths=options["expose_paths"],
|
|
host=host,
|
|
port=port,
|
|
token_env=token_env,
|
|
streamable_http_path=http_path,
|
|
allowed_hosts=args.allowed_hosts,
|
|
allowed_origins=args.allowed_origins,
|
|
enable_file_assets=enable_file_assets,
|
|
enable_ingest=enable_ingest,
|
|
enable_query=enable_query,
|
|
project_root=Path.cwd(),
|
|
ingest_runtime_root=ingest_runtime_root,
|
|
query_runtime_root=query_runtime_root,
|
|
opencode_path=opencode_path,
|
|
ingest_opencode_model=ingest_opencode_model,
|
|
ingest_opencode_variant=ingest_opencode_variant,
|
|
query_opencode_model=query_opencode_model,
|
|
query_opencode_variant=query_opencode_variant,
|
|
ingest_max_file_bytes=ingest_max_file_bytes,
|
|
ingest_max_chunk_bytes=ingest_max_chunk_bytes,
|
|
ingest_max_files_per_job=ingest_max_files_per_job,
|
|
query_max_chars=query_max_chars,
|
|
)
|
|
except (RuntimeError, FileNotFoundError, AssetAccessError, AssetNotFoundError, AssetTooLargeError, IngestError, QueryError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _resolve_db_path(vault: Path, raw_path: str | Path | None) -> Path:
|
|
if raw_path is None:
|
|
return db_path_for_vault(vault)
|
|
db_path = Path(clean_config_value(str(raw_path))).expanduser()
|
|
if not db_path.is_absolute():
|
|
db_path = vault / db_path
|
|
return db_path
|
|
|
|
|
|
def _as_mcp_results(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return [_as_mcp_result(row) for row in rows]
|
|
|
|
|
|
def _as_mcp_result(row: dict[str, Any]) -> dict[str, Any]:
|
|
resource_uri = str(row.get("resource_uri") or "")
|
|
return {
|
|
**row,
|
|
"resource_link": {
|
|
"uri": resource_uri,
|
|
"name": row.get("original_name") or row.get("title") or resource_uri,
|
|
"mime_type": row.get("mime_type") or "application/octet-stream",
|
|
},
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|