SharePoint MCP Server

# Prompt: Build SharePoint MCP Server with FastMCP ## Project Goal Create a production-ready MCP server for Microsoft SharePoint using **FastMCP** (the high-level API from the official MCP Python SDK). The server enables Claude Desktop and Claude Code to interact with SharePoint through Microsoft Graph API. **Installation options:** 1. **uvx** (recommended for end users): `uvx sharepoint-mcp` — auto-installs from PyPI 2. **uv --directory** (for local development): Clone repo and run with uv 3. **Claude Code**: Direct integration via MCP configuration --- ## Reference Implementation Follow the patterns from the official MCP Python SDK: https://github.com/modelcontextprotocol/python-sdk Key patterns to use: - **FastMCP** for server creation (NOT low-level Server class) - **Decorators**: `@mcp.tool()`, `@mcp.resource()`, `@mcp.prompt()` - **Context parameter** for logging, progress, and session access - **Pydantic models** for structured output validation - **Lifespan management** for auth/client initialization - **Type hints** throughout for automatic schema generation --- ## MCP Tools to Implement | Tool | Description | Structured Output | |------|-------------|-------------------| | `list_sites` | List SharePoint sites user can access | `list[Site]` | | `list_libraries` | Get document libraries for a site | `list[Library]` | | `list_files` | Browse files in library/folder with pagination | `FileListResult` | | `search_files` | Search files across SharePoint | `list[SearchResult]` | | `get_file_content` | Download/read file content | `FileContent` | | `upload_file` | Upload file to a library | `UploadResult` | | `get_file_metadata` | Get detailed file metadata | `FileMetadata` | | `create_sharing_link` | Generate sharing link for file | `SharingLink` | | `list_items` | Get items from SharePoint list | `list[ListItem]` | | `create_list_item` | Add item to SharePoint list | `ListItem` | --- ## Project Structure ``` sharepoint-mcp/ ├── .gitignore ├── .python-version # 3.11 ├── LICENSE # MIT ├── README.md # Comprehensive documentation ├── pyproject.toml # uv/hatch configuration ├── uv.lock # Generated lock file │ ├── src/ │ └── sharepoint_mcp/ │ ├── __init__.py # Version and exports │ ├── __main__.py # Entry point: main() │ ├── server.py # FastMCP server with all tools │ ├── auth.py # MSAL OAuth + keyring storage │ ├── graph.py # Microsoft Graph API client │ ├── config.py # Configuration management │ │ │ └── models/ # Pydantic models for structured output │ ├── __init__.py │ ├── site.py # Site, Library models │ ├── file.py # FileMetadata, FileContent, etc. │ └── list_item.py # ListItem model │ ├── tests/ │ ├── conftest.py # Fixtures │ ├── test_server.py # Server tool tests │ ├── test_auth.py # Auth flow tests │ └── test_graph.py # Graph client tests │ └── docs/ ├── azure-setup.md # Azure AD app registration guide └── troubleshooting.md # Common issues ``` --- ## Implementation Details ### 1. pyproject.toml ```toml [project] name = "sharepoint-mcp" version = "0.1.0" description = "MCP server for Microsoft SharePoint integration via Graph API" readme = "README.md" license = { text = "MIT" } requires-python = ">=3.11" keywords = ["mcp", "sharepoint", "microsoft-graph", "claude"] classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", ] dependencies = [ "mcp[cli]>=1.0.0", "msal>=1.31.0", "httpx>=0.27.0", "keyring>=25.0.0", "platformdirs>=4.0.0", "pydantic>=2.0.0", ] [project.optional-dependencies] dev = [ "pytest>=8.0.0", "pytest-asyncio>=0.23.0", "pytest-cov>=4.1.0", "respx>=0.21.0", "ruff>=0.3.0", "mypy>=1.8.0", ] [project.scripts] sharepoint-mcp = "sharepoint_mcp:main" [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["src/sharepoint_mcp"] [tool.ruff] line-length = 100 target-version = "py311" [tool.mypy] python_version = "3.11" strict = true [tool.pytest.ini_options] asyncio_mode = "auto" ``` ### 2. Pydantic Models (src/sharepoint_mcp/models/) ```python # models/site.py from pydantic import BaseModel, Field class Site(BaseModel): """SharePoint site information.""" id: str = Field(description="Unique site identifier") name: str = Field(description="Site display name") url: str = Field(description="Site URL") description: str | None = Field(default=None, description="Site description") class Library(BaseModel): """Document library in a SharePoint site.""" id: str = Field(description="Library/drive identifier") name: str = Field(description="Library display name") web_url: str = Field(description="URL to access library") item_count: int = Field(default=0, description="Number of items") # models/file.py from pydantic import BaseModel, Field from datetime import datetime class FileMetadata(BaseModel): """File metadata from SharePoint.""" id: str name: str size: int = Field(description="File size in bytes") mime_type: str | None = None web_url: str created_at: datetime modified_at: datetime created_by: str | None = None modified_by: str | None = None class FileContent(BaseModel): """File content result.""" name: str mime_type: str content: str = Field(description="Base64 encoded for binary, plain text for text files") is_text: bool = Field(description="Whether content is plain text or base64") class FileListResult(BaseModel): """Paginated file list result.""" files: list[FileMetadata] next_cursor: str | None = Field(default=None, description="Cursor for next page") total_count: int | None = None class UploadResult(BaseModel): """Result of file upload.""" id: str name: str web_url: str size: int class SharingLink(BaseModel): """Generated sharing link.""" link_url: str link_type: str = Field(description="view, edit, or embed") expires_at: datetime | None = None class SearchResult(BaseModel): """Search result item.""" id: str name: str site_name: str web_url: str snippet: str | None = Field(default=None, description="Content snippet with match") # models/list_item.py from pydantic import BaseModel, Field from typing import Any class ListItem(BaseModel): """SharePoint list item.""" id: str fields: dict[str, Any] = Field(description="Item field values") created_at: datetime modified_at: datetime ``` ### 3. Server Implementation (src/sharepoint_mcp/server.py) Use **FastMCP with lifespan** for auth/client setup: ```python """ SharePoint MCP Server Provides tools for interacting with Microsoft SharePoint via Graph API. Uses FastMCP for the high-level decorator-based API. """ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession from .auth import AuthManager from .graph import GraphClient from .config import load_config, Config from .models import ( Site, Library, FileMetadata, FileContent, FileListResult, UploadResult, SharingLink, SearchResult, ListItem ) @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.""" config = load_config() 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: 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. """, 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 await ctx.info(f"Listing sites{f' matching \"{search}\"' if search else ''}") 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 await ctx.info(f"Listing files in {library_id}/{folder_path or 'root'}") 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 truncated. """ 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, message="Downloading...") content = await graph.download_file(site_id, file_id) await ctx.report_progress(progress=1, total=1, message="Complete") 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 await ctx.info(f"Uploading {file_name} to {folder_path or 'root'}") 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 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 await ctx.info(f"Searching for '{query}'") 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, ) -> 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.""" graph = ctx.request_context.lifespan_context.graph user = await graph.get_current_user() return f"Authenticated as: {user['displayName']} ({user['mail']})" # ============================================================================ # PROMPT: Common SharePoint tasks # ============================================================================ @mcp.prompt() def find_document(filename: str, site_name: str = "") -> str: """Generate a prompt to find a specific document.""" if site_name: return f"Search for the document '{filename}' in the SharePoint site '{site_name}' and show me its details." return f"Search for the document '{filename}' across all my SharePoint sites 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.""" return f"List the files in the '{library_name}' library of the '{site_name}' SharePoint site and give me a summary of what's there." ``` ### 4. Authentication (src/sharepoint_mcp/auth.py) ```python """ OAuth authentication using MSAL with keyring token storage. Implements: - Browser-based OAuth flow with PKCE - Token storage in OS keychain (macOS Keychain, Windows Credential Manager, etc.) - Automatic token refresh """ import asyncio import base64 import hashlib import json import logging import secrets import webbrowser from datetime import datetime, timedelta from http.server import HTTPServer, BaseHTTPRequestHandler from typing import Any from urllib.parse import parse_qs, urlparse import keyring import msal from .config import Config logger = logging.getLogger(__name__) KEYRING_SERVICE = "sharepoint-mcp" SCOPES = [ "https://graph.microsoft.com/Sites.Read.All", "https://graph.microsoft.com/Files.ReadWrite.All", "https://graph.microsoft.com/User.Read", "offline_access", ] class AuthManager: """Manages OAuth authentication and token lifecycle.""" def __init__(self, config: Config): self.config = config self.app = msal.PublicClientApplication( client_id=config.client_id, authority=f"https://login.microsoftonline.com/{config.tenant_id}", ) self._token_cache: dict[str, Any] | None = None async def ensure_authenticated(self) -> str: """Ensure we have a valid access token. Returns the token.""" # Try to get cached token token_data = self._load_token_from_keyring() if token_data: # Check if token is still valid (with 5 min buffer) expires_at = datetime.fromisoformat(token_data["expires_at"]) if expires_at > datetime.now() + timedelta(minutes=5): return token_data["access_token"] # Try to refresh if "refresh_token" in token_data: try: new_token = await self._refresh_token(token_data["refresh_token"]) return new_token except Exception as e: logger.warning(f"Token refresh failed: {e}") # Need fresh authentication return await self._run_oauth_flow() async def get_access_token(self) -> str: """Get current access token, refreshing if needed.""" return await self.ensure_authenticated() async def _run_oauth_flow(self) -> str: """Run interactive OAuth flow with browser.""" # Generate PKCE values code_verifier = secrets.token_urlsafe(32) code_challenge = base64.urlsafe_b64encode( hashlib.sha256(code_verifier.encode()).digest() ).decode().rstrip("=") # Build auth URL auth_url = self.app.get_authorization_request_url( scopes=SCOPES, redirect_uri=self.config.redirect_uri, code_challenge=code_challenge, code_challenge_method="S256", ) # Start local server to receive callback auth_code = await self._get_auth_code_from_browser(auth_url) # Exchange code for tokens result = self.app.acquire_token_by_authorization_code( code=auth_code, scopes=SCOPES, redirect_uri=self.config.redirect_uri, code_verifier=code_verifier, ) if "error" in result: raise AuthenticationError(f"Token acquisition failed: {result['error_description']}") # Store tokens self._save_token_to_keyring(result) return result["access_token"] async def _get_auth_code_from_browser(self, auth_url: str) -> str: """Open browser and wait for OAuth callback.""" auth_code: str | None = None class CallbackHandler(BaseHTTPRequestHandler): def do_GET(self): nonlocal auth_code query = parse_qs(urlparse(self.path).query) if "code" in query: auth_code = query["code"][0] self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(b""" <html><body> <h1>Authentication successful!</h1> <p>You can close this window and return to Claude.</p> </body></html> """) else: self.send_response(400) self.end_headers() def log_message(self, format, *args): pass # Suppress logging server = HTTPServer(("localhost", 8765), CallbackHandler) server.timeout = 120 # 2 minute timeout logger.info("Opening browser for authentication...") webbrowser.open(auth_url) # Wait for callback in background thread loop = asyncio.get_event_loop() await loop.run_in_executor(None, server.handle_request) server.server_close() if not auth_code: raise AuthenticationError("Authentication timed out or was cancelled") return auth_code async def _refresh_token(self, refresh_token: str) -> str: """Refresh access token using refresh token.""" result = self.app.acquire_token_by_refresh_token( refresh_token=refresh_token, scopes=SCOPES, ) if "error" in result: raise AuthenticationError(f"Token refresh failed: {result['error_description']}") self._save_token_to_keyring(result) return result["access_token"] def _save_token_to_keyring(self, result: dict) -> None: """Save tokens to OS keychain.""" expires_at = datetime.now() + timedelta(seconds=result["expires_in"]) token_data = { "access_token": result["access_token"], "refresh_token": result.get("refresh_token"), "expires_at": expires_at.isoformat(), } keyring.set_password(KEYRING_SERVICE, self.config.client_id, json.dumps(token_data)) logger.info("Tokens saved to keychain") def _load_token_from_keyring(self) -> dict | None: """Load tokens from OS keychain.""" try: data = keyring.get_password(KEYRING_SERVICE, self.config.client_id) if data: return json.loads(data) except Exception as e: logger.warning(f"Failed to load token from keyring: {e}") return None def clear_tokens(self) -> None: """Clear stored tokens (for re-authentication).""" try: keyring.delete_password(KEYRING_SERVICE, self.config.client_id) logger.info("Tokens cleared from keychain") except keyring.errors.PasswordDeleteError: pass class AuthenticationError(Exception): """Authentication-related errors.""" pass ``` ### 5. Graph Client (src/sharepoint_mcp/graph.py) ```python """ Microsoft Graph API client with automatic token refresh and rate limiting. """ import asyncio import base64 import logging from datetime import datetime from typing import Any import httpx from .auth import AuthManager from .models import ( Site, Library, FileMetadata, FileContent, FileListResult, UploadResult, SharingLink, SearchResult, ListItem ) logger = logging.getLogger(__name__) GRAPH_BASE = "https://graph.microsoft.com/v1.0" MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB class GraphClient: """Async client for Microsoft Graph API.""" def __init__(self, auth: AuthManager): self.auth = auth self._client: httpx.AsyncClient | None = None async def _get_client(self) -> httpx.AsyncClient: """Get or create HTTP client with current auth token.""" token = await self.auth.get_access_token() if self._client is None: self._client = httpx.AsyncClient( base_url=GRAPH_BASE, timeout=30.0, ) self._client.headers["Authorization"] = f"Bearer {token}" return self._client async def close(self) -> None: """Close the HTTP client.""" if self._client: await self._client.aclose() self._client = None async def _request( self, method: str, path: str, **kwargs, ) -> dict[str, Any]: """Make authenticated request with retry logic.""" client = await self._get_client() for attempt in range(3): response = await client.request(method, path, **kwargs) if response.status_code == 429: # Rate limited - wait and retry retry_after = int(response.headers.get("Retry-After", 5)) logger.warning(f"Rate limited, waiting {retry_after}s") await asyncio.sleep(retry_after) continue if response.status_code == 401: # Token expired - refresh and retry await self.auth.ensure_authenticated() client = await self._get_client() continue response.raise_for_status() return response.json() if response.content else {} raise GraphAPIError("Max retries exceeded") # ========== Sites ========== async def get_sites(self, search: str | None = None) -> list[Site]: """List SharePoint sites.""" if search: data = await self._request("GET", f"/sites?search={search}") else: data = await self._request("GET", "/sites?search=*") return [ Site( id=s["id"], name=s["displayName"], url=s["webUrl"], description=s.get("description"), ) for s in data.get("value", []) ] async def get_libraries(self, site_id: str) -> list[Library]: """List document libraries in a site.""" data = await self._request("GET", f"/sites/{site_id}/drives") return [ Library( id=d["id"], name=d["name"], web_url=d["webUrl"], item_count=d.get("quota", {}).get("used", 0), ) for d in data.get("value", []) ] # ========== Files ========== async def list_files( self, site_id: str, library_id: str, folder_path: str = "", cursor: str | None = None, limit: int = 50, ) -> FileListResult: """List files in a library/folder with pagination.""" if cursor: # Use the cursor URL directly data = await self._request("GET", cursor) else: path = f"/sites/{site_id}/drives/{library_id}/root" if folder_path: path += f":/{folder_path}:" path += f"/children?$top={limit}" data = await self._request("GET", path) files = [ FileMetadata( id=item["id"], name=item["name"], size=item.get("size", 0), mime_type=item.get("file", {}).get("mimeType"), web_url=item["webUrl"], created_at=datetime.fromisoformat(item["createdDateTime"].rstrip("Z")), modified_at=datetime.fromisoformat(item["lastModifiedDateTime"].rstrip("Z")), created_by=item.get("createdBy", {}).get("user", {}).get("displayName"), modified_by=item.get("lastModifiedBy", {}).get("user", {}).get("displayName"), ) for item in data.get("value", []) if "file" in item # Only files, not folders ] return FileListResult( files=files, next_cursor=data.get("@odata.nextLink"), total_count=data.get("@odata.count"), ) async def download_file(self, site_id: str, file_id: str) -> FileContent: """Download file content.""" # Get metadata first meta = await self.get_file_metadata(site_id, file_id) if meta.size > MAX_FILE_SIZE: raise GraphAPIError(f"File too large ({meta.size} bytes). Max: {MAX_FILE_SIZE}") # Download content client = await self._get_client() response = await client.get(f"/sites/{site_id}/drive/items/{file_id}/content") response.raise_for_status() content = response.content is_text = meta.mime_type and meta.mime_type.startswith(("text/", "application/json")) return FileContent( name=meta.name, mime_type=meta.mime_type or "application/octet-stream", content=content.decode("utf-8") if is_text else base64.b64encode(content).decode(), is_text=is_text, ) async def get_file_metadata(self, site_id: str, file_id: str) -> FileMetadata: """Get file metadata.""" data = await self._request("GET", f"/sites/{site_id}/drive/items/{file_id}") return FileMetadata( id=data["id"], name=data["name"], size=data.get("size", 0), mime_type=data.get("file", {}).get("mimeType"), web_url=data["webUrl"], created_at=datetime.fromisoformat(data["createdDateTime"].rstrip("Z")), modified_at=datetime.fromisoformat(data["lastModifiedDateTime"].rstrip("Z")), created_by=data.get("createdBy", {}).get("user", {}).get("displayName"), modified_by=data.get("lastModifiedBy", {}).get("user", {}).get("displayName"), ) async def upload_file( self, site_id: str, library_id: str, file_name: str, content: str, folder_path: str = "", is_base64: bool = False, ) -> UploadResult: """Upload file to SharePoint.""" file_content = base64.b64decode(content) if is_base64 else content.encode() path = f"/sites/{site_id}/drives/{library_id}/root" if folder_path: path += f":/{folder_path}/{file_name}:" else: path += f":/{file_name}:" path += "/content" client = await self._get_client() response = await client.put(path, content=file_content) response.raise_for_status() data = response.json() return UploadResult( id=data["id"], name=data["name"], web_url=data["webUrl"], size=data["size"], ) async def create_sharing_link( self, site_id: str, file_id: str, link_type: str = "view", expiration_days: int | None = None, ) -> SharingLink: """Create sharing link for a file.""" body: dict[str, Any] = { "type": link_type, "scope": "organization", } if expiration_days: from datetime import timedelta expiry = datetime.now() + timedelta(days=expiration_days) body["expirationDateTime"] = expiry.isoformat() data = await self._request( "POST", f"/sites/{site_id}/drive/items/{file_id}/createLink", json=body, ) return SharingLink( link_url=data["link"]["webUrl"], link_type=link_type, expires_at=datetime.fromisoformat(data["expirationDateTime"].rstrip("Z")) if "expirationDateTime" in data else None, ) async def search_files( self, query: str, site_id: str | None = None, ) -> list[SearchResult]: """Search for files across SharePoint.""" body = { "requests": [{ "entityTypes": ["driveItem"], "query": {"queryString": query}, }] } data = await self._request("POST", "/search/query", json=body) results = [] for hit in data.get("value", [{}])[0].get("hitsContainers", [{}])[0].get("hits", []): resource = hit.get("resource", {}) results.append(SearchResult( id=resource.get("id", ""), name=resource.get("name", ""), site_name=resource.get("parentReference", {}).get("siteId", "").split(",")[1] if "parentReference" in resource else "", web_url=resource.get("webUrl", ""), snippet=hit.get("summary"), )) return results # ========== Lists ========== async def get_list_items( self, site_id: str, list_id: str, limit: int = 100, ) -> list[ListItem]: """Get items from a SharePoint list.""" data = await self._request( "GET", f"/sites/{site_id}/lists/{list_id}/items?expand=fields&$top={limit}", ) return [ ListItem( id=item["id"], fields=item.get("fields", {}), created_at=datetime.fromisoformat(item["createdDateTime"].rstrip("Z")), modified_at=datetime.fromisoformat(item["lastModifiedDateTime"].rstrip("Z")), ) for item in data.get("value", []) ] async def create_list_item( self, site_id: str, list_id: str, fields: dict, ) -> ListItem: """Create item in a SharePoint list.""" data = await self._request( "POST", f"/sites/{site_id}/lists/{list_id}/items", json={"fields": fields}, ) return ListItem( id=data["id"], fields=data.get("fields", {}), created_at=datetime.fromisoformat(data["createdDateTime"].rstrip("Z")), modified_at=datetime.fromisoformat(data["lastModifiedDateTime"].rstrip("Z")), ) # ========== User ========== async def get_current_user(self) -> dict[str, Any]: """Get current authenticated user info.""" return await self._request("GET", "/me") class GraphAPIError(Exception): """Graph API errors.""" pass ``` ### 6. Configuration (src/sharepoint_mcp/config.py) ```python """Configuration management with environment variable and file support.""" import json import os from dataclasses import dataclass from pathlib import Path import platformdirs @dataclass class Config: """SharePoint MCP configuration.""" client_id: str tenant_id: str redirect_uri: str = "http://localhost:8765/callback" log_level: str = "INFO" def load_config() -> Config: """Load configuration from environment or config file.""" # Try environment variables first client_id = os.environ.get("AZURE_CLIENT_ID") or os.environ.get("SHAREPOINT_CLIENT_ID") tenant_id = os.environ.get("AZURE_TENANT_ID") or os.environ.get("SHAREPOINT_TENANT_ID") if client_id and tenant_id: return Config( client_id=client_id, tenant_id=tenant_id, log_level=os.environ.get("LOG_LEVEL", "INFO"), ) # Try config file config_dir = Path(platformdirs.user_config_dir("sharepoint-mcp")) config_file = config_dir / "config.json" if config_file.exists(): with open(config_file) as f: data = json.load(f) return Config( client_id=data["client_id"], tenant_id=data["tenant_id"], redirect_uri=data.get("redirect_uri", "http://localhost:8765/callback"), log_level=data.get("log_level", "INFO"), ) raise ConfigurationError( "Missing configuration. Set AZURE_CLIENT_ID and AZURE_TENANT_ID environment variables, " f"or create {config_file}" ) class ConfigurationError(Exception): """Configuration-related errors.""" pass ``` ### 7. Entry Points **src/sharepoint_mcp/__init__.py:** ```python from .server import mcp __version__ = "0.1.0" def main(): """Entry point for the MCP server.""" mcp.run() __all__ = ["main", "mcp", "__version__"] ``` **src/sharepoint_mcp/__main__.py:** ```python from . import main if __name__ == "__main__": main() ``` --- ## README.md Structure Create a comprehensive README with these sections: ### 1. Overview & Features - What the server does - List of available tools - Supported platforms ### 2. Quick Start (3 steps) 1. Install uv 2. Register Azure AD app (link to detailed guide) 3. Add to Claude Desktop config ### 3. Installation Options **Option A: uvx (recommended for users)** ```json { "mcpServers": { "sharepoint": { "command": "uvx", "args": ["sharepoint-mcp"], "env": { "AZURE_CLIENT_ID": "your-client-id", "AZURE_TENANT_ID": "your-tenant-id" } } } } ``` **Option B: uv --directory (for development)** ```json { "mcpServers": { "sharepoint": { "command": "uv", "args": ["--directory", "/path/to/sharepoint-mcp", "run", "sharepoint-mcp"], "env": { "AZURE_CLIENT_ID": "your-client-id", "AZURE_TENANT_ID": "your-tenant-id" } } } } ``` **Option C: Claude Code** ```bash # Add via Claude Code CLI claude mcp add sharepoint -- uvx sharepoint-mcp # Or manually in ~/.claude/claude_code_config.json ``` ### 4. Azure AD App Registration Step-by-step guide with: - Portal navigation - Required permissions (Sites.Read.All, Files.ReadWrite.All, User.Read, offline_access) - Redirect URI setup (http://localhost:8765/callback) - Enable public client flows ### 5. Available Tools Document each tool with parameters and example prompts: - `list_sites` - "Show me my SharePoint sites" - `list_files` - "List files in the Marketing site's Documents library" - etc. ### 6. First Run / Authentication - Browser opens automatically - Grant consent - Token stored securely in OS keychain - Automatic refresh ### 7. Troubleshooting - Authentication issues - Permission errors - Re-authentication: `uvx sharepoint-mcp --reset-auth` - Log locations ### 8. Development Setup ```bash git clone https://github.com/yourusername/sharepoint-mcp.git cd sharepoint-mcp uv sync --dev uv run pytest ``` --- ## Testing Requirements ### Unit Tests (tests/test_server.py) - Mock Graph API responses with `respx` - Test each tool function - Test error handling - Coverage target: >80% ### Integration Tests - Require real Azure AD credentials - Skip if credentials not available - Use pytest markers: `@pytest.mark.integration` --- ## Success Criteria ### For End Users - [ ] Install with uvx + JSON config (no Python knowledge needed) - [ ] OAuth flow works seamlessly with browser popup - [ ] All 10 tools work with real SharePoint data - [ ] Clear error messages for auth/permission issues - [ ] Works on macOS, Windows, Linux ### For Developers - [ ] Clean FastMCP implementation following SDK patterns - [ ] Type hints throughout with strict mypy - [ ] >80% test coverage - [ ] Comprehensive docstrings ### Technical - [ ] Pydantic models for all structured output - [ ] Context used for logging/progress - [ ] Lifespan management for auth setup - [ ] Token refresh works automatically - [ ] Rate limiting handled gracefully --- ## Deliverables 1. **Complete codebase** following structure above 2. **pyproject.toml** ready for PyPI publishing 3. **README.md** with all installation options (uvx, uv, Claude Code) 4. **Azure setup guide** in docs/ 5. **Test suite** with mocked unit tests 6. **Working OAuth flow** with keyring storage --- ## Key Differences from Low-Level Implementation This prompt uses **FastMCP** (high-level API) instead of low-level Server class: | Aspect | FastMCP (This Prompt) | Low-Level Server | |--------|----------------------|------------------| | Tool definition | `@mcp.tool()` decorator | `@server.list_tools()` + `@server.call_tool()` | | Schema generation | Automatic from type hints | Manual JSON Schema | | Structured output | Return Pydantic models directly | Manual serialization | | Context access | `ctx: Context` parameter | `server.request_context` | | Lifespan | `lifespan=` parameter | Manual async context | | Running | `mcp.run()` | Manual stdio setup | Follow the patterns in https://github.com/modelcontextprotocol/python-sdk exactly.