{ "cells": [ { "cell_type": "raw", "metadata": { "vscode": { "languageId": "raw" } }, "source": [ "# šŸ“ File Operations with FastMCP\n", "\n", "Welcome to intermediate MCP development! In this notebook, you'll learn to build a production-ready file processing server using FastMCP's modern patterns and type-safe operations.\n", "\n", "## šŸŽÆ Learning Objectives\n", "\n", "By the end of this notebook, you will:\n", "- Build a type-safe file processing server with FastMCP\n", "- Implement file resources and prompts\n", "- Create secure file operation tools\n", "- Use Pydantic models for validation\n", "- Apply MCP best practices for file handling\n", "\n", "## šŸ› ļø What You'll Build\n", "\n", "```python\n", "from mcp.server.fastmcp import FastMCP\n", "from pydantic import BaseModel\n", "\n", "# Initialize FastMCP server\n", "mcp = FastMCP(\"file_tools\")\n", "\n", "class FileInfo(BaseModel):\n", " \"\"\"File metadata model\"\"\"\n", " path: str\n", " size: int\n", " mime_type: str\n", " hash: str\n", "\n", "@mcp.resource(\"file://{path}\")\n", "async def get_file_info(path: str) -> FileInfo:\n", " \"\"\"Get file metadata\"\"\"\n", " ...\n", "\n", "@mcp.tool()\n", "async def read_file(path: str) -> Dict:\n", " \"\"\"Read file contents safely\"\"\"\n", " ...\n", "\n", "@mcp.prompt()\n", "def error_prompt() -> str:\n", " \"\"\"Handle file errors\"\"\"\n", " return \"\"\"File operation failed...\"\"\"\n", "```\n", "\n", "## šŸ” Modern MCP Patterns\n", "\n", "FastMCP provides robust file handling through:\n", "- **Type-safe operations** with Pydantic models\n", "- **Resource providers** for file metadata\n", "- **System prompts** for error handling\n", "- **Async/await** for performance\n", "- **Structured responses** for consistency\n", "\n", "## šŸ“š Table of Contents\n", "\n", "1. [FastMCP Setup](#fastmcp-setup)\n", "2. [File Resources](#file-resources)\n", "3. [File Operation Tools](#file-tools)\n", "4. [System Prompts](#system-prompts)\n", "5. [Error Handling](#error-handling)\n", "6. [Production Patterns](#production-patterns)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# First, let's set up our FastMCP server with necessary imports\n", "from mcp.server.fastmcp import FastMCP\n", "import mcp.types as types\n", "from pydantic import BaseModel\n", "from typing import Dict, List, Optional\n", "from pathlib import Path\n", "import hashlib\n", "import mimetypes\n", "from datetime import datetime\n", "\n", "# Initialize our FastMCP server\n", "mcp = FastMCP(\"file_processor\")\n", "\n", "# Define our data models\n", "class FileInfo(BaseModel):\n", " \"\"\"File metadata model\"\"\"\n", " path: str\n", " size: int\n", " modified: float\n", " mime_type: str\n", " hash: Optional[str]\n", "\n", "class FileContent(BaseModel):\n", " \"\"\"File content model\"\"\"\n", " content: str\n", " encoding: str\n", " lines: int\n", " info: FileInfo\n", "\n", "# System prompts\n", "@mcp.prompt()\n", "def system_prompt() -> str:\n", " \"\"\"Define the file processor's role\"\"\"\n", " return \"\"\"\n", " You are a file processing assistant that helps users work with files safely.\n", " Always validate paths and file operations before execution.\n", " Never perform dangerous operations without confirmation.\n", " \"\"\"\n", "\n", "@mcp.prompt()\n", "def error_prompt() -> str:\n", " \"\"\"Handle file operation errors\"\"\"\n", " return \"\"\"\n", " I encountered an error while processing the file.\n", " This could be due to:\n", " - Invalid file path\n", " - Permission issues\n", " - File not found\n", " - Unsupported file type\n", " Please check the file and try again.\n", " \"\"\"\n", "\n", "print(\"āœ… FastMCP server initialized with models and prompts!\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Now let's implement our file resource provider\n", "@mcp.resource(\"file://{path}\")\n", "async def get_file_info(path: str) -> Dict:\n", " \"\"\"Get metadata about a file\"\"\"\n", " try:\n", " full_path = Path(path).resolve()\n", " \n", " if not full_path.exists():\n", " return {\"error\": \"File not found\"}\n", " \n", " stat = full_path.stat()\n", " mime_type, _ = mimetypes.guess_type(str(full_path))\n", " \n", " # Calculate file hash\n", " sha256_hash = hashlib.sha256()\n", " with open(full_path, \"rb\") as f:\n", " for byte_block in iter(lambda: f.read(4096), b\"\"):\n", " sha256_hash.update(byte_block)\n", " \n", " return FileInfo(\n", " path=str(full_path),\n", " size=stat.st_size,\n", " modified=stat.st_mtime,\n", " mime_type=mime_type or \"application/octet-stream\",\n", " hash=sha256_hash.hexdigest()\n", " ).dict()\n", " \n", " except Exception as e:\n", " return {\"error\": str(e)}\n", "\n", "# Create a test file\n", "with open(\"test.txt\", \"w\") as f:\n", " f.write(\"Hello FastMCP!\")\n", "\n", "# Test our resource\n", "import asyncio\n", "result = asyncio.run(get_file_info(\"test.txt\"))\n", "print(\"šŸ“„ File Info:\", result)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Let's implement our file operation tools\n", "@mcp.tool()\n", "async def read_file(file_path: str, encoding: str = \"utf-8\") -> Dict:\n", " \"\"\"\n", " Read a file safely\n", " \n", " Args:\n", " file_path: Path to the file\n", " encoding: File encoding (default: utf-8)\n", " \n", " Returns:\n", " Dictionary with file contents and metadata\n", " \"\"\"\n", " try:\n", " # Get file info from resource\n", " file_info = await get_file_info(file_path)\n", " \n", " if \"error\" in file_info:\n", " raise Exception(file_info[\"error\"])\n", " \n", " # Validate file type\n", " if not file_info[\"mime_type\"].startswith((\"text/\", \"application/\")):\n", " raise Exception(\"Unsupported file type\")\n", " \n", " # Read file\n", " with open(file_path, \"r\", encoding=encoding) as f:\n", " content = f.read()\n", " \n", " return {\n", " \"success\": True,\n", " \"data\": FileContent(\n", " content=content,\n", " encoding=encoding,\n", " lines=len(content.splitlines()),\n", " info=FileInfo(**file_info)\n", " ).dict()\n", " }\n", " \n", " except Exception as e:\n", " return {\n", " \"success\": False,\n", " \"error\": str(e),\n", " \"prompt\": \"error_prompt\"\n", " }\n", "\n", "@mcp.tool()\n", "async def write_file(file_path: str, content: str, encoding: str = \"utf-8\") -> Dict:\n", " \"\"\"\n", " Write content to a file safely\n", " \n", " Args:\n", " file_path: Path to write to\n", " content: Content to write\n", " encoding: File encoding (default: utf-8)\n", " \n", " Returns:\n", " Dictionary with operation results\n", " \"\"\"\n", " try:\n", " # Validate path\n", " full_path = Path(file_path).resolve()\n", " \n", " # Create parent directories\n", " full_path.parent.mkdir(parents=True, exist_ok=True)\n", " \n", " # Write file\n", " with open(full_path, \"w\", encoding=encoding) as f:\n", " f.write(content)\n", " \n", " # Get updated file info\n", " file_info = await get_file_info(str(full_path))\n", " \n", " return {\n", " \"success\": True,\n", " \"data\": {\n", " \"path\": str(full_path),\n", " \"size\": file_info[\"size\"],\n", " \"lines\": len(content.splitlines()),\n", " \"encoding\": encoding\n", " }\n", " }\n", " \n", " except Exception as e:\n", " return {\n", " \"success\": False,\n", " \"error\": str(e),\n", " \"prompt\": \"error_prompt\"\n", " }\n", "\n", "# Test our tools\n", "result = asyncio.run(write_file(\"example.txt\", \"Testing FastMCP tools!\\nThis is line 2.\"))\n", "print(\"āœļø Write result:\", result)\n", "\n", "result = asyncio.run(read_file(\"example.txt\"))\n", "print(\"\\nšŸ“– Read result:\", result)\n" ] }, { "cell_type": "raw", "metadata": { "vscode": { "languageId": "raw" } }, "source": [ "# šŸš€ Running the FastMCP Server\n", "\n", "Now that we have our file processing server built with FastMCP patterns, let's run it:\n", "\n", "```python\n", "if __name__ == \"__main__\":\n", " # Run the MCP server\n", " print(\"šŸ“ Starting File Processor MCP Server...\")\n", " mcp.run(transport=\"streamable-http\")\n", "```\n", "\n", "## šŸŽÆ Key Takeaways\n", "\n", "1. **Type Safety**\n", " - Use Pydantic models for data validation\n", " - Define clear input/output types\n", " - Catch errors early with type checking\n", "\n", "2. **Resource Providers**\n", " - Abstract file metadata access\n", " - Cache common operations\n", " - Provide consistent interfaces\n", "\n", "3. **System Prompts**\n", " - Guide AI behavior\n", " - Handle errors gracefully\n", " - Provide clear user feedback\n", "\n", "4. **Best Practices**\n", " - Always validate paths\n", " - Use async/await for I/O\n", " - Structure responses consistently\n", " - Handle errors with prompts\n", "\n", "## šŸ”œ Next Steps\n", "\n", "Continue to [API Integration](07_api_integration.ipynb) to learn how to connect your MCP server to external services!\n" ] } ], "metadata": { "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 2 }