SharePoint MCP Server

"""SharePoint MCP Server Provides tools for interacting with Microsoft SharePoint via Graph API. Uses FastMCP for the high-level decorator-based API. """ import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass from typing import Any from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession from .auth import AuthManager from .config import Config, load_config from .graph import GraphClient from .models import ( FileContent, FileListResult, FileMetadata, Library, ListItem, SearchResult, SharingLink, Site, UploadResult, ) logger = logging.getLogger(__name__) @dataclass class AppContext: """Application context with initialized dependencies.""" config: Config auth: AuthManager graph: GraphClient @asynccontextmanager async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """Initialize auth and Graph client on startup. Args: server: FastMCP server instance Yields: AppContext with initialized services """ config = load_config() # Set up logging logging.basicConfig( level=getattr(logging, config.log_level), format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) logger.info("Initializing SharePoint MCP server") auth = AuthManager(config) # Ensure we have valid tokens (triggers OAuth if needed) await auth.ensure_authenticated() graph = GraphClient(auth) try: yield AppContext(config=config, auth=auth, graph=graph) finally: logger.info("Shutting down SharePoint MCP server") await graph.close() # Create FastMCP server with lifespan mcp = FastMCP( "SharePoint", instructions=""" SharePoint MCP server for file and site management. Use list_sites first to discover available sites, then explore libraries and files. Authentication is handled automatically via OAuth. Common workflows: 1. Browse files: list_sites → list_libraries → list_files → get_file_content 2. Search: search_files with a query 3. Upload: list_sites → list_libraries → upload_file 4. Share: get_file_metadata → create_sharing_link """, lifespan=app_lifespan, ) # ============================================================================ # SITE TOOLS # ============================================================================ @mcp.tool() async def list_sites( ctx: Context[ServerSession, AppContext], search: str | None = None, ) -> list[Site]: """List SharePoint sites the user has access to. Args: search: Optional search query to filter sites by name Returns: List of accessible SharePoint sites """ graph = ctx.request_context.lifespan_context.graph search_msg = f' matching "{search}"' if search else "" await ctx.info(f"Listing sites{search_msg}") return await graph.get_sites(search=search) @mcp.tool() async def list_libraries( ctx: Context[ServerSession, AppContext], site_id: str, ) -> list[Library]: """List document libraries in a SharePoint site. Args: site_id: The site ID (from list_sites) Returns: List of document libraries in the site """ graph = ctx.request_context.lifespan_context.graph await ctx.info(f"Listing libraries for site {site_id}") return await graph.get_libraries(site_id) # ============================================================================ # FILE TOOLS # ============================================================================ @mcp.tool() async def list_files( ctx: Context[ServerSession, AppContext], site_id: str, library_id: str, folder_path: str = "", cursor: str | None = None, limit: int = 50, ) -> FileListResult: """List files in a document library or folder. Args: site_id: The site ID library_id: The library/drive ID (from list_libraries) folder_path: Path within library (empty for root) cursor: Pagination cursor from previous response limit: Maximum items to return (default 50, max 200) Returns: Paginated list of files with metadata """ graph = ctx.request_context.lifespan_context.graph location = f"{library_id}/{folder_path or 'root'}" await ctx.info(f"Listing files in {location}") return await graph.list_files( site_id=site_id, library_id=library_id, folder_path=folder_path, cursor=cursor, limit=min(limit, 200), ) @mcp.tool() async def get_file_content( ctx: Context[ServerSession, AppContext], site_id: str, file_id: str, ) -> FileContent: """Download and read file content. Args: site_id: The site ID file_id: The file/item ID Returns: File content (text for text files, base64 for binary) Note: Large files (>10MB) will be rejected. """ graph = ctx.request_context.lifespan_context.graph await ctx.info(f"Downloading file {file_id}") # Report progress for large files await ctx.report_progress(progress=0, total=1) content = await graph.download_file(site_id, file_id) await ctx.report_progress(progress=1, total=1) return content @mcp.tool() async def upload_file( ctx: Context[ServerSession, AppContext], site_id: str, library_id: str, file_name: str, content: str, folder_path: str = "", is_base64: bool = False, ) -> UploadResult: """Upload a file to SharePoint. Args: site_id: The site ID library_id: The library/drive ID file_name: Name for the uploaded file content: File content (text or base64-encoded) folder_path: Destination folder path (empty for library root) is_base64: Set True if content is base64-encoded binary Returns: Upload result with file URL """ graph = ctx.request_context.lifespan_context.graph destination = f"{folder_path or 'root'}/{file_name}" await ctx.info(f"Uploading {file_name} to {destination}") return await graph.upload_file( site_id=site_id, library_id=library_id, file_name=file_name, content=content, folder_path=folder_path, is_base64=is_base64, ) @mcp.tool() async def get_file_metadata( ctx: Context[ServerSession, AppContext], site_id: str, file_id: str, ) -> FileMetadata: """Get detailed metadata for a file. Args: site_id: The site ID file_id: The file/item ID Returns: Detailed file metadata including size, dates, and author info """ graph = ctx.request_context.lifespan_context.graph await ctx.info(f"Getting metadata for file {file_id}") return await graph.get_file_metadata(site_id, file_id) @mcp.tool() async def create_sharing_link( ctx: Context[ServerSession, AppContext], site_id: str, file_id: str, link_type: str = "view", expiration_days: int | None = None, ) -> SharingLink: """Create a sharing link for a file. Args: site_id: The site ID file_id: The file/item ID link_type: Type of link - "view" (read-only) or "edit" expiration_days: Optional link expiration in days Returns: Generated sharing link """ graph = ctx.request_context.lifespan_context.graph await ctx.info(f"Creating {link_type} link for file {file_id}") return await graph.create_sharing_link( site_id=site_id, file_id=file_id, link_type=link_type, expiration_days=expiration_days, ) @mcp.tool() async def search_files( ctx: Context[ServerSession, AppContext], query: str, site_id: str | None = None, ) -> list[SearchResult]: """Search for files across SharePoint. Args: query: Search query (supports KQL syntax) site_id: Optional site ID to limit search scope Returns: List of matching files with snippets """ graph = ctx.request_context.lifespan_context.graph scope = f" in site {site_id}" if site_id else " across all sites" await ctx.info(f"Searching for '{query}'{scope}") return await graph.search_files(query=query, site_id=site_id) # ============================================================================ # LIST TOOLS # ============================================================================ @mcp.tool() async def list_items( ctx: Context[ServerSession, AppContext], site_id: str, list_id: str, limit: int = 100, ) -> list[ListItem]: """Get items from a SharePoint list. Args: site_id: The site ID list_id: The list ID limit: Maximum items to return (default 100) Returns: List of items with their field values """ graph = ctx.request_context.lifespan_context.graph await ctx.info(f"Getting items from list {list_id}") return await graph.get_list_items(site_id, list_id, limit) @mcp.tool() async def create_list_item( ctx: Context[ServerSession, AppContext], site_id: str, list_id: str, fields: dict[str, Any], ) -> ListItem: """Create a new item in a SharePoint list. Args: site_id: The site ID list_id: The list ID fields: Field values for the new item (as key-value pairs) Returns: The created list item """ graph = ctx.request_context.lifespan_context.graph await ctx.info(f"Creating item in list {list_id}") return await graph.create_list_item(site_id, list_id, fields) # ============================================================================ # RESOURCE: Current user info # ============================================================================ @mcp.resource("sharepoint://user") async def get_current_user(ctx: Context[ServerSession, AppContext]) -> str: """Get information about the authenticated user. Returns: User information as formatted text """ graph = ctx.request_context.lifespan_context.graph user = await graph.get_current_user() email = user.get("mail", user.get("userPrincipalName")) return f"Authenticated as: {user['displayName']} ({email})" # ============================================================================ # PROMPTS: Common SharePoint tasks # ============================================================================ @mcp.prompt() def find_document(filename: str, site_name: str = "") -> str: """Generate a prompt to find a specific document. Args: filename: Name of the document to find site_name: Optional site name to search within Returns: Formatted prompt """ if site_name: return ( f"Search for the document '{filename}' in the SharePoint site " f"'{site_name}' and show me its details." ) return ( f"Search for the document '{filename}' across all my SharePoint sites " f"and show me where it's located." ) @mcp.prompt() def summarize_library(site_name: str, library_name: str = "Documents") -> str: """Generate a prompt to summarize a document library. Args: site_name: Name of the SharePoint site library_name: Name of the library (default: Documents) Returns: Formatted prompt """ return ( f"List the files in the '{library_name}' library of the '{site_name}' " f"SharePoint site and give me a summary of what's there." )