{
  "cells": [
    {
      "cell_type": "raw",
      "metadata": {
        "vscode": {
          "languageId": "raw"
        }
      },
      "source": [
        "# 🔧 Advanced Tool Composition in MCP\n",
        "\n",
        "Learn how to compose complex tools from simpler components using advanced patterns and techniques. This notebook covers tool composition, dependency injection, tool coordination, and advanced integration patterns.\n",
        "\n",
        "## 🎯 Learning Objectives\n",
        "\n",
        "By the end of this notebook, you will:\n",
        "- Design composable tools\n",
        "- Implement tool coordination\n",
        "- Handle tool dependencies\n",
        "- Create tool pipelines\n",
        "- Build tool hierarchies\n",
        "\n",
        "## 📋 Prerequisites\n",
        "\n",
        "- Completed notebooks 01-14\n",
        "- Understanding of design patterns\n",
        "- Knowledge of dependency injection\n",
        "- Familiarity with async programming\n",
        "\n",
        "## 🔑 Key Concepts\n",
        "\n",
        "1. **Tool Composition**\n",
        "   - Component design\n",
        "   - Interface contracts\n",
        "   - Dependency management\n",
        "   - State sharing\n",
        "\n",
        "2. **Integration Patterns**\n",
        "   - Pipeline pattern\n",
        "   - Facade pattern\n",
        "   - Adapter pattern\n",
        "   - Decorator pattern\n",
        "\n",
        "3. **Tool Coordination**\n",
        "   - Event handling\n",
        "   - State management\n",
        "   - Resource sharing\n",
        "   - Error propagation\n",
        "\n",
        "## 📚 Table of Contents\n",
        "\n",
        "1. [Tool Components](#components)\n",
        "2. [Integration Patterns](#patterns)\n",
        "3. [Tool Coordination](#coordination)\n",
        "4. [Best Practices](#practices)\n",
        "5. [Exercises](#exercises)\n",
        "\n",
        "---\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from typing import List, Dict, Any, Optional, Protocol, Type\n",
        "from abc import ABC, abstractmethod\n",
        "import asyncio\n",
        "from datetime import datetime\n",
        "from dataclasses import dataclass\n",
        "import modelcontextprotocol as mcp\n",
        "from pydantic import BaseModel, Field\n",
        "\n",
        "# Tool interface\n",
        "class Tool(Protocol):\n",
        "    \"\"\"Protocol for tool interface.\"\"\"\n",
        "    \n",
        "    async def execute(self, input_data: Any) -> Any:\n",
        "        \"\"\"Execute the tool.\"\"\"\n",
        "        pass\n",
        "\n",
        "# Base component\n",
        "class ToolComponent(ABC):\n",
        "    \"\"\"Abstract base class for tool components.\"\"\"\n",
        "    \n",
        "    @abstractmethod\n",
        "    async def process(self, data: Any) -> Any:\n",
        "        \"\"\"Process input data.\"\"\"\n",
        "        pass\n",
        "\n",
        "# Pipeline components\n",
        "class DataValidator(ToolComponent):\n",
        "    \"\"\"Validates input data.\"\"\"\n",
        "    \n",
        "    async def process(self, data: Any) -> Any:\n",
        "        if not isinstance(data, dict):\n",
        "            raise ValueError(\"Input must be a dictionary\")\n",
        "        return data\n",
        "\n",
        "class DataTransformer(ToolComponent):\n",
        "    \"\"\"Transforms input data.\"\"\"\n",
        "    \n",
        "    async def process(self, data: Any) -> Any:\n",
        "        # Add timestamp\n",
        "        data[\"timestamp\"] = datetime.now()\n",
        "        return data\n",
        "\n",
        "class DataEnricher(ToolComponent):\n",
        "    \"\"\"Enriches data with additional information.\"\"\"\n",
        "    \n",
        "    async def process(self, data: Any) -> Any:\n",
        "        # Add metadata\n",
        "        data[\"metadata\"] = {\n",
        "            \"version\": \"1.0\",\n",
        "            \"source\": \"pipeline\"\n",
        "        }\n",
        "        return data\n",
        "\n",
        "# Pipeline pattern\n",
        "class Pipeline:\n",
        "    \"\"\"Tool pipeline implementation.\"\"\"\n",
        "    \n",
        "    def __init__(self):\n",
        "        self.components: List[ToolComponent] = []\n",
        "        \n",
        "    def add_component(self, component: ToolComponent) -> None:\n",
        "        \"\"\"Add component to pipeline.\"\"\"\n",
        "        self.components.append(component)\n",
        "        \n",
        "    async def execute(self, data: Any) -> Any:\n",
        "        \"\"\"Execute pipeline components in sequence.\"\"\"\n",
        "        result = data\n",
        "        for component in self.components:\n",
        "            result = await component.process(result)\n",
        "        return result\n",
        "\n",
        "# Facade pattern\n",
        "class ToolFacade:\n",
        "    \"\"\"Facade for complex tool interactions.\"\"\"\n",
        "    \n",
        "    def __init__(self):\n",
        "        self.pipeline = Pipeline()\n",
        "        self.pipeline.add_component(DataValidator())\n",
        "        self.pipeline.add_component(DataTransformer())\n",
        "        self.pipeline.add_component(DataEnricher())\n",
        "        \n",
        "    async def process_data(self, data: Any) -> Any:\n",
        "        \"\"\"Process data through pipeline.\"\"\"\n",
        "        return await self.pipeline.execute(data)\n",
        "\n",
        "# Adapter pattern\n",
        "class LegacyTool:\n",
        "    \"\"\"Legacy tool with different interface.\"\"\"\n",
        "    \n",
        "    def process_legacy(self, data: str) -> str:\n",
        "        \"\"\"Legacy processing method.\"\"\"\n",
        "        return f\"Processed: {data}\"\n",
        "\n",
        "class LegacyToolAdapter(Tool):\n",
        "    \"\"\"Adapter for legacy tool.\"\"\"\n",
        "    \n",
        "    def __init__(self, legacy_tool: LegacyTool):\n",
        "        self.legacy_tool = legacy_tool\n",
        "        \n",
        "    async def execute(self, input_data: Any) -> Any:\n",
        "        \"\"\"Adapt legacy tool interface.\"\"\"\n",
        "        if isinstance(input_data, dict):\n",
        "            input_data = str(input_data)\n",
        "        return self.legacy_tool.process_legacy(input_data)\n",
        "\n",
        "# Decorator pattern\n",
        "class LoggingDecorator(Tool):\n",
        "    \"\"\"Decorator for adding logging.\"\"\"\n",
        "    \n",
        "    def __init__(self, tool: Tool):\n",
        "        self.tool = tool\n",
        "        \n",
        "    async def execute(self, input_data: Any) -> Any:\n",
        "        \"\"\"Add logging to tool execution.\"\"\"\n",
        "        print(f\"Executing tool with input: {input_data}\")\n",
        "        result = await self.tool.execute(input_data)\n",
        "        print(f\"Tool execution result: {result}\")\n",
        "        return result\n",
        "\n",
        "class TimingDecorator(Tool):\n",
        "    \"\"\"Decorator for timing execution.\"\"\"\n",
        "    \n",
        "    def __init__(self, tool: Tool):\n",
        "        self.tool = tool\n",
        "        \n",
        "    async def execute(self, input_data: Any) -> Any:\n",
        "        \"\"\"Add timing to tool execution.\"\"\"\n",
        "        start = datetime.now()\n",
        "        result = await self.tool.execute(input_data)\n",
        "        duration = (datetime.now() - start).total_seconds()\n",
        "        print(f\"Tool execution took {duration:.3f} seconds\")\n",
        "        return result\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Example tool implementation\n",
        "class DataProcessor(Tool):\n",
        "    \"\"\"Example data processing tool.\"\"\"\n",
        "    \n",
        "    async def execute(self, input_data: Any) -> Any:\n",
        "        \"\"\"Process input data.\"\"\"\n",
        "        await asyncio.sleep(0.5)  # Simulate processing\n",
        "        if isinstance(input_data, dict):\n",
        "            input_data[\"processed\"] = True\n",
        "        return input_data\n",
        "\n",
        "# MCP models\n",
        "class ProcessRequest(BaseModel):\n",
        "    \"\"\"Request for data processing.\"\"\"\n",
        "    data: Dict[str, Any] = Field(..., description=\"Data to process\")\n",
        "    options: Optional[Dict[str, Any]] = Field(default=None, description=\"Processing options\")\n",
        "\n",
        "class ProcessResponse(BaseModel):\n",
        "    \"\"\"Response from data processing.\"\"\"\n",
        "    result: Dict[str, Any] = Field(..., description=\"Processed data\")\n",
        "    metadata: Dict[str, Any] = Field(..., description=\"Processing metadata\")\n",
        "\n",
        "# Create composite tool\n",
        "async def create_composite_tool() -> Tool:\n",
        "    \"\"\"Create a composite tool with all patterns.\"\"\"\n",
        "    \n",
        "    # Create base tool\n",
        "    processor = DataProcessor()\n",
        "    \n",
        "    # Add decorators\n",
        "    logged_processor = LoggingDecorator(processor)\n",
        "    timed_processor = TimingDecorator(logged_processor)\n",
        "    \n",
        "    # Create facade\n",
        "    facade = ToolFacade()\n",
        "    \n",
        "    # Create adapter for legacy integration\n",
        "    legacy_tool = LegacyTool()\n",
        "    adapter = LegacyToolAdapter(legacy_tool)\n",
        "    \n",
        "    # Combine everything into a pipeline\n",
        "    pipeline = Pipeline()\n",
        "    pipeline.add_component(DataValidator())\n",
        "    pipeline.add_component(DataTransformer())\n",
        "    pipeline.add_component(DataEnricher())\n",
        "    \n",
        "    class CompositeTool(Tool):\n",
        "        async def execute(self, input_data: Any) -> Any:\n",
        "            # Process through pipeline\n",
        "            result = await pipeline.execute(input_data)\n",
        "            \n",
        "            # Process through decorated tool\n",
        "            result = await timed_processor.execute(result)\n",
        "            \n",
        "            # Process through facade\n",
        "            result = await facade.process_data(result)\n",
        "            \n",
        "            # Process through adapter\n",
        "            adapter_result = await adapter.execute(result)\n",
        "            result[\"legacy_result\"] = adapter_result\n",
        "            \n",
        "            return result\n",
        "    \n",
        "    return CompositeTool()\n",
        "\n",
        "# Test the composite tool\n",
        "async def test_composite_tool():\n",
        "    print(\"Testing composite tool...\")\n",
        "    \n",
        "    # Create tool\n",
        "    tool = await create_composite_tool()\n",
        "    \n",
        "    # Create test data\n",
        "    input_data = {\n",
        "        \"id\": \"test-1\",\n",
        "        \"value\": 42,\n",
        "        \"options\": {\n",
        "            \"process\": True\n",
        "        }\n",
        "    }\n",
        "    \n",
        "    # Process data\n",
        "    print(\"\\nProcessing data...\")\n",
        "    result = await tool.execute(input_data)\n",
        "    \n",
        "    print(\"\\nFinal result:\")\n",
        "    print(f\"ID: {result['id']}\")\n",
        "    print(f\"Value: {result['value']}\")\n",
        "    print(f\"Timestamp: {result['timestamp']}\")\n",
        "    print(f\"Metadata: {result['metadata']}\")\n",
        "    print(f\"Legacy Result: {result['legacy_result']}\")\n",
        "\n",
        "# Run test\n",
        "await test_composite_tool()\n"
      ]
    }
  ],
  "metadata": {
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}