feat(cli): Replace mcp run with click CLI and runtime options
BREAKING CHANGE: FASTMCP_-prefixed env vars have been replaced by CLI
arguments. Refer to the README for updated usage.
Usage: python -m nextcloud_mcp_server.app [OPTIONS]
Options:
-h, --host TEXT
-p, --port INTEGER
-w, --workers INTEGER
-r, --reload
--log-level [critical|error|warning|info|debug|trace]
-t, --transport [sse|streamable-http]
-e, --enable-app [notes|tables|webdav|calendar|contacts]
Enable specific Nextcloud app APIs. Can be
specified multiple times. If not specified,
all apps are enabled.
--help Show this message and exit.
This commit is contained in:
+98
-22
@@ -1,12 +1,17 @@
|
||||
import click
|
||||
import logging
|
||||
import uvicorn
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from dataclasses import dataclass
|
||||
|
||||
from starlette.applications import Starlette
|
||||
from starlette.routing import Mount
|
||||
|
||||
from mcp.server.fastmcp import Context, FastMCP
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
from nextcloud_mcp_server.config import setup_logging
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
from nextcloud_mcp_server.server import (
|
||||
configure_calendar_tools,
|
||||
configure_contacts_tools,
|
||||
@@ -15,7 +20,8 @@ from nextcloud_mcp_server.server import (
|
||||
configure_webdav_tools,
|
||||
)
|
||||
|
||||
setup_logging()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -37,28 +43,98 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
|
||||
await client.close()
|
||||
|
||||
|
||||
# Create an MCP server
|
||||
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan)
|
||||
def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
|
||||
setup_logging()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
# Create an MCP server
|
||||
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan)
|
||||
|
||||
@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
|
||||
return await client.capabilities()
|
||||
|
||||
# Define available apps and their configuration functions
|
||||
available_apps = {
|
||||
"notes": configure_notes_tools,
|
||||
"tables": configure_tables_tools,
|
||||
"webdav": configure_webdav_tools,
|
||||
"calendar": configure_calendar_tools,
|
||||
"contacts": configure_contacts_tools,
|
||||
}
|
||||
|
||||
# If no specific apps are specified, enable all
|
||||
if enabled_apps is None:
|
||||
enabled_apps = list(available_apps.keys())
|
||||
|
||||
# Configure only the enabled apps
|
||||
for app_name in enabled_apps:
|
||||
if app_name in available_apps:
|
||||
logger.info(f"Configuring {app_name} tools")
|
||||
available_apps[app_name](mcp)
|
||||
else:
|
||||
logger.warning(
|
||||
f"Unknown app: {app_name}. Available apps: {list(available_apps.keys())}"
|
||||
)
|
||||
|
||||
mcp_app = mcp.sse_app() if transport == "sse" else mcp.streamable_http_app()
|
||||
|
||||
app = Starlette(routes=[Mount("/", app=mcp_app)])
|
||||
|
||||
return app
|
||||
|
||||
|
||||
@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
|
||||
return await client.capabilities()
|
||||
@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"]),
|
||||
)
|
||||
@click.option(
|
||||
"--transport", "-t", default="sse", type=click.Choice(["sse", "streamable-http"])
|
||||
)
|
||||
@click.option(
|
||||
"--enable-app",
|
||||
"-e",
|
||||
multiple=True,
|
||||
type=click.Choice(["notes", "tables", "webdav", "calendar", "contacts"]),
|
||||
help="Enable specific Nextcloud app APIs. Can be specified multiple times. If not specified, all apps are enabled.",
|
||||
)
|
||||
def run(
|
||||
host: str,
|
||||
port: int,
|
||||
workers: int,
|
||||
reload: bool,
|
||||
log_level: str,
|
||||
transport: str,
|
||||
enable_app: tuple[str, ...],
|
||||
):
|
||||
enabled_apps = list(enable_app) if enable_app else None
|
||||
|
||||
if reload or workers:
|
||||
app = "nextcloud_mcp_server.app:get_app"
|
||||
factory = True
|
||||
else:
|
||||
app = get_app(transport=transport, enabled_apps=enabled_apps)
|
||||
factory = False
|
||||
|
||||
uvicorn.run(
|
||||
app=app,
|
||||
factory=factory,
|
||||
host=host,
|
||||
port=port,
|
||||
reload=reload,
|
||||
workers=workers,
|
||||
log_level=log_level,
|
||||
)
|
||||
|
||||
|
||||
configure_notes_tools(mcp)
|
||||
configure_tables_tools(mcp)
|
||||
configure_webdav_tools(mcp)
|
||||
configure_calendar_tools(mcp)
|
||||
configure_contacts_tools(mcp)
|
||||
|
||||
|
||||
def run():
|
||||
mcp.run()
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
|
||||
Reference in New Issue
Block a user