77 lines
2.9 KiB
Python
Executable File
77 lines
2.9 KiB
Python
Executable File
import sys
|
|
import logging
|
|
import socket
|
|
from fastmcp import FastMCP
|
|
|
|
from typing import Optional, List
|
|
from docsorter.selection import find_presentations, resolve_presentations
|
|
from docsorter.pptx_ops import list_slide_titles, list_slide_notes, copy_slides, delete_slides
|
|
from docsorter.logging_utils import setup_logging, notify_log
|
|
|
|
|
|
def main() -> None:
|
|
logger = setup_logging(stream=sys.stderr)
|
|
mcp = FastMCP("docsorter")
|
|
|
|
@mcp.tool()
|
|
async def m_find(start_dir: str, pattern: Optional[str] = None, max_results: int = 50) -> dict:
|
|
notify_log(mcp, "Searching presentations…")
|
|
logger.info(f"Finding presentations with params: start_dir={start_dir}, pattern={pattern}, max_results={max_results}")
|
|
pres = find_presentations(
|
|
start_dir, pattern, max_results
|
|
)
|
|
return {"presentations": pres}
|
|
|
|
@mcp.tool()
|
|
async def m_resolve(start_dir: str, name_or_pattern: str, limit: int = 2) -> dict:
|
|
notify_log(mcp, f"Resolving presentation: {name_or_pattern}")
|
|
logger.info(f"Resolving presentations with params: start_dir={start_dir}, name_or_pattern={name_or_pattern}, limit={limit}")
|
|
pres = resolve_presentations(
|
|
start_dir, name_or_pattern, limit
|
|
)
|
|
return {"presentations": pres}
|
|
|
|
@mcp.tool()
|
|
async def m_list(path: str) -> dict:
|
|
notify_log(mcp, f"Listing slides for: {path}")
|
|
logger.info(f"Listing slides for presentation: {path}")
|
|
slides = list_slide_titles(path)
|
|
return {"slides": slides}
|
|
|
|
@mcp.tool()
|
|
async def m_notes(path: str, slides: Optional[List[int]] = None) -> dict:
|
|
notify_log(mcp, f"Fetching notes for: {path}")
|
|
logger.info(f"Getting notes for presentation: {path}")
|
|
notes = list_slide_notes(path, slides)
|
|
return {"notes": notes}
|
|
|
|
@mcp.tool()
|
|
async def m_copy(src_path: str, dst_path: str, slides: List[int], insert_position: Optional[int] = None) -> dict:
|
|
notify_log(
|
|
mcp,
|
|
f"Copying {len(slides)} slides from "
|
|
f"{src_path} to {dst_path}"
|
|
)
|
|
logger.info(f"Copying slides with params: src_path={src_path}, dst_path={dst_path}, slides={slides}, insert_position={insert_position}")
|
|
report = copy_slides(
|
|
src_path,
|
|
dst_path,
|
|
slides,
|
|
insert_position
|
|
)
|
|
return {"report": report}
|
|
|
|
@mcp.tool()
|
|
async def m_delete(path: str, slides: List[int]) -> dict:
|
|
notify_log(mcp, f"Deleting slides from: {path}")
|
|
logger.info(f"Deleting slides from presentation: {path}")
|
|
report = delete_slides(path, slides)
|
|
return {"report": report}
|
|
|
|
# Centralized error mapping to JSON-RPC codes
|
|
|
|
# Run SSE JSON-RPC
|
|
port = 59001
|
|
host = "0.0.0.0"
|
|
logger.info(f"Docsorter MCP server started at http://{socket.gethostname()}:{port}/sse/")
|
|
mcp.run(transport="sse", host=host, port=port) |