{
  "cells": [
    {
      "cell_type": "raw",
      "metadata": {
        "vscode": {
          "languageId": "raw"
        }
      },
      "source": [
        "# 🎯 Custom Resources in MCP\n",
        "\n",
        "Learn how to create and manage custom resource providers in MCP. This notebook covers advanced resource management, dynamic resource creation, and resource lifecycle management.\n",
        "\n",
        "## 🎯 Learning Objectives\n",
        "\n",
        "By the end of this notebook, you will:\n",
        "- Create custom resource providers\n",
        "- Manage resource lifecycles\n",
        "- Handle resource dependencies\n",
        "- Implement resource cleanup\n",
        "- Build dynamic resource allocation\n",
        "\n",
        "## 📋 Prerequisites\n",
        "\n",
        "- Completed notebooks 01-10\n",
        "- Strong understanding of Python OOP\n",
        "- Familiarity with context managers\n",
        "- Basic knowledge of resource management\n",
        "\n",
        "## 🔑 Key Concepts\n",
        "\n",
        "1. **Resource Types**\n",
        "   - Static resources\n",
        "   - Dynamic resources\n",
        "   - Pooled resources\n",
        "   - Shared resources\n",
        "\n",
        "2. **Resource Management**\n",
        "   - Lifecycle hooks\n",
        "   - Dependency resolution\n",
        "   - Resource cleanup\n",
        "   - Error recovery\n",
        "\n",
        "3. **Resource Patterns**\n",
        "   - Factory pattern\n",
        "   - Pool pattern\n",
        "   - Proxy pattern\n",
        "   - Observer pattern\n",
        "\n",
        "## 📚 Table of Contents\n",
        "\n",
        "1. [Understanding Resources](#understanding)\n",
        "2. [Resource Providers](#providers)\n",
        "3. [Resource Lifecycle](#lifecycle)\n",
        "4. [Advanced Patterns](#patterns)\n",
        "5. [Best Practices](#practices)\n",
        "\n",
        "---\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from typing import Dict, Any, Optional, List, Type\n",
        "from abc import ABC, abstractmethod\n",
        "import asyncio\n",
        "from dataclasses import dataclass, field\n",
        "from datetime import datetime\n",
        "import modelcontextprotocol as mcp\n",
        "from pydantic import BaseModel, Field\n",
        "\n",
        "# Base resource class\n",
        "class Resource(ABC):\n",
        "    \"\"\"Abstract base class for all resources.\"\"\"\n",
        "    \n",
        "    def __init__(self, resource_id: str, config: Dict[str, Any]):\n",
        "        self.resource_id = resource_id\n",
        "        self.config = config\n",
        "        self.created_at = datetime.now()\n",
        "        self.last_accessed = datetime.now()\n",
        "        self._is_initialized = False\n",
        "        \n",
        "    @abstractmethod\n",
        "    async def initialize(self) -> None:\n",
        "        \"\"\"Initialize the resource.\"\"\"\n",
        "        pass\n",
        "    \n",
        "    @abstractmethod\n",
        "    async def cleanup(self) -> None:\n",
        "        \"\"\"Clean up the resource.\"\"\"\n",
        "        pass\n",
        "    \n",
        "    def update_access_time(self):\n",
        "        \"\"\"Update the last access time.\"\"\"\n",
        "        self.last_accessed = datetime.now()\n",
        "        \n",
        "    @property\n",
        "    def is_initialized(self) -> bool:\n",
        "        return self._is_initialized\n",
        "    \n",
        "    def __repr__(self) -> str:\n",
        "        return f\"{self.__class__.__name__}(id={self.resource_id})\"\n",
        "\n",
        "# Resource provider\n",
        "class ResourceProvider(ABC):\n",
        "    \"\"\"Abstract base class for resource providers.\"\"\"\n",
        "    \n",
        "    def __init__(self):\n",
        "        self.resources: Dict[str, Resource] = {}\n",
        "        \n",
        "    @abstractmethod\n",
        "    async def create_resource(self, resource_id: str, config: Dict[str, Any]) -> Resource:\n",
        "        \"\"\"Create a new resource.\"\"\"\n",
        "        pass\n",
        "    \n",
        "    @abstractmethod\n",
        "    async def get_resource(self, resource_id: str) -> Resource:\n",
        "        \"\"\"Get an existing resource.\"\"\"\n",
        "        pass\n",
        "    \n",
        "    async def delete_resource(self, resource_id: str) -> None:\n",
        "        \"\"\"Delete a resource.\"\"\"\n",
        "        if resource_id in self.resources:\n",
        "            resource = self.resources[resource_id]\n",
        "            await resource.cleanup()\n",
        "            del self.resources[resource_id]\n",
        "            \n",
        "    async def cleanup_all(self) -> None:\n",
        "        \"\"\"Clean up all resources.\"\"\"\n",
        "        for resource_id in list(self.resources.keys()):\n",
        "            await self.delete_resource(resource_id)\n",
        "\n",
        "# Resource pool\n",
        "class ResourcePool:\n",
        "    \"\"\"Manages a pool of resources.\"\"\"\n",
        "    \n",
        "    def __init__(self, provider: ResourceProvider, max_size: int = 10):\n",
        "        self.provider = provider\n",
        "        self.max_size = max_size\n",
        "        self.available: List[Resource] = []\n",
        "        self.in_use: Dict[str, Resource] = {}\n",
        "        \n",
        "    async def acquire(self, config: Dict[str, Any]) -> Resource:\n",
        "        \"\"\"Acquire a resource from the pool.\"\"\"\n",
        "        if self.available:\n",
        "            resource = self.available.pop()\n",
        "        elif len(self.in_use) < self.max_size:\n",
        "            resource = await self.provider.create_resource(\n",
        "                f\"pool-resource-{len(self.in_use)}\", \n",
        "                config\n",
        "            )\n",
        "        else:\n",
        "            raise RuntimeError(\"Resource pool exhausted\")\n",
        "            \n",
        "        self.in_use[resource.resource_id] = resource\n",
        "        return resource\n",
        "        \n",
        "    async def release(self, resource: Resource) -> None:\n",
        "        \"\"\"Release a resource back to the pool.\"\"\"\n",
        "        if resource.resource_id in self.in_use:\n",
        "            del self.in_use[resource.resource_id]\n",
        "            self.available.append(resource)\n",
        "            \n",
        "    async def cleanup(self) -> None:\n",
        "        \"\"\"Clean up all resources in the pool.\"\"\"\n",
        "        await self.provider.cleanup_all()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "import sqlite3\n",
        "from contextlib import asynccontextmanager\n",
        "\n",
        "class DatabaseResource(Resource):\n",
        "    \"\"\"A database connection resource.\"\"\"\n",
        "    \n",
        "    def __init__(self, resource_id: str, config: Dict[str, Any]):\n",
        "        super().__init__(resource_id, config)\n",
        "        self.connection: Optional[sqlite3.Connection] = None\n",
        "        \n",
        "    async def initialize(self) -> None:\n",
        "        \"\"\"Initialize the database connection.\"\"\"\n",
        "        if not self._is_initialized:\n",
        "            self.connection = sqlite3.connect(self.config[\"database\"])\n",
        "            self._is_initialized = True\n",
        "            \n",
        "    async def cleanup(self) -> None:\n",
        "        \"\"\"Close the database connection.\"\"\"\n",
        "        if self.connection:\n",
        "            self.connection.close()\n",
        "            self.connection = None\n",
        "            self._is_initialized = False\n",
        "            \n",
        "    def execute_query(self, query: str, parameters: Optional[tuple] = None) -> List[tuple]:\n",
        "        \"\"\"Execute a query on the database.\"\"\"\n",
        "        if not self.connection:\n",
        "            raise RuntimeError(\"Database not initialized\")\n",
        "            \n",
        "        self.update_access_time()\n",
        "        cursor = self.connection.cursor()\n",
        "        try:\n",
        "            if parameters:\n",
        "                cursor.execute(query, parameters)\n",
        "            else:\n",
        "                cursor.execute(query)\n",
        "            return cursor.fetchall()\n",
        "        finally:\n",
        "            cursor.close()\n",
        "\n",
        "class DatabaseProvider(ResourceProvider):\n",
        "    \"\"\"Provider for database connections.\"\"\"\n",
        "    \n",
        "    async def create_resource(self, resource_id: str, config: Dict[str, Any]) -> Resource:\n",
        "        \"\"\"Create a new database connection.\"\"\"\n",
        "        resource = DatabaseResource(resource_id, config)\n",
        "        await resource.initialize()\n",
        "        self.resources[resource_id] = resource\n",
        "        return resource\n",
        "        \n",
        "    async def get_resource(self, resource_id: str) -> Resource:\n",
        "        \"\"\"Get an existing database connection.\"\"\"\n",
        "        if resource_id not in self.resources:\n",
        "            raise KeyError(f\"Resource {resource_id} not found\")\n",
        "        return self.resources[resource_id]\n",
        "\n",
        "# MCP models\n",
        "class DatabaseConfig(BaseModel):\n",
        "    database: str = Field(..., description=\"Database file path\")\n",
        "    pool_size: int = Field(default=5, description=\"Maximum number of connections\")\n",
        "    \n",
        "class QueryRequest(BaseModel):\n",
        "    query: str = Field(..., description=\"SQL query to execute\")\n",
        "    parameters: Optional[tuple] = Field(default=None, description=\"Query parameters\")\n",
        "    \n",
        "class QueryResult(BaseModel):\n",
        "    results: List[tuple] = Field(..., description=\"Query results\")\n",
        "    resource_id: str = Field(..., description=\"ID of the resource used\")\n",
        "\n",
        "# MCP tool\n",
        "class DatabasePoolTool:\n",
        "    \"\"\"MCP tool for managing a pool of database connections.\"\"\"\n",
        "    \n",
        "    def __init__(self, config: DatabaseConfig):\n",
        "        self.provider = DatabaseProvider()\n",
        "        self.pool = ResourcePool(self.provider, config.pool_size)\n",
        "        self.config = {\"database\": config.database}\n",
        "        \n",
        "    async def execute_query(self, request: QueryRequest) -> QueryResult:\n",
        "        \"\"\"Execute a query using a connection from the pool.\"\"\"\n",
        "        resource = await self.pool.acquire(self.config)\n",
        "        try:\n",
        "            results = resource.execute_query(request.query, request.parameters)\n",
        "            return QueryResult(\n",
        "                results=results,\n",
        "                resource_id=resource.resource_id\n",
        "            )\n",
        "        finally:\n",
        "            await self.pool.release(resource)\n",
        "            \n",
        "    async def cleanup(self):\n",
        "        \"\"\"Clean up all resources.\"\"\"\n",
        "        await self.pool.cleanup()\n",
        "\n",
        "# Create MCP server\n",
        "config = DatabaseConfig(database=\":memory:\", pool_size=3)\n",
        "db_tool = DatabasePoolTool(config)\n",
        "server = mcp.Server()\n",
        "server.add_tool(\"database\", db_tool.execute_query, QueryRequest, QueryResult)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "async def test_database_pool():\n",
        "    # Create test table\n",
        "    create_table = QueryRequest(\n",
        "        query=\"\"\"\n",
        "        CREATE TABLE IF NOT EXISTS users (\n",
        "            id INTEGER PRIMARY KEY,\n",
        "            name TEXT NOT NULL,\n",
        "            email TEXT UNIQUE NOT NULL\n",
        "        )\n",
        "        \"\"\"\n",
        "    )\n",
        "    result = await db_tool.execute_query(create_table)\n",
        "    print(\"Created table using resource:\", result.resource_id)\n",
        "    \n",
        "    # Insert test data\n",
        "    insert_data = QueryRequest(\n",
        "        query=\"INSERT INTO users (name, email) VALUES (?, ?)\",\n",
        "        parameters=(\"John Doe\", \"john@example.com\")\n",
        "    )\n",
        "    result = await db_tool.execute_query(insert_data)\n",
        "    print(\"\\nInserted data using resource:\", result.resource_id)\n",
        "    \n",
        "    # Query data\n",
        "    select_data = QueryRequest(\n",
        "        query=\"SELECT * FROM users\"\n",
        "    )\n",
        "    result = await db_tool.execute_query(select_data)\n",
        "    print(\"\\nQueried data using resource:\", result.resource_id)\n",
        "    print(\"Results:\", result.results)\n",
        "    \n",
        "    # Test pool exhaustion\n",
        "    print(\"\\nTesting pool exhaustion...\")\n",
        "    resources = []\n",
        "    try:\n",
        "        for i in range(5):  # Try to get more than pool_size connections\n",
        "            result = await db_tool.execute_query(select_data)\n",
        "            print(f\"Got resource: {result.resource_id}\")\n",
        "    except RuntimeError as e:\n",
        "        print(f\"Pool exhausted as expected: {e}\")\n",
        "        \n",
        "    # Clean up\n",
        "    await db_tool.cleanup()\n",
        "    print(\"\\nCleaned up all resources\")\n",
        "\n",
        "# Run the test\n",
        "await test_database_pool()\n"
      ]
    },
    {
      "cell_type": "raw",
      "metadata": {
        "vscode": {
          "languageId": "raw"
        }
      },
      "source": [
        "# Advanced Resource Management Patterns\n",
        "\n",
        "When implementing custom resources in MCP, consider these advanced patterns:\n",
        "\n",
        "## 1. Resource Factory Pattern\n",
        "\n",
        "Use factories to create resources with complex initialization:\n",
        "```python\n",
        "class ResourceFactory:\n",
        "    @classmethod\n",
        "    def create_resource(cls, resource_type: str, config: Dict[str, Any]) -> Resource:\n",
        "        if resource_type == \"database\":\n",
        "            return DatabaseResource(str(uuid.uuid4()), config)\n",
        "        elif resource_type == \"api\":\n",
        "            return APIResource(str(uuid.uuid4()), config)\n",
        "        # Add more resource types\n",
        "```\n",
        "\n",
        "## 2. Resource Proxy Pattern\n",
        "\n",
        "Add a layer of indirection for resource access:\n",
        "```python\n",
        "class ResourceProxy:\n",
        "    def __init__(self, resource: Resource):\n",
        "        self._resource = resource\n",
        "        \n",
        "    def __getattr__(self, name):\n",
        "        # Add logging, metrics, or validation\n",
        "        return getattr(self._resource, name)\n",
        "```\n",
        "\n",
        "## 3. Resource Observer Pattern\n",
        "\n",
        "Monitor resource lifecycle events:\n",
        "```python\n",
        "class ResourceObserver:\n",
        "    def on_create(self, resource: Resource): pass\n",
        "    def on_acquire(self, resource: Resource): pass\n",
        "    def on_release(self, resource: Resource): pass\n",
        "    def on_delete(self, resource: Resource): pass\n",
        "```\n",
        "\n",
        "## Best Practices\n",
        "\n",
        "1. **Resource Lifecycle**\n",
        "   - Initialize resources lazily\n",
        "   - Clean up resources promptly\n",
        "   - Handle initialization failures\n",
        "   - Implement proper shutdown\n",
        "\n",
        "2. **Resource Pool Management**\n",
        "   - Set appropriate pool sizes\n",
        "   - Implement connection timeouts\n",
        "   - Handle pool exhaustion\n",
        "   - Monitor pool health\n",
        "\n",
        "3. **Error Handling**\n",
        "   - Catch resource-specific errors\n",
        "   - Implement retry logic\n",
        "   - Handle cleanup failures\n",
        "   - Log resource events\n",
        "\n",
        "4. **Performance**\n",
        "   - Cache resource handles\n",
        "   - Use connection pooling\n",
        "   - Implement resource limits\n",
        "   - Monitor resource usage\n",
        "\n",
        "## Exercise\n",
        "\n",
        "Improve our database pool implementation with:\n",
        "\n",
        "1. Add resource event logging\n",
        "2. Implement connection timeouts\n",
        "3. Add retry logic for failed operations\n",
        "4. Create a resource monitoring system\n"
      ]
    },
    {
      "cell_type": "raw",
      "metadata": {
        "vscode": {
          "languageId": "raw"
        }
      },
      "source": [
        "# 📊 Custom Resources & Advanced MCP\n",
        "\n",
        "Master advanced MCP concepts! Learn to create custom resource types, implement complex data providers, and build sophisticated AI-accessible data sources.\n",
        "\n",
        "## 🎯 Learning Objectives\n",
        "\n",
        "By the end of this notebook, you will:\n",
        "- Design and implement custom resource types\n",
        "- Build dynamic resource providers\n",
        "- Create resource discovery and enumeration systems\n",
        "- Implement resource caching and synchronization\n",
        "- Master advanced MCP resource patterns\n",
        "\n",
        "## 🔧 Resource Types\n",
        "\n",
        "- **📁 File Resources**: Dynamic file system access\n",
        "- **📊 Data Resources**: Structured data from databases and APIs\n",
        "- **🔄 Stream Resources**: Real-time data streams\n",
        "- **🎯 Computed Resources**: On-demand calculated data\n",
        "- **🌐 Remote Resources**: Federated resource access\n",
        "\n",
        "## 📚 Table of Contents\n",
        "\n",
        "1. [Resource Architecture](#architecture)\n",
        "2. [Custom Resource Implementation](#implementation)  \n",
        "3. [Resource Discovery & Metadata](#discovery)\n",
        "4. [Performance & Caching](#performance)\n",
        "5. [Advanced Resource Patterns](#patterns)\n",
        "\n",
        "---\n"
      ]
    }
  ],
  "metadata": {
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}