[{'Text': '"""Tool management for dynamic tool registration with MCP notifications.\n\nThis module provides a ToolManager class that enables dynamic tool registration\nat runtime. When tools are added or removed through ToolManager, FastMCP\'s\nadd_tool/remove_tool methods are used, and developers can manually send\nchange notifications to clients using the MCP Context.\n\nSee: https://modelcontextprotocol.io/specification/2025-03-26/server/tools#list-changed-notification\n"""\n\nfrom typing import Callable, Optional, Any\nfrom dataclasses import dataclass\nfrom mcp.server.fastmcp import FastMCP\n\n\n@dataclass\nclass RegisteredTool:\n    """Information about a registered tool."""\n\n    name: str\n    """Unique name for the tool."""\n\n    description: str\n    """Human-readable description of the tool."""\n\n    function: Callable\n    """The actual function to call."""\n\n    annotations: Optional[dict[str, str]] = None\n    """Optional annotations describing tool behavior."""\n\n\nclass ToolManager:\n    """Manages dynamic tool registration.\n\n    This class provides a clean interface for adding and removing tools at runtime.\n    When tools are registered, they\'re added to FastMCP, which can then send\n    change notifications to clients.\n\n    Important: To notify clients when tools change, use the MCP Context\'s\n    send_notification method in your tool functions:\n\n        from fastmcp.server.context import Context\n        import mcp.types\n\n        @mcp.tool\n        async def my_tool(ctx: Context):\n            # After adding a tool:\n            await ctx.send_notification(mcp.types.ToolListChangedNotification())\n\n    Usage:\n        >>> from point_topic_mcp.core.tool_manager import ToolManager\n        >>>\n        >>> tool_manager = ToolManager(mcp)\n        >>>\n        >>> async def my_new_tool(param: str) -> str:\n        ...     return f"Result: {param}"\n        >>>\n        >>> tool_manager.register_tool(\n        ...     name="my_tool",\n        ...     description="A new tool",\n        ...     function=my_new_tool\n        ... )\n\n    Note:\n        Tools must be registered before the MCP server starts for startup tools.\n        For dynamic tool registration during runtime, use this manager after\n        the server has started.\n    """\n\n    def __init__(self, mcp_server: FastMCP):\n        """Initialize the tool manager.\n\n        Args:\n            mcp_server: The FastMCP server instance to manage tools for\n        """\n        self.mcp = mcp_server\n        self._registered_tools: dict[str, RegisteredTool] = {}\n\n    def register_tool(\n        self,\n        name: str,\n        description: str,\n        function: Callable,\n        annotations: Optional[dict[str, str]] = None,\n    ) -> None:\n        """Register a new tool with the MCP server.\n\n        After registering, you should send a change notification to clients:\n\n            await ctx.send_notification(mcp.types.ToolListChangedNotification())\n\n        Args:\n            name: Unique identifier for the tool\n            description: Human-readable description of what the tool does\n            function: The function to call when the tool is invoked\n            annotations: Optional annotations describing tool behavior\n\n        Raises:\n            ValueError: If a tool with the same name is already registered\n\n        Example:\n            >>> def get_user_data(user_id: int) -> dict:\n            ...     return {"id": user_id, "name": "John"}\n            >>>\n            >>> tool_manager.register_tool(\n            ...     name="get_user",\n            ...     description="Fetch user data by ID",\n            ...     function=get_user_data\n            ... )\n        """\n        if name in self._registered_tools:\n            raise ValueError(\n                f"Tool \'{name}\' is already registered. "\n                f"Use unregister_tool() to remove it first."\n            )\n\n        # Register the tool with FastMCP\n        self.mcp.add_tool(function, name=name)\n\n        # Track the tool in our registry\n        tool_info = RegisteredTool(\n            name=name,\n            description=description,\n            function=function,\n            annotations=annotations,\n        )\n        self._registered_tools[name] = tool_info\n\n    def unregister_tool(self, name: str) -> None:\n        """Unregister a previously registered tool.\n\n        After unregistering, you should send a change notification to clients:\n\n            await ctx.send_notification(mcp.types.ToolListChangedNotification())\n\n        Args:\n            name: The name of the tool to unregister\n\n        Raises:\n            ValueError: If the tool is not registered\n\n        Example:\n            >>> tool_manager.unregister_tool("get_user")\n        """\n        if name not in self._registered_tools:\n            raise ValueError(\n                f"Tool \'{name}\' is not registered. Use register_tool() to add it first."\n            )\n\n        # Remove from FastMCP\n        self.mcp.remove_tool(name)\n\n        # Remove from our registry\n        del self._registered_tools[name]\n\n    def list_tools(self) -> dict[str, RegisteredTool]:\n        """List all dynamically registered tools.\n\n        Note: This returns only tools registered through this ToolManager.\n        To get all tools including those registered at startup, use\n        `mcp.list_tools()` instead.\n\n        Returns:\n            Dictionary mapping tool names to RegisteredTool objects\n\n        Example:\n            >>> tools = tool_manager.list_tools()\n            >>> for name, tool_info in tools.items():\n            ...     print(f"{name}: {tool_info.description}")\n        """\n        return self._registered_tools.copy()\n\n    def get_tool(self, name: str) -> Optional[RegisteredTool]:\n        """Get information about a specific tool.\n\n        Args:\n            name: The name of the tool\n\n        Returns:\n            RegisteredTool information or None if not found\n\n        Example:\n            >>> tool = tool_manager.get_tool("get_user")\n            >>> if tool:\n            ...     print(f"Tool: {tool.name} - {tool.description}")\n        """\n        return self._registered_tools.get(name)\n\n    def is_registered(self, name: str) -> bool:\n        """Check if a tool is registered.\n\n        Args:\n            name: The name of the tool\n\n        Returns:\n            True if the tool is registered, False otherwise\n\n        Example:\n            >>> if not tool_manager.is_registered("get_user"):\n            ...     tool_manager.register_tool(...)\n        """\n        return name in self._registered_tools\n'}]