API Reference

This section provides detailed API documentation for HoloViz MCP.

Core Modules

holoviz_mcp

Accessible imports for the holoviz_mcp package.

mcp = FastMCP(name='holoviz', instructions="\n This MCP server provides comprehensive tools, resources and prompts for exploring data, creating data visualizations,\n data tools, dashboards and data apps using the HoloViz ecosystem.\n\n Use this MCP server to get help, resources, and prompts for working with the HoloViz ecosystem effectively.\n\n HoloViz provides a set of core Python packages that make visualization easier, more accurate, and more powerful:\n\n - [Colorcet](https://colorcet.holoviz.org): for perceptually uniform colormaps.\n - [Datashader](https://datashader.org): for rendering even the largest datasets.\n - [GeoViews](https://geoviews.org): to extend HoloViews for geographic data.\n - [HoloViews](https://holoviews.org): to create advanced, interactive and high quality data visualizations.\n - [hvPlot](https://hvplot.holoviz.org): to quickly generate interactive plots from your data.\n - [Lumen](https://lumen.holoviz.org): to build data-driven dashboards from a simple YAML specification that's well suited to modern AI tools like LLMs.\n - [Panel](https://panel.holoviz.org): for making data tools, dashboards and data apps using the Holoviz and wider PyData ecosystems.\n - [Param](https://param.holoviz.org): to create declarative user-configurable objects.\n ", lifespan=app_lifespan) module-attribute

Server

HoloViz MCP Server.

This MCP server provides comprehensive tools, resources and prompts for working with the HoloViz ecosystem, including Panel and hvPlot following best practices.

The server is composed of multiple sub-servers that provide various functionalities:

• Documentation: Search and access HoloViz documentation as context
• hvPlot: Tools, resources and prompts for using hvPlot to develop quick, interactive plots in Python
• Panel: Tools, resources and prompts for using Panel Material UI

logger = logging.getLogger(__name__) module-attribute

mcp = FastMCP(name='holoviz', instructions="\n This MCP server provides comprehensive tools, resources and prompts for exploring data, creating data visualizations,\n data tools, dashboards and data apps using the HoloViz ecosystem.\n\n Use this MCP server to get help, resources, and prompts for working with the HoloViz ecosystem effectively.\n\n HoloViz provides a set of core Python packages that make visualization easier, more accurate, and more powerful:\n\n - [Colorcet](https://colorcet.holoviz.org): for perceptually uniform colormaps.\n - [Datashader](https://datashader.org): for rendering even the largest datasets.\n - [GeoViews](https://geoviews.org): to extend HoloViews for geographic data.\n - [HoloViews](https://holoviews.org): to create advanced, interactive and high quality data visualizations.\n - [hvPlot](https://hvplot.holoviz.org): to quickly generate interactive plots from your data.\n - [Lumen](https://lumen.holoviz.org): to build data-driven dashboards from a simple YAML specification that's well suited to modern AI tools like LLMs.\n - [Panel](https://panel.holoviz.org): for making data tools, dashboards and data apps using the Holoviz and wider PyData ecosystems.\n - [Param](https://param.holoviz.org): to create declarative user-configurable objects.\n ", lifespan=app_lifespan) module-attribute

main()

Set up and run the composed MCP server.

Source code in src/holoviz_mcp/server.py

def main() -> None:
    """Set up and run the composed MCP server."""
    pid = f"Process ID: {os.getpid()}"
    print(pid)  # noqa: T201

    async def setup_and_run() -> None:
        await setup_composed_server()
        config = get_config()

        # Pass host and port for HTTP transport
        if config.server.transport == "http":
            await mcp.run_async(
                transport=config.server.transport,
                host=config.server.host,
                port=config.server.port,
            )
        else:
            await mcp.run_async(transport=config.server.transport)

    asyncio.run(setup_and_run())

setup_composed_server() async

Set up the composed server by importing all sub-servers with prefixes.

This uses static composition (import_server), which copies components from sub-servers into the main server with appropriate prefixes.

Source code in src/holoviz_mcp/server.py

async def setup_composed_server() -> None:
    """Set up the composed server by importing all sub-servers with prefixes.

    This uses static composition (import_server), which copies components
    from sub-servers into the main server with appropriate prefixes.
    """
    await mcp.import_server(holoviz_mcp, prefix="holoviz")
    await mcp.import_server(hvplot_mcp, prefix="hvplot")
    await mcp.import_server(panel_mcp, prefix="panel")
    await mcp.import_server(holoviews_mcp, prefix="holoviews")

Panel MCP

Panel MCP Server Package.

This package provides Model Context Protocol (MCP) tools for working with Panel, the Python library for creating interactive web applications and dashboards.

The package includes: - Component discovery and introspection tools - Parameter information extraction - URL proxying utilities for remote environments - Data models for component metadata

Main modules: - server: MCP server implementation with Panel-specific tools - data: Component metadata collection and utilities - models: Pydantic models for component information

Server

Panel MCP Server.

This MCP server provides tools, resources and prompts for using Panel to develop quick, interactive applications, tools and dashboards in Python using best practices.

Use this server to access:

• Panel Components: Detailed information about specific Panel components like widgets (input), panes (output) and layouts.

COMPONENTS = [] module-attribute

logger = logging.getLogger(__name__) module-attribute

mcp = FastMCP(name='panel', instructions='\n [Panel](https://panel.holoviz.org/) MCP Server.\n\n This MCP server provides tools, resources and prompts for using Panel to develop quick, interactive\n applications, tools and dashboards in Python using best practices.\n\n DO use this server to search for specific Panel components and access detailed information including docstrings and parameter information.\n ') module-attribute

PlaywrightManager

Persistent Playwright browser for fast repeated screenshots.

Source code in src/holoviz_mcp/panel_mcp/server.py

class PlaywrightManager:
    """Persistent Playwright browser for fast repeated screenshots."""

    def __init__(self):
        self._playwright = None
        self._browser = None
        self._lock = asyncio.Lock()

    async def get_browser(self):
        """Get a connected Playwright browser instance, launching if necessary."""
        async with self._lock:
            if self._browser is not None and self._browser.is_connected():
                return self._browser
            await self._cleanup()
            from playwright.async_api import async_playwright

            self._playwright = await async_playwright().start()
            self._browser = await self._playwright.chromium.launch(headless=True)
            return self._browser

    async def _cleanup(self):
        if self._browser is not None:
            try:
                await self._browser.close()
            except Exception:
                pass
            self._browser = None
        if self._playwright is not None:
            try:
                await self._playwright.stop()
            except Exception:
                pass
            self._playwright = None

    async def close(self):
        """Clean up Playwright resources."""
        async with self._lock:
            await self._cleanup()

close() async

Clean up Playwright resources.

Source code in src/holoviz_mcp/panel_mcp/server.py

async def close(self):
    """Clean up Playwright resources."""
    async with self._lock:
        await self._cleanup()

get_browser() async

Get a connected Playwright browser instance, launching if necessary.

Source code in src/holoviz_mcp/panel_mcp/server.py

async def get_browser(self):
    """Get a connected Playwright browser instance, launching if necessary."""
    async with self._lock:
        if self._browser is not None and self._browser.is_connected():
            return self._browser
        await self._cleanup()
        from playwright.async_api import async_playwright

        self._playwright = await async_playwright().start()
        self._browser = await self._playwright.chromium.launch(headless=True)
        return self._browser

get_component(ctx, name=None, module_path=None, package=None) async

Get complete details about a single Panel component including docstring and parameters.

Use this tool when you need full information about a specific Panel component, including its docstring, parameter specifications, and initialization signature. This is the most comprehensive tool for component information.

IMPORTANT: This tool returns exactly one component. If your criteria match multiple components, you'll get an error asking you to be more specific.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required
name	str	Component name to match (case-insensitive). If None, must specify other criteria. Examples: "Button", "TextInput", "Slider"	None
module_path	str	Full module path to match. If None, uses name and package to find component. Examples: "panel.widgets.Button", "panel_material_ui.Button"	None
package	str	Package name to filter by. If None, searches all packages. Examples: "panel" or "panel_material_ui"	None

Returns:

Type	Description
ComponentDetails	Complete component information including docstring, parameters, and initialization signature.

Raises:

Type	Description
ValueError	If no components match the criteria or if multiple components match (be more specific).

Examples:

Get Panel's Button component:

>>> get_component(name="Button", package="panel")
ComponentDetails(name="Button", package="panel", docstring="A clickable button...", parameters={...})

Get Material UI Button component:

>>> get_component(name="Button", package="panel_material_ui")
ComponentDetails(name="Button", package="panel_material_ui", ...)

Get component by exact module path:

>>> get_component(module_path="panel.widgets.button.Button")
ComponentDetails(name="Button", module_path="panel.widgets.button.Button", ...)

Source code in src/holoviz_mcp/panel_mcp/server.py

@mcp.tool()
async def get_component(ctx: Context, name: str | None = None, module_path: str | None = None, package: str | None = None) -> ComponentDetails:
    """
    Get complete details about a single Panel component including docstring and parameters.

    Use this tool when you need full information about a specific Panel component, including
    its docstring, parameter specifications, and initialization signature. This is the most
    comprehensive tool for component information.

    IMPORTANT: This tool returns exactly one component. If your criteria match multiple components,
    you'll get an error asking you to be more specific.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).
    name : str, optional
        Component name to match (case-insensitive). If None, must specify other criteria.
        Examples: "Button", "TextInput", "Slider"
    module_path : str, optional
        Full module path to match. If None, uses name and package to find component.
        Examples: "panel.widgets.Button", "panel_material_ui.Button"
    package : str, optional
        Package name to filter by. If None, searches all packages.
        Examples: "panel" or "panel_material_ui"

    Returns
    -------
    ComponentDetails
        Complete component information including docstring, parameters, and initialization signature.

    Raises
    ------
    ValueError
        If no components match the criteria or if multiple components match (be more specific).

    Examples
    --------
    Get Panel's Button component:
    >>> get_component(name="Button", package="panel")
    ComponentDetails(name="Button", package="panel", docstring="A clickable button...", parameters={...})

    Get Material UI Button component:
    >>> get_component(name="Button", package="panel_material_ui")
    ComponentDetails(name="Button", package="panel_material_ui", ...)

    Get component by exact module path:
    >>> get_component(module_path="panel.widgets.button.Button")
    ComponentDetails(name="Button", module_path="panel.widgets.button.Button", ...)
    """
    components_list = await _get_component(ctx, name, module_path, package)

    if not components_list:
        raise ValueError(f"No components found matching criteria: '{name}', '{module_path}', '{package}'. Please check your inputs.")
    if len(components_list) > 1:
        module_paths = "'" + "','".join([component.module_path for component in components_list]) + "'"
        raise ValueError(f"Multiple components found matching criteria: {module_paths}. Please refine your search.")
    component = components_list[0]
    return component

get_component_parameters(ctx, name=None, module_path=None, package=None) async

Get detailed parameter information for a single Panel component.

Use this tool when you need to understand the parameters of a specific Panel component, including their types, default values, documentation, and constraints. This is useful for understanding how to properly initialize and configure a component.

IMPORTANT: This tool returns parameters for exactly one component. If your criteria match multiple components, you'll get an error asking you to be more specific.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required
name	str	Component name to match (case-insensitive). If None, must specify other criteria. Examples: "Button", "TextInput", "Slider"	None
module_path	str	Full module path to match. If None, uses name and package to find component. Examples: "panel.widgets.Button", "panel_material_ui.Button"	None
package	str	Package name to filter by. If None, searches all packages. Examples: "hvplot", "panel" or "panel_material_ui"	None

Returns:

Type	Description
dict[str, ParameterInfo]	Dictionary mapping parameter names to their detailed information, including: - type: Parameter type (e.g., 'String', 'Number', 'Boolean') - default: Default value - doc: Parameter documentation - bounds: Value constraints for numeric parameters - objects: Available options for selector parameters

Raises:

Type	Description
ValueError	If no components match the criteria or if multiple components match (be more specific).

Examples:

Get Button parameters:

>>> get_component_parameters(name="Button", package="panel")
{"name": ParameterInfo(type="String", default="Button", doc="The text displayed on the button"), ...}

Get TextInput parameters:

>>> get_component_parameters(name="TextInput", package="panel")
{"value": ParameterInfo(type="String", default="", doc="The current text value"), ...}

Get parameters by exact module path:

>>> get_component_parameters(module_path="panel.widgets.Slider")
{"start": ParameterInfo(type="Number", default=0, bounds=(0, 100)), ...}

Source code in src/holoviz_mcp/panel_mcp/server.py

@mcp.tool()
async def get_component_parameters(ctx: Context, name: str | None = None, module_path: str | None = None, package: str | None = None) -> dict[str, ParameterInfo]:
    """
    Get detailed parameter information for a single Panel component.

    Use this tool when you need to understand the parameters of a specific Panel component,
    including their types, default values, documentation, and constraints. This is useful
    for understanding how to properly initialize and configure a component.

    IMPORTANT: This tool returns parameters for exactly one component. If your criteria
    match multiple components, you'll get an error asking you to be more specific.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).
    name : str, optional
        Component name to match (case-insensitive). If None, must specify other criteria.
        Examples: "Button", "TextInput", "Slider"
    module_path : str, optional
        Full module path to match. If None, uses name and package to find component.
        Examples: "panel.widgets.Button", "panel_material_ui.Button"
    package : str, optional
        Package name to filter by. If None, searches all packages.
        Examples: "hvplot", "panel" or "panel_material_ui"

    Returns
    -------
    dict[str, ParameterInfo]
        Dictionary mapping parameter names to their detailed information, including:
        - type: Parameter type (e.g., 'String', 'Number', 'Boolean')
        - default: Default value
        - doc: Parameter documentation
        - bounds: Value constraints for numeric parameters
        - objects: Available options for selector parameters

    Raises
    ------
    ValueError
        If no components match the criteria or if multiple components match (be more specific).

    Examples
    --------
    Get Button parameters:
    >>> get_component_parameters(name="Button", package="panel")
    {"name": ParameterInfo(type="String", default="Button", doc="The text displayed on the button"), ...}

    Get TextInput parameters:
    >>> get_component_parameters(name="TextInput", package="panel")
    {"value": ParameterInfo(type="String", default="", doc="The current text value"), ...}

    Get parameters by exact module path:
    >>> get_component_parameters(module_path="panel.widgets.Slider")
    {"start": ParameterInfo(type="Number", default=0, bounds=(0, 100)), ...}
    """
    components_list = await _get_component(ctx, name, module_path, package)

    if not components_list:
        raise ValueError(f"No components found matching criteria: '{name}', '{module_path}', '{package}'. Please check your inputs.")
    if len(components_list) > 1:
        module_paths = "'" + "','".join([component.module_path for component in components_list]) + "'"
        raise ValueError(f"Multiple components found matching criteria: {module_paths}. Please refine your search.")

    component = components_list[0]
    return component.parameters

inspect_app(url='http://localhost:5006/', width=1920, height=1200, full_page=False, delay=2, save_screenshot=False, screenshot=True, console_logs=True, log_level=None) async

Inspect a running Panel app by capturing a screenshot and/or browser console logs.

Panel apps (especially custom components) often have JavaScript errors that are invisible to users and LLMs. This tool captures both a visual screenshot and the browser console output in a single call, making it easy to diagnose rendering issues, JS errors, and runtime warnings.

Arguments

url : str, default="http://localhost:5006/" The URL of the page to inspect. width : int, default=1920 The width of the browser viewport. height : int, default=1200 The height of the browser viewport. full_page : bool, default=False Whether to capture the full scrollable page. delay : int, default=2 Seconds to wait after page load before capturing, to allow dynamic content to render. save_screenshot : bool | str, default=False Whether and where to save the screenshot to disk: - True: Save to default screenshots directory (~/.holoviz-mcp/screenshots/) with auto-generated filename - False: Don't save screenshot to disk (only return to AI) - str: Save to specified absolute path (raises ValueError if path is not absolute) screenshot : bool, default=True Whether to capture a screenshot of the page. console_logs : bool, default=True Whether to capture browser console log messages. log_level : str | None, default=None Filter console logs by level. If None, all levels are captured. Common levels: 'log', 'info', 'warning', 'error', 'debug'.

Source code in src/holoviz_mcp/panel_mcp/server.py

@mcp.tool()
async def inspect_app(
    url: str = "http://localhost:5006/",
    width: int = 1920,
    height: int = 1200,
    full_page: bool = False,
    delay: int = 2,
    save_screenshot: bool | str = False,
    screenshot: bool = True,
    console_logs: bool = True,
    log_level: str | None = None,
) -> list[TextContent | ImageContent]:
    """
    Inspect a running Panel app by capturing a screenshot and/or browser console logs.

    Panel apps (especially custom components) often have JavaScript errors that are
    invisible to users and LLMs. This tool captures both a visual screenshot and the
    browser console output in a single call, making it easy to diagnose rendering
    issues, JS errors, and runtime warnings.

    Arguments
    ----------
    url : str, default="http://localhost:5006/"
        The URL of the page to inspect.
    width : int, default=1920
        The width of the browser viewport.
    height : int, default=1200
        The height of the browser viewport.
    full_page : bool, default=False
        Whether to capture the full scrollable page.
    delay : int, default=2
        Seconds to wait after page load before capturing, to allow dynamic content to render.
    save_screenshot : bool | str, default=False
        Whether and where to save the screenshot to disk:
        - True: Save to default screenshots directory (~/.holoviz-mcp/screenshots/) with auto-generated filename
        - False: Don't save screenshot to disk (only return to AI)
        - str: Save to specified absolute path (raises ValueError if path is not absolute)
    screenshot : bool, default=True
        Whether to capture a screenshot of the page.
    console_logs : bool, default=True
        Whether to capture browser console log messages.
    log_level : str | None, default=None
        Filter console logs by level. If None, all levels are captured.
        Common levels: 'log', 'info', 'warning', 'error', 'debug'.
    """
    if not screenshot and not console_logs:
        raise ValueError("At least one of 'screenshot' or 'console_logs' must be True.")

    manager = _get_playwright_manager()
    browser = await manager.get_browser()
    page = await browser.new_page(
        ignore_https_errors=True,
        viewport={"width": width, "height": height},
    )

    # Collect console log entries
    collected_logs: list[ConsoleLogEntry] = []

    if console_logs:

        def _on_console(msg):
            collected_logs.append(
                ConsoleLogEntry(
                    level=msg.type,
                    message=msg.text,
                    timestamp=datetime.now().isoformat(),
                )
            )

        page.on("console", _on_console)

    try:
        await page.goto(url, wait_until="networkidle")
        await sleep(delay=delay)
        buffer = await page.screenshot(type="png", full_page=full_page) if screenshot else None
    finally:
        await page.close()

    result: list[TextContent | ImageContent] = []

    # Handle screenshot saving and result
    if screenshot and buffer is not None:
        # Coerce string "false"/"true" to bool (MCP clients may serialize bools as strings)
        save = save_screenshot
        if isinstance(save, str) and save.lower() in ("true", "false"):
            save = save.lower() == "true"

        if save:
            if isinstance(save, str):
                # Custom path specified
                save_path = Path(save)
                if not save_path.is_absolute():
                    raise ValueError(f"save_screenshot path must be absolute, got: {save_screenshot}")
            else:
                # Default path - use screenshots_dir from config
                screenshots_dir = _config.server.screenshots_dir
                screenshots_dir.mkdir(parents=True, exist_ok=True)

                # Generate filename with timestamp and UUID
                timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
                unique_id = str(uuid4())[:8]
                filename = f"screenshot_{timestamp}_{unique_id}.png"
                save_path = screenshots_dir / filename

            # Ensure parent directory exists
            save_path.parent.mkdir(parents=True, exist_ok=True)

            # Write the screenshot to disk
            save_path.write_bytes(buffer)
            logger.info(f"Screenshot saved to: {save_path}")

        image = Image(data=buffer, format="png")
        result.append(image.to_image_content())

    # Handle console logs result
    if console_logs:
        filtered_logs = collected_logs
        if log_level is not None:
            filtered_logs = [entry for entry in collected_logs if entry.level == log_level]

        logs_json = json.dumps([entry.model_dump() for entry in filtered_logs], indent=2)
        result.append(TextContent(type="text", text=logs_json))

    return result

list_components(ctx, name=None, module_path=None, package=None) async

Get a summary list of Panel components without detailed docstring and parameter information.

Use this tool to get an overview of available Panel components when you want to browse or discover components without needing full parameter details. This is faster than get_component and provides just the essential information.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required
name	str	Component name to filter by (case-insensitive). If None, returns all components. Examples: "Button", "TextInput", "Slider"	None
module_path	str	Module path prefix to filter by. If None, returns all components. Examples: "panel.widgets" to get all widgets, "panel.pane" to get all panes	None
package	str	Package name to filter by. If None, returns all components. Examples: "hvplot", "panel" or "panel_material_ui"	None

Returns:

Type	Description
list[ComponentSummary]	List of component summaries containing name, package, description, and module path. No parameter details are included for faster responses.

Examples:

Get all available components:

>>> list_components()
[ComponentSummary(name="Button", package="panel", description="A clickable button widget", ...)]

Get all Material UI components:

>>> list_components(package="panel_material_ui")
[ComponentSummary(name="Button", package="panel_material_ui", ...)]

Get all Button components from all packages:

>>> list_components(name="Button")
[ComponentSummary(name="Button", package="panel", ...), ComponentSummary(name="Button", package="panel_material_ui", ...)]

Source code in src/holoviz_mcp/panel_mcp/server.py

@mcp.tool()
async def list_components(ctx: Context, name: str | None = None, module_path: str | None = None, package: str | None = None) -> list[ComponentSummary]:
    """
    Get a summary list of Panel components without detailed docstring and parameter information.

    Use this tool to get an overview of available Panel components when you want to browse
    or discover components without needing full parameter details. This is faster than
    get_component and provides just the essential information.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).
    name : str, optional
        Component name to filter by (case-insensitive). If None, returns all components.
        Examples: "Button", "TextInput", "Slider"
    module_path : str, optional
        Module path prefix to filter by. If None, returns all components.
        Examples: "panel.widgets" to get all widgets, "panel.pane" to get all panes
    package : str, optional
        Package name to filter by. If None, returns all components.
        Examples: "hvplot", "panel" or "panel_material_ui"

    Returns
    -------
    list[ComponentSummary]
        List of component summaries containing name, package, description, and module path.
        No parameter details are included for faster responses.

    Examples
    --------
    Get all available components:
    >>> list_components()
    [ComponentSummary(name="Button", package="panel", description="A clickable button widget", ...)]

    Get all Material UI components:
    >>> list_components(package="panel_material_ui")
    [ComponentSummary(name="Button", package="panel_material_ui", ...)]

    Get all Button components from all packages:
    >>> list_components(name="Button")
    [ComponentSummary(name="Button", package="panel", ...), ComponentSummary(name="Button", package="panel_material_ui", ...)]
    """
    components_list = []

    for component in await _get_all_components(ctx=ctx):
        if name and component.name.lower() != name.lower():
            continue
        if package and component.package != package:
            continue
        if module_path and not component.module_path.startswith(module_path):
            continue
        components_list.append(component.to_base())

    return components_list

list_packages(ctx) async

List all installed packages that provide Panel UI components.

Use this tool to discover what Panel-related packages are available in your environment. This helps you understand which packages you can use in the 'package' parameter of other tools.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required

Returns:

Type	Description
list[str]	List of package names that provide Panel components, sorted alphabetically. Examples: ["panel"] or ["panel", "panel_material_ui"]

Examples:

Use this tool to see available packages:

>>> list_packages()
["panel", "panel_material_ui"]

Then use those package names in other tools:

>>> list_components(package="panel_material_ui")
>>> search("button", package="panel")

Source code in src/holoviz_mcp/panel_mcp/server.py

@mcp.tool()
async def list_packages(ctx: Context) -> list[str]:
    """
    List all installed packages that provide Panel UI components.

    Use this tool to discover what Panel-related packages are available in your environment.
    This helps you understand which packages you can use in the 'package' parameter of other tools.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).

    Returns
    -------
    list[str]
        List of package names that provide Panel components, sorted alphabetically.
        Examples: ["panel"] or ["panel", "panel_material_ui"]

    Examples
    --------
    Use this tool to see available packages:
    >>> list_packages()
    ["panel", "panel_material_ui"]

    Then use those package names in other tools:
    >>> list_components(package="panel_material_ui")
    >>> search("button", package="panel")
    """
    return sorted(set(component.package for component in await _get_all_components(ctx)))

search_components(ctx, query, package=None, limit=10) async

Search for Panel components by search query and optional package filter.

Use this tool to find components when you don't know the exact name but have keywords. The search looks through component names, module paths, and documentation to find matches.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required
query	str	Search term to look for. Can be component names, functionality keywords, or descriptions. Examples: "button", "input", "text", "chart", "plot", "slider", "select"	required
package	str	Package name to filter results. If None, searches all packages. Examples: "hvplot", "panel", or "panel_material_ui"	None
limit	int	Maximum number of results to return. Default is 10.	10

Returns:

Type	Description
list[ComponentSummarySearchResult]	List of matching components with relevance scores (0-100, where 100 is exact match). Results are sorted by relevance score in descending order.

Examples:

Search for button components:

>>> search_components("button")
[ComponentSummarySearchResult(name="Button", package="panel", relevance_score=80, ...)]

Search within a specific package:

>>> search_components("input", package="panel_material_ui")
[ComponentSummarySearchResult(name="TextInput", package="panel_material_ui", ...)]

Find chart components with limited results:

>>> search_components("chart", limit=5)
[ComponentSummarySearchResult(name="Bokeh", package="panel", ...)]

Source code in src/holoviz_mcp/panel_mcp/server.py

@mcp.tool()
async def search_components(ctx: Context, query: str, package: str | None = None, limit: int = 10) -> list[ComponentSummarySearchResult]:
    """
    Search for Panel components by search query and optional package filter.

    Use this tool to find components when you don't know the exact name but have keywords.
    The search looks through component names, module paths, and documentation to find matches.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).
    query : str
        Search term to look for. Can be component names, functionality keywords, or descriptions.
        Examples: "button", "input", "text", "chart", "plot", "slider", "select"
    package : str, optional
        Package name to filter results. If None, searches all packages.
        Examples: "hvplot", "panel", or "panel_material_ui"
    limit : int, optional
        Maximum number of results to return. Default is 10.

    Returns
    -------
    list[ComponentSummarySearchResult]
        List of matching components with relevance scores (0-100, where 100 is exact match).
        Results are sorted by relevance score in descending order.

    Examples
    --------
    Search for button components:
    >>> search_components("button")
    [ComponentSummarySearchResult(name="Button", package="panel", relevance_score=80, ...)]

    Search within a specific package:
    >>> search_components("input", package="panel_material_ui")
    [ComponentSummarySearchResult(name="TextInput", package="panel_material_ui", ...)]

    Find chart components with limited results:
    >>> search_components("chart", limit=5)
    [ComponentSummarySearchResult(name="Bokeh", package="panel", ...)]
    """
    query_lower = query.lower()

    matches = []
    for component in await _get_all_components(ctx=ctx):
        score = 0
        if package and component.package.lower() != package.lower():
            continue

        if component.name.lower() == query_lower or component.module_path.lower() == query_lower:
            score = 100
        elif query_lower in component.name.lower():
            score = 80
        elif query_lower in component.module_path.lower():
            score = 60
        elif query_lower in component.docstring.lower():
            score = 40
        elif any(word in component.docstring.lower() for word in query_lower.split()):
            score = 20

        if score > 0:
            matches.append(ComponentSummarySearchResult.from_component(component=component, relevance_score=score))

    matches.sort(key=lambda x: x.relevance_score, reverse=True)
    if len(matches) > limit:
        matches = matches[:limit]

    return matches

Models

Pydantic models for Panel component metadata collection.

This module defines the data models used to represent Panel UI component information, including parameter details, component summaries, and search results.

ComponentDetails

Bases: ComponentSummary

Complete information about a Panel UI component.

This model includes all available information about a component: summary information, initialization signature, full docstring, and detailed parameter specifications.

Source code in src/holoviz_mcp/panel_mcp/models.py

class ComponentDetails(ComponentSummary):
    """
    Complete information about a Panel UI component.

    This model includes all available information about a component:
    summary information, initialization signature, full docstring,
    and detailed parameter specifications.

    """

    init_signature: str = Field(description="Signature of the component's __init__ method.")
    docstring: str = Field(description="Docstring of the component, providing detailed information about its usage.")
    parameters: dict[str, ParameterInfo] = Field(
        description="Dictionary of parameters for the component, where keys are parameter names and values are ParameterInfo objects."
    )

    def to_base(self) -> ComponentSummary:
        """
        Convert to a basic component summary.

        Strips away detailed information to create a lightweight
        summary suitable for listings and overviews.

        Returns
        -------
        ComponentSummary
            A summary version of this component.
        """
        return ComponentSummary(
            module_path=self.module_path,
            name=self.name,
            package=self.package,
            description=self.description,
        )

docstring = Field(description='Docstring of the component, providing detailed information about its usage.') class-attribute instance-attribute

init_signature = Field(description="Signature of the component's __init__ method.") class-attribute instance-attribute

parameters = Field(description='Dictionary of parameters for the component, where keys are parameter names and values are ParameterInfo objects.') class-attribute instance-attribute

to_base()

Convert to a basic component summary.

Strips away detailed information to create a lightweight summary suitable for listings and overviews.

Returns:

Type	Description
ComponentSummary	A summary version of this component.

Source code in src/holoviz_mcp/panel_mcp/models.py

def to_base(self) -> ComponentSummary:
    """
    Convert to a basic component summary.

    Strips away detailed information to create a lightweight
    summary suitable for listings and overviews.

    Returns
    -------
    ComponentSummary
        A summary version of this component.
    """
    return ComponentSummary(
        module_path=self.module_path,
        name=self.name,
        package=self.package,
        description=self.description,
    )

ComponentSummary

Bases: BaseModel

High-level information about a Panel UI component.

This model provides a compact representation of a component without detailed parameter information or docstrings. Used for listings and quick overviews.

Source code in src/holoviz_mcp/panel_mcp/models.py

class ComponentSummary(BaseModel):
    """
    High-level information about a Panel UI component.

    This model provides a compact representation of a component without
    detailed parameter information or docstrings. Used for listings and
    quick overviews.
    """

    module_path: str = Field(description="Full module path of the component, e.g., 'panel.widgets.Button' or 'panel_material_ui.Button'.")
    name: str = Field(description="Name of the component, e.g., 'Button' or 'TextInput'.")
    package: str = Field(description="Package name of the component, e.g., 'panel' or 'panel_material_ui'.")
    description: str = Field(description="Short description of the component's purpose and functionality.")

description = Field(description="Short description of the component's purpose and functionality.") class-attribute instance-attribute

module_path = Field(description="Full module path of the component, e.g., 'panel.widgets.Button' or 'panel_material_ui.Button'.") class-attribute instance-attribute

name = Field(description="Name of the component, e.g., 'Button' or 'TextInput'.") class-attribute instance-attribute

package = Field(description="Package name of the component, e.g., 'panel' or 'panel_material_ui'.") class-attribute instance-attribute

ComponentSummarySearchResult

Bases: ComponentSummary

Component summary with search relevance scoring.

Extends ComponentSummary with a relevance score for search results, allowing proper ranking and filtering of search matches.

Source code in src/holoviz_mcp/panel_mcp/models.py

class ComponentSummarySearchResult(ComponentSummary):
    """
    Component summary with search relevance scoring.

    Extends ComponentSummary with a relevance score for search results,
    allowing proper ranking and filtering of search matches.

    """

    relevance_score: int = Field(default=0, description="Relevance score for search results")

    @classmethod
    def from_component(cls, component: ComponentDetails, relevance_score: int) -> ComponentSummarySearchResult:
        """
        Create a search result from a component and relevance score.

        Parameters
        ----------
        component : ComponentDetails
            The component to create a search result from.
        relevance_score : int
            The relevance score (0-100) for this search result.

        Returns
        -------
        ComponentSummarySearchResult
            A search result summary of the component.
        """
        return cls(
            module_path=component.module_path, name=component.name, package=component.package, description=component.description, relevance_score=relevance_score
        )

relevance_score = Field(default=0, description='Relevance score for search results') class-attribute instance-attribute

from_component(component, relevance_score) classmethod

Create a search result from a component and relevance score.

Parameters:

Name	Type	Description	Default
component	ComponentDetails	The component to create a search result from.	required
relevance_score	int	The relevance score (0-100) for this search result.	required

Returns:

Type	Description
ComponentSummarySearchResult	A search result summary of the component.

Source code in src/holoviz_mcp/panel_mcp/models.py

@classmethod
def from_component(cls, component: ComponentDetails, relevance_score: int) -> ComponentSummarySearchResult:
    """
    Create a search result from a component and relevance score.

    Parameters
    ----------
    component : ComponentDetails
        The component to create a search result from.
    relevance_score : int
        The relevance score (0-100) for this search result.

    Returns
    -------
    ComponentSummarySearchResult
        A search result summary of the component.
    """
    return cls(
        module_path=component.module_path, name=component.name, package=component.package, description=component.description, relevance_score=relevance_score
    )

ConsoleLogEntry

Bases: BaseModel

A single browser console log entry captured during app inspection.

Source code in src/holoviz_mcp/panel_mcp/models.py

class ConsoleLogEntry(BaseModel):
    """A single browser console log entry captured during app inspection."""

    level: str = Field(description="Console message level: 'log', 'info', 'warning', 'error', 'debug', etc.")
    message: str = Field(description="The text content of the console message.")
    timestamp: Optional[str] = Field(default=None, description="ISO 8601 timestamp when the message was captured.")

level = Field(description="Console message level: 'log', 'info', 'warning', 'error', 'debug', etc.") class-attribute instance-attribute

message = Field(description='The text content of the console message.') class-attribute instance-attribute

timestamp = Field(default=None, description='ISO 8601 timestamp when the message was captured.') class-attribute instance-attribute

ParameterInfo

Bases: BaseModel

Information about a Panel component parameter.

This model captures parameter metadata including type, default value, documentation, and type-specific attributes like bounds or options.

Source code in src/holoviz_mcp/panel_mcp/models.py

class ParameterInfo(BaseModel):
    """
    Information about a Panel component parameter.

    This model captures parameter metadata including type, default value,
    documentation, and type-specific attributes like bounds or options.
    """

    model_config = ConfigDict(extra="allow")  # Allow additional fields we don't know about

    # Common attributes that most parameters have
    type: str = Field(description="The type of the parameter, e.g., 'Parameter', 'Number', 'Selector'.")
    default: Optional[Any] = Field(default=None, description="The default value for the parameter.")
    doc: Optional[str] = Field(default=None, description="Documentation string for the parameter.")
    # Optional attributes that may not be present
    allow_None: Optional[bool] = Field(default=None, description="Whether the parameter accepts None values.")
    constant: Optional[bool] = Field(default=None, description="Whether the parameter is constant (cannot be changed after initialization).")
    readonly: Optional[bool] = Field(default=None, description="Whether the parameter is read-only.")
    per_instance: Optional[bool] = Field(default=None, description="Whether the parameter is per-instance or shared across instances.")

    # Type-specific attributes (will be present only for relevant parameter types)
    objects: Optional[Any] = Field(default=None, description="Available options for Selector-type parameters.")
    bounds: Optional[Any] = Field(default=None, description="Value bounds for Number-type parameters.")
    regex: Optional[str] = Field(default=None, description="Regular expression pattern for String-type parameters.")

allow_None = Field(default=None, description='Whether the parameter accepts None values.') class-attribute instance-attribute

bounds = Field(default=None, description='Value bounds for Number-type parameters.') class-attribute instance-attribute

constant = Field(default=None, description='Whether the parameter is constant (cannot be changed after initialization).') class-attribute instance-attribute

default = Field(default=None, description='The default value for the parameter.') class-attribute instance-attribute

doc = Field(default=None, description='Documentation string for the parameter.') class-attribute instance-attribute

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

objects = Field(default=None, description='Available options for Selector-type parameters.') class-attribute instance-attribute

per_instance = Field(default=None, description='Whether the parameter is per-instance or shared across instances.') class-attribute instance-attribute

readonly = Field(default=None, description='Whether the parameter is read-only.') class-attribute instance-attribute

regex = Field(default=None, description='Regular expression pattern for String-type parameters.') class-attribute instance-attribute

type = Field(description="The type of the parameter, e.g., 'Parameter', 'Number', 'Selector'.") class-attribute instance-attribute

Data

Data collection module for Panel component metadata.

This module provides functionality to collect metadata about Panel UI components, including their documentation, parameter schema, and module information. It supports collecting information from panel.viewable.Viewable subclasses across different Panel-related packages.

collect_component_info(cls)

Collect comprehensive information about a Panel component class.

Extracts metadata including docstring, parameter information, method signatures, and other relevant details from a Panel component class. Handles parameter introspection safely, converting non-serializable values appropriately.

Parameters:

Name	Type	Description	Default
cls	type	The Panel component class to analyze.	required

Returns:

Type	Description
ComponentDetails	A complete model containing all collected component information.

Source code in src/holoviz_mcp/panel_mcp/data.py

def collect_component_info(cls: type) -> ComponentDetails:
    """
    Collect comprehensive information about a Panel component class.

    Extracts metadata including docstring, parameter information, method signatures,
    and other relevant details from a Panel component class. Handles parameter
    introspection safely, converting non-serializable values appropriately.

    Parameters
    ----------
    cls : type
        The Panel component class to analyze.

    Returns
    -------
    ComponentDetails
        A complete model containing all collected component information.
    """
    # Extract docstring
    docstring = cls.__doc__ if cls.__doc__ else ""

    # Extract description (first sentence from docstring)
    description = ""
    if docstring:
        # Clean the docstring and get first sentence
        cleaned_docstring = docstring.strip()
        if cleaned_docstring:
            # Find first sentence ending with period, exclamation, or question mark
            import re

            sentences = re.split(r"[.!?]", cleaned_docstring)
            if sentences:
                description = sentences[0].strip()
                # Remove leading/trailing whitespace and normalize spaces
                description = " ".join(description.split())

    # Extract parameters information
    parameters = {}
    if hasattr(cls, "param"):
        for param_name in sorted(cls.param):
            # Skip private parameters
            if param_name.startswith("_"):
                continue

            param_obj = cls.param[param_name]
            param_data = {}

            # Get common parameter attributes (skip private ones)
            for attr in ["default", "doc", "allow_None", "constant", "readonly", "per_instance"]:
                if hasattr(param_obj, attr) and getattr(param_obj, attr):
                    value = getattr(param_obj, attr)
                    if isinstance(value, str):
                        value = dedent(value).strip()
                    # Handle non-JSON serializable values
                    try:
                        json.dumps(value)
                        param_data[attr] = value
                    except (TypeError, ValueError):
                        param_data[attr] = "NON_JSON_SERIALIZABLE_VALUE"

            # Get type-specific attributes
            param_type = type(param_obj).__name__
            param_data["type"] = param_type

            # For Selector parameters, get options
            if hasattr(param_obj, "objects") and param_obj.objects:
                try:
                    json.dumps(param_obj.objects)
                    param_data["objects"] = param_obj.objects
                except (TypeError, ValueError):
                    param_data["objects"] = "NON_JSON_SERIALIZABLE_VALUE"

            # For Number parameters, get bounds
            if hasattr(param_obj, "bounds") and param_obj.bounds:
                try:
                    json.dumps(param_obj.bounds)
                    param_data["bounds"] = param_obj.bounds
                except (TypeError, ValueError):
                    param_data["bounds"] = "NON_JSON_SERIALIZABLE_VALUE"

            # For String parameters, get regex
            if hasattr(param_obj, "regex") and param_obj.regex:
                try:
                    json.dumps(param_obj.regex)
                    param_data["regex"] = param_obj.regex
                except (TypeError, ValueError):
                    param_data["regex"] = "NON_JSON_SERIALIZABLE_VALUE"

            # Create ParameterInfo model
            parameters[param_name] = ParameterInfo(**param_data)

    # Get __init__ method signature
    init_signature = ""
    if hasattr(cls, "__init__"):
        try:
            import inspect

            sig = inspect.signature(cls.__init__)  # type: ignore[misc]
            init_signature = str(sig)
        except Exception as e:
            init_signature = f"Error getting signature: {e}"

    # Read reference guide content
    # Create and return ComponentInfo model
    return ComponentDetails(
        name=cls.__name__,
        description=description,
        package=cls.__module__.split(".")[0],
        module_path=f"{cls.__module__}.{cls.__name__}",
        init_signature=init_signature,
        docstring=docstring,
        parameters=parameters,
    )

find_all_subclasses(cls)

Recursively find all subclasses of a given class.

This function performs a depth-first search through the class hierarchy to find all classes that inherit from the given base class, either directly or through inheritance chains.

Parameters:

Name	Type	Description	Default
cls	type	The base class to find subclasses for.	required

Returns:

Type	Description
set[type]	Set of all subclasses found recursively, not including the base class itself.

Source code in src/holoviz_mcp/panel_mcp/data.py

def find_all_subclasses(cls: type) -> set[type]:
    """
    Recursively find all subclasses of a given class.

    This function performs a depth-first search through the class hierarchy
    to find all classes that inherit from the given base class, either directly
    or through inheritance chains.

    Parameters
    ----------
    cls : type
        The base class to find subclasses for.

    Returns
    -------
    set[type]
        Set of all subclasses found recursively, not including the base class itself.
    """
    subclasses = set()
    for subclass in cls.__subclasses__():
        subclasses.add(subclass)
        subclasses.update(find_all_subclasses(subclass))
    return subclasses

get_components(parent=Viewable)

Get detailed information about all Panel component subclasses.

Discovers all subclasses of the specified parent class (typically Viewable), filters out private classes, and collects comprehensive metadata for each. Results are sorted alphabetically by module path for consistency.

Parameters:

Name	Type	Description	Default
parent	type	The parent class to search for subclasses. Defaults to panel.viewable.Viewable.	Viewable

Returns:

Type	Description
list[ComponentDetails]	List of detailed component information models, sorted by module path.

Source code in src/holoviz_mcp/panel_mcp/data.py

def get_components(parent=Viewable) -> list[ComponentDetails]:
    """
    Get detailed information about all Panel component subclasses.

    Discovers all subclasses of the specified parent class (typically Viewable),
    filters out private classes, and collects comprehensive metadata for each.
    Results are sorted alphabetically by module path for consistency.

    Parameters
    ----------
    parent : type, optional
        The parent class to search for subclasses. Defaults to panel.viewable.Viewable.

    Returns
    -------
    list[ComponentDetails]
        List of detailed component information models, sorted by module path.
    """
    all_subclasses = find_all_subclasses(parent)

    # Filter to only those in panel_material_ui package and exclude private classes
    subclasses = [cls for cls in all_subclasses if not cls.__name__.startswith("_")]

    # Collect component information
    component_data = [collect_component_info(cls) for cls in subclasses]

    # Sort by module_path for consistent ordering
    component_data.sort(key=lambda x: x.module_path)
    return component_data

load_components(filepath)

Load component data from a JSON file.

Reads and deserializes component data that was previously saved using save_components(). Validates the file exists before attempting to load.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the saved component data JSON file.	required

Returns:

Type	Description
list[ComponentDetails]	Loaded component data as Pydantic model instances.

Raises:

Type	Description
FileNotFoundError	If the specified file does not exist.

Source code in src/holoviz_mcp/panel_mcp/data.py

def load_components(filepath: str) -> list[ComponentDetails]:
    """
    Load component data from a JSON file.

    Reads and deserializes component data that was previously saved using
    save_components(). Validates the file exists before attempting to load.

    Parameters
    ----------
    filepath : str
        Path to the saved component data JSON file.

    Returns
    -------
    list[ComponentDetails]
        Loaded component data as Pydantic model instances.

    Raises
    ------
    FileNotFoundError
        If the specified file does not exist.
    """
    file_path = Path(filepath)

    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {filepath}")

    with open(file_path, "r", encoding="utf-8") as f:
        json_data = json.load(f)

    # Convert JSON data back to Pydantic models
    return [ComponentDetails(**item) for item in json_data]

save_components(data, filename)

Save component data to a JSON file.

Serializes a list of ComponentDetails objects to JSON format for persistence. The JSON is formatted with indentation for human readability.

Parameters:

Name	Type	Description	Default
data	list[ComponentDetails]	Component data to save, typically from get_components().	required
filename	str	Path where the JSON file should be created.	required

Returns:

Type	Description
str	Absolute path to the created file.

Source code in src/holoviz_mcp/panel_mcp/data.py

def save_components(data: list[ComponentDetails], filename: str) -> str:
    """
    Save component data to a JSON file.

    Serializes a list of ComponentDetails objects to JSON format for persistence.
    The JSON is formatted with indentation for human readability.

    Parameters
    ----------
    data : list[ComponentDetails]
        Component data to save, typically from get_components().
    filename : str
        Path where the JSON file should be created.

    Returns
    -------
    str
        Absolute path to the created file.
    """
    filepath = Path(filename)

    # Convert Pydantic models to dict for JSON serialization
    json_data = [component.model_dump() for component in data]

    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(json_data, f, indent=2, ensure_ascii=False)

    return str(filepath)

to_proxy_url(url, jupyter_server_proxy_url='')

Convert localhost URLs to Jupyter server proxy URLs when applicable.

This function handles URL conversion for environments where localhost access needs to be proxied (like JupyterHub, Binder, etc.). It supports both 'localhost' and '127.0.0.1' addresses and preserves paths and query parameters.

Parameters:

Name	Type	Description	Default
url	str	The original URL to potentially convert. Can be any URL, but only localhost and 127.0.0.1 URLs will be converted.	required
jupyter_server_proxy_url	str	Base URL for the Jupyter server proxy. If None or empty, no conversion is performed. Defaults to the configured proxy URL.	''

Returns:

Type	Description
str	The converted proxy URL if applicable, otherwise the original URL. Proxy URLs maintain the original port, path, and query parameters.

Examples:

>>> to_proxy_url("http://localhost:5007/app")
"https://hub.example.com/user/alice/proxy/5007/app"

>>> to_proxy_url("https://external.com/page")
"https://external.com/page"  # No conversion for external URLs

Source code in src/holoviz_mcp/panel_mcp/data.py

def to_proxy_url(url: str, jupyter_server_proxy_url: str = "") -> str:
    """
    Convert localhost URLs to Jupyter server proxy URLs when applicable.

    This function handles URL conversion for environments where localhost access
    needs to be proxied (like JupyterHub, Binder, etc.). It supports both
    'localhost' and '127.0.0.1' addresses and preserves paths and query parameters.

    Parameters
    ----------
    url : str
        The original URL to potentially convert. Can be any URL, but only
        localhost and 127.0.0.1 URLs will be converted.
    jupyter_server_proxy_url : str, optional
        Base URL for the Jupyter server proxy. If None or empty, no conversion
        is performed. Defaults to the configured proxy URL.

    Returns
    -------
    str
        The converted proxy URL if applicable, otherwise the original URL.
        Proxy URLs maintain the original port, path, and query parameters.

    Examples
    --------
    >>> to_proxy_url("http://localhost:5007/app")
    "https://hub.example.com/user/alice/proxy/5007/app"

    >>> to_proxy_url("https://external.com/page")
    "https://external.com/page"  # No conversion for external URLs
    """
    if jupyter_server_proxy_url and jupyter_server_proxy_url.strip():
        # Check if this is a localhost or 127.0.0.1 URL
        if url.startswith("http://localhost:"):
            # Parse the URL to extract port, path, and query
            url_parts = url.replace("http://localhost:", "")
        elif url.startswith("http://127.0.0.1:"):
            # Parse the URL to extract port, path, and query
            url_parts = url.replace("http://127.0.0.1:", "")
        else:
            # Not a local URL, return original
            proxy_url = url
            return proxy_url

        # Find the port (everything before the first slash or end of string)
        if "/" in url_parts:
            port = url_parts.split("/", 1)[0]
            path_and_query = "/" + url_parts.split("/", 1)[1]
        else:
            port = url_parts
            path_and_query = "/"

        # Validate that port is a valid number
        if port and port.isdigit() and 1 <= int(port) <= 65535:
            # Build the proxy URL
            proxy_url = f"{jupyter_server_proxy_url}{port}{path_and_query}"
        else:
            # Invalid port, return original URL
            proxy_url = url
    else:
        proxy_url = url
    return proxy_url

hvPlot MCP

hvPlot MCP Server.

Server

hvPlot MCP Server.

This MCP server provides tools, resources, and prompts for using hvPlot to develop quick, interactive plots in Python using best practices.

Use this server to: - List available hvPlot plot types (e.g., 'line', 'scatter', 'bar', ...) - Get docstrings and function signatures for hvPlot plot types

mcp = FastMCP(name='hvplot', instructions='\n [hvPlot](https://hvplot.holoviz.org/) MCP Server.\n\n This MCP server provides tools, resources, and prompts for using hvPlot to develop quick, interactive plots\n in Python using best practices. Use this server to:\n\n - List available hvPlot plot types\n - Get docstrings and function signatures for hvPlot plot types') module-attribute

get_docstring(ctx, plot_type, docstring=True, generic=True, style=True) async

Get the hvPlot docstring for a specific plot type, including available options and usage details.

Use this tool to retrieve the full docstring for a plot type, including generic and style options. Equivalent to hvplot.help(plot_type) in the hvPlot API.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required
plot_type	str	The type of plot to provide help for (e.g., 'line', 'scatter').	required
docstring	bool	Whether to include the docstring in the output.	True
generic	bool	Whether to include generic plotting options shared by all plot types.	True
style	str or bool	Plotting backend to use for style options. If True, automatically infers the backend.	True

Returns:

Type	Description
str	The docstring for the specified plot type, including all relevant options and usage information.

Examples:

>>> get_docstring(plot_type='line')

Source code in src/holoviz_mcp/hvplot_mcp/server.py

@mcp.tool()
async def get_docstring(
    ctx: Context, plot_type: str, docstring: bool = True, generic: bool = True, style: Union[Literal["matplotlib", "bokeh", "plotly"], bool] = True
) -> str:
    """
    Get the hvPlot docstring for a specific plot type, including available options and usage details.

    Use this tool to retrieve the full docstring for a plot type, including generic and style options.
    Equivalent to `hvplot.help(plot_type)` in the hvPlot API.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).
    plot_type : str
        The type of plot to provide help for (e.g., 'line', 'scatter').
    docstring : bool, default=True
        Whether to include the docstring in the output.
    generic : bool, default=True
        Whether to include generic plotting options shared by all plot types.
    style : str or bool, default=True
        Plotting backend to use for style options. If True, automatically infers the backend.

    Returns
    -------
    str
        The docstring for the specified plot type, including all relevant options and usage information.

    Examples
    --------
    >>> get_docstring(plot_type='line')
    """
    doc, _ = _help(plot_type=plot_type, docstring=docstring, generic=generic, style=style)
    return doc

get_signature(ctx, plot_type, style=True) async

Get the function signature for a specific hvPlot plot type.

Use this tool to retrieve the Python function signature for a plot type, showing all accepted arguments and their defaults.

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required
plot_type	str	The type of plot to provide help for (e.g., 'line', 'scatter').	required
style	str or bool	Plotting backend to use for style options. If True, automatically infers the backend (ignored here).	True

Returns:

Type	Description
str	The function signature for the specified plot type.

Examples:

>>> get_signature(plot_type='line')

Source code in src/holoviz_mcp/hvplot_mcp/server.py

@mcp.tool()
async def get_signature(ctx: Context, plot_type: str, style: Union[Literal["matplotlib", "bokeh", "plotly"], bool] = True) -> str:
    """
    Get the function signature for a specific hvPlot plot type.

    Use this tool to retrieve the Python function signature for a plot type, showing all accepted arguments and their defaults.

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).
    plot_type : str
        The type of plot to provide help for (e.g., 'line', 'scatter').
    style : str or bool, default=True
        Plotting backend to use for style options. If True, automatically infers the backend (ignored here).

    Returns
    -------
    str
        The function signature for the specified plot type.

    Examples
    --------
    >>> get_signature(plot_type='line')
    """
    _, sig = _help(plot_type=plot_type, docstring=True, generic=True, style=style)
    return str(sig)

list_plot_types(ctx) async

List all available hvPlot plot types supported in the current environment.

Use this tool to discover what plot types you can generate with hvPlot.

Note: The plot types are also called "kinds".

Parameters:

Name	Type	Description	Default
ctx	Context	FastMCP context (automatically provided by the MCP framework).	required

Returns:

Type	Description
list[str]	Sorted list of all plot type names (e.g., 'line', 'scatter', 'bar', ...).

Examples:

>>> list_plot_types()
['area', 'bar', 'box', 'contour', ...]

Source code in src/holoviz_mcp/hvplot_mcp/server.py

@mcp.tool()
async def list_plot_types(ctx: Context) -> list[str]:
    """
    List all available hvPlot plot types supported in the current environment.

    Use this tool to discover what plot types you can generate with hvPlot.

    Note: The plot types are also called "kinds".

    Parameters
    ----------
    ctx : Context
        FastMCP context (automatically provided by the MCP framework).

    Returns
    -------
    list[str]
        Sorted list of all plot type names (e.g., 'line', 'scatter', 'bar', ...).

    Examples
    --------
    >>> list_plot_types()
    ['area', 'bar', 'box', 'contour', ...]
    """
    from hvplot.converter import HoloViewsConverter

    return sorted(HoloViewsConverter._kind_mapping)

Documentation MCP

docs_mcp package.

Server

HoloViz Documentation MCP Server.

This server provides tools, resources and prompts for accessing documentation related to the HoloViz ecosystems and any user-defined/internal documentation you have configured.

Use this server to search and access documentation for HoloViz libraries (Panel, hvPlot, etc.) and your custom projects.

config = get_config() module-attribute

mcp = FastMCP(name='documentation', instructions='\n [HoloViz](https://holoviz.org/) Documentation MCP Server.\n\n This server provides tools, resources and prompts for accessing documentation related to the HoloViz ecosystems\n and any user-defined/internal documentation you have configured.\n\n Use this server to search and access documentation for HoloViz libraries (Panel, hvPlot, etc.) and your custom projects.\n ') module-attribute

app_lifespan(server) async

Lifespan context manager for HoloViz MCP server.

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Lifespan context manager for HoloViz MCP server."""
    # Initialize resources on startup
    try:
        config = get_config()
        # Only start display manager if in subprocess mode
        if config.display.enabled and config.display.mode == "subprocess":
            _get_display_manager()  # Ensure display manager is started
        yield None
    except Exception as e:
        logger.error(f"Error during app lifespan: {e}")
        raise
    finally:
        # Clean up resources on shutdown
        pass

display(code, name='', description='', method='jupyter', ctx=None) async

Display Python code visualization in a browser.

This tool executes Python code and renders it in a Panel web interface, returning a URL where you can view the output. The code is validated before execution and any errors are reported immediately.

Use this tool to when ever the user asks to show, display, visualize data, plots, dashboards, and other Python objects.

Parameters:

Name	Type	Description	Default
code	str	The Python code to execute. For "jupyter" method, the last line is displayed. For "panel" method, objects marked .servable() are displayed.	required
name	str	A name for the visualization (displayed in admin/feed views)	''
description	str	A short description of the visualization	''
method	(jupyter, panel)	Execution mode: - "jupyter": Execute code and display the last expression's result. The last expression must be dedented fully. DO use this for standard data visualizations like plots, dataframes, etc. that do not import and use Panel directly. - "panel": Execute code and and display Panel objects marked .servable() DO use this for code that imports and uses Panel to create dashboards, apps, and complex layouts.	"jupyter"

Returns:

Type	Description
str	URL to view the rendered visualization (e.g., http://localhost:5005/view?id=abc123)

Raises:

Type	Description
RuntimeError	If the display server is not enabled or not running
ValueError	If code execution fails (syntax error, runtime error)

Examples:

Simple visualization with jupyter method:

>>> code = '''
... import pandas as pd
... df = pd.DataFrame({'x': [1, 2, 3], 'y': [4, 5, 6]})
... df
... '''
>>> url = await display(code, name="Sample DataFrame")

Panel app with servable:

>>> code = '''
... import panel as pn
... pn.extension()
... pn.pane.Markdown("# Hello World").servable()
... '''
>>> url = await display(code, method="panel")

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
async def display(
    code: str,
    name: str = "",
    description: str = "",
    method: Literal["jupyter", "panel"] = "jupyter",
    ctx: Context | None = None,
) -> str:
    """Display Python code visualization in a browser.

    This tool executes Python code and renders it in a Panel web interface,
    returning a URL where you can view the output. The code is validated
    before execution and any errors are reported immediately.

    Use this tool to when ever the user asks to show, display, visualize data, plots, dashboards, and other Python objects.

    Parameters
    ----------
    code : str
        The Python code to execute. For "jupyter" method, the last line is displayed.
        For "panel" method, objects marked .servable() are displayed.
    name : str, optional
        A name for the visualization (displayed in admin/feed views)
    description : str, optional
        A short description of the visualization
    method : {"jupyter", "panel"}, default "jupyter"
        Execution mode:
        - "jupyter": Execute code and display the last expression's result. The last expression must be dedented fully.
            DO use this for standard data visualizations like plots, dataframes, etc. that do not import and use Panel directly.
        - "panel": Execute code and and display Panel objects marked .servable()
            DO use this for code that imports and uses Panel to create dashboards, apps, and complex layouts.

    Returns
    -------
    str
        URL to view the rendered visualization (e.g., http://localhost:5005/view?id=abc123)

    Raises
    ------
    RuntimeError
        If the display server is not enabled or not running
    ValueError
        If code execution fails (syntax error, runtime error)

    Examples
    --------
    Simple visualization with jupyter method:
    >>> code = '''
    ... import pandas as pd
    ... df = pd.DataFrame({'x': [1, 2, 3], 'y': [4, 5, 6]})
    ... df
    ... '''
    >>> url = await display(code, name="Sample DataFrame")

    Panel app with servable:
    >>> code = '''
    ... import panel as pn
    ... pn.extension()
    ... pn.pane.Markdown("# Hello World").servable()
    ... '''
    >>> url = await display(code, method="panel")
    """
    config = get_config()

    if not config.display.enabled:
        return "Error: Display server is not enabled. Set display.enabled=true in config."

    # Get client
    client = _get_display_client()
    if not client:
        return "Error: Failed to initialize display client. Check logs for details."

    # Check health with mode-aware logic
    if not client.is_healthy():
        if config.display.mode == "subprocess":
            # Try to restart in subprocess mode
            if ctx:
                await ctx.info("Display server is not healthy, attempting restart...")

            manager = _get_display_manager()
            if manager and manager.restart():
                # Recreate client with new base URL
                global _display_client
                if _display_client:
                    _display_client.close()
                _display_client = DisplayClient(base_url=manager.get_base_url())
                client = _display_client
            else:
                return "Error: Display server is not healthy and failed to restart."
        else:
            # Fail fast in remote mode
            return "Error: Display server is not healthy. Check remote server status."

    # Send request to Panel server
    try:
        response = client.create_snippet(
            code=code,
            name=name,
            description=description,
            method=method,
        )
        url = response.get("url", "")

        # Check for errors in response
        if error_message := response.get("error_message", None):
            return f"""
Visualization created with errors. View here {url}

{error_message}
"""

        return f"Visualization created successfully!\n\nView here {url}"

    except Exception as e:
        logger.exception(f"Error creating visualization: {e}")

        if ctx:
            await ctx.error(f"Failed to create visualization: {e}")

        return f"Error: Failed to create visualization: {str(e)}"

get_document(path, project, ctx) async

Retrieve a specific document by path and project.

Use this tool to look up a specific document within a project.

Args: path: The relative path to the source document (e.g., "index.md", "how_to/customize.md") project: the name of the project (e.g., "panel", "panel-material-ui", "hvplot", "my-custom-project")

Returns:

Type	Description
The markdown content of the specified document.

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
async def get_document(path: str, project: str, ctx: Context) -> Document:
    """Retrieve a specific document by path and project.

    Use this tool to look up a specific document within a project.

    Args:
        path: The relative path to the source document (e.g., "index.md", "how_to/customize.md")
        project: the name of the project (e.g., "panel", "panel-material-ui", "hvplot", "my-custom-project")

    Returns
    -------
        The markdown content of the specified document.
    """
    indexer = get_indexer()
    return await indexer.get_document(path, project, ctx=ctx)

get_indexer()

Get or create the global DocumentationIndexer instance.

Source code in src/holoviz_mcp/holoviz_mcp/server.py

def get_indexer() -> DocumentationIndexer:
    """Get or create the global DocumentationIndexer instance."""
    global _indexer
    if _indexer is None:
        _indexer = DocumentationIndexer()
    return _indexer

get_reference_guide(component, project=None, content=True, ctx=None) async

Find reference guides for specific components in HoloViz or user-defined projects.

Reference guides are a subset of all documents that focus on specific UI components or plot types, such as:

• panel: "Button", "TextInput", ...
• hvplot: "bar", "scatter", ...
• my-custom-project: custom components from your organization

DO use this tool to easily find reference guides for specific components in HoloViz libraries and your custom projects.

Args: component (str): Name of the component (e.g., "Button", "TextInput", "bar", "scatter") project (str, optional): Project name. Defaults to None (searches all projects). Options: "panel", "panel-material-ui", "hvplot", "param", "holoviews" content (bool, optional): Whether to include full content. Defaults to True. Set to False to only return metadata for faster responses.

Returns:

Type	Description
list[Document]: A list of reference guides for the component with full content.

Examples:

>>> get_reference_guide("Button")  # Find Button component guide across all projects
>>> get_reference_guide("Button", "panel")  # Find Panel Button component guide specifically
>>> get_reference_guide("TextInput", "panel-material-ui")  # Find Material UI TextInput guide
>>> get_reference_guide("bar", "hvplot")  # Find hvplot bar chart reference
>>> get_reference_guide("scatter", "hvplot")  # Find hvplot scatter plot reference
>>> get_reference_guide("Audio", content=False)  # Don't include Markdown content for faster response

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
async def get_reference_guide(
    component: str,
    project: str | None = None,
    content: bool = True,
    ctx: Context | None = None,
) -> list[Document]:
    """Find reference guides for specific components in HoloViz or user-defined projects.

    Reference guides are a subset of all documents that focus on specific UI components
    or plot types, such as:

    - `panel`: "Button", "TextInput", ...
    - `hvplot`: "bar", "scatter", ...
    - `my-custom-project`: custom components from your organization

    DO use this tool to easily find reference guides for specific components in HoloViz libraries and your custom projects.

    Args:
        component (str): Name of the component (e.g., "Button", "TextInput", "bar", "scatter")
        project (str, optional): Project name. Defaults to None (searches all projects).
            Options: "panel", "panel-material-ui", "hvplot", "param", "holoviews"
        content (bool, optional): Whether to include full content. Defaults to True.
            Set to False to only return metadata for faster responses.

    Returns
    -------
        list[Document]: A list of reference guides for the component with full content.

    Examples
    --------
    >>> get_reference_guide("Button")  # Find Button component guide across all projects
    >>> get_reference_guide("Button", "panel")  # Find Panel Button component guide specifically
    >>> get_reference_guide("TextInput", "panel-material-ui")  # Find Material UI TextInput guide
    >>> get_reference_guide("bar", "hvplot")  # Find hvplot bar chart reference
    >>> get_reference_guide("scatter", "hvplot")  # Find hvplot scatter plot reference
    >>> get_reference_guide("Audio", content=False)  # Don't include Markdown content for faster response
    """
    indexer = get_indexer()
    return await indexer.search_get_reference_guide(component, project, content, ctx=ctx)

get_skill(name)

Get the specified skill for usage with LLMs.

Use list_skills tool to see available skills.

Args: name (str): The name of the skill to get. For example, "panel", "panel-material-ui", "panel-holoviews", "panel-custom-components" etc.

Returns:

Type	Description
str: A string containing the skill in Markdown format.

Examples:

>>> get_skill("holoviews")  # Best practices for using HoloViews
>>> get_skill("hvplot")  # Best practices for using hvPlot
>>> get_skill("panel-material-ui")  # Best practices for using Panel Material UI
>>> get_skill("panel")  # Best practices for using Panel

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
def get_skill(name: str) -> str:
    """Get the specified skill for usage with LLMs.

    Use list_skills tool to see available skills.

    Args:
        name (str): The name of the skill to get. For example, "panel", "panel-material-ui",
        "panel-holoviews", "panel-custom-components" etc.


    Returns
    -------
        str: A string containing the skill in Markdown format.

    Examples
    --------
    >>> get_skill("holoviews")  # Best practices for using HoloViews
    >>> get_skill("hvplot")  # Best practices for using hvPlot
    >>> get_skill("panel-material-ui")  # Best practices for using Panel Material UI
    >>> get_skill("panel")  # Best practices for using Panel
    """
    return _get_skill(name)

list_projects() async

List all HoloViz and user-defined projects with indexed documentation.

This includes both built-in HoloViz projects (panel, hvplot, etc.) and any custom/internal documentation projects you have configured.

Returns:

Type	Description
list[str]: A list of project names that have documentation available.	Names are returned in hyphenated format (e.g., "panel-material-ui", "my-custom-project").

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
async def list_projects() -> list[str]:
    """List all HoloViz and user-defined projects with indexed documentation.

    This includes both built-in HoloViz projects (panel, hvplot, etc.) and any custom/internal
    documentation projects you have configured.

    Returns
    -------
        list[str]: A list of project names that have documentation available.
                   Names are returned in hyphenated format (e.g., "panel-material-ui", "my-custom-project").
    """
    indexer = get_indexer()
    return await indexer.list_projects()

list_skills()

List all available skills.

Use get_skill tool to retrieve a specific skill.

Returns:

Type	Description
list[str]: A list of the skills available.	Names are returned in hyphenated format (e.g., "panel-material-ui", "panel-custom-components" and "a-custom-skill").

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
def list_skills() -> list[str]:
    """List all available skills.

    Use get_skill tool to retrieve a specific skill.

    Returns
    -------
        list[str]: A list of the skills available.
            Names are returned in hyphenated format (e.g., "panel-material-ui", "panel-custom-components" and "a-custom-skill").
    """
    return _list_skills()

search(query, project=None, content='truncated', max_results=2, max_content_chars=10000, ctx=None) async

Search the HoloViz and any user defined project documentation using semantic similarity.

IMPORTANT: This is a general purpose search tool. Not just for searching the HoloViz documentation.

DO use this tool to search the HoloViz project documentation DO use this tool to search any additional user-defined project documentation. DO use the holoviz_list_projects tool to list the available projects.

BEST PRACTICES: - For initial exploration, use content=False to get an overview of available documents - Use content="chunk" for quick snippets, content="full" for complete documents - Adjust max_content_chars if you need more or less content per result - Set max_content_chars=None to get untruncated content (use with caution for large docs)

QUERY OPTIMIZATION: The search uses context-aware truncation that centers returned content on query keywords. To get the most relevant excerpts:

• Use SPECIFIC terms: "CheckboxEditor SelectEditor" > "editor dropdown"
• Use UNIQUE identifiers: "background_gradient text_gradient" > "styling colors"
• Avoid COMMON terms that appear everywhere: "pandas", "import", "data", "widget"
• Include CLASS/FUNCTION names: "add_filter RangeSlider" > "filtering with widgets"
• Use MULTIPLE specific terms: Helps the algorithm find the right section
• Target FEATURE-SPECIFIC vocabulary: Terms unique to the feature you're looking for

Example: Instead of "how to add pagination to a table", use "pagination page_size local remote" This ensures the truncated content focuses on the pagination section, not generic table info.

Args: query (str): Search query using natural language or specific keywords. Natural language works for finding documents, but specific terms work better for content truncation. See QUERY OPTIMIZATION above.

    Good examples: "Button onClick on_click callback event", "hvPlot bar chart kind options"
    Okay examples: "how to style Material UI components", "interactive plotting with widgets"
project (str, optional): Optional project filter. Defaults to None.
    Examples: "panel", "hvplot", "my-custom-project""
content (str | bool, optional): Controls what content is returned. Defaults to "truncated".
    - "truncated": Full document content, smart-truncated around query keywords (default)
    - "chunk": Only the best-matching chunk from the document
    - "full": Full document content with no truncation (can be very large)
    - False: No content, metadata only (fastest)
    For backward compat, True maps to "truncated".
max_results (int, optional): Maximum number of results to return. Defaults to 2.
    Increase if you need more options, but be mindful of response size.
max_content_chars (int | None, optional): Maximum characters of content per result.
    Defaults to 10000. Set to None for untruncated content (may cause token limit errors).
    Content is truncated at word boundaries with an ellipsis indicator.

Returns:

Type	Description
list[Document]: A list of relevant documents ordered by relevance.

Examples:

>>> search("Button onClick on_click callback event", "panel-material-ui")  # Optimized: specific class and methods
>>> search("hvPlot bar chart kind colormap", "hvplot")  # Optimized: feature-specific terms
>>> search("FlexBox GridBox layout responsive sizing", "panel")  # Optimized: specific layout classes
>>> search("Tabulator pagination page_size local remote", "panel", max_results=3)  # Optimized with more results
>>> search("Param Parameter depends watch", content=False)  # Quick metadata search with specific terms
>>> search("stream follow rollover patch", max_content_chars=5000)  # Streaming-specific methods
>>> search("custom database connector SQL query", "my-custom-project")  # User project with specific terms
>>> search("Tabulator formatters", content="full")  # Full document content, no truncation
>>> search("Button widget", content="chunk")  # Only the best-matching chunk

Source code in src/holoviz_mcp/holoviz_mcp/server.py

@mcp.tool()
async def search(
    query: str,
    project: str | None = None,
    content: str | bool = "truncated",
    max_results: int = 2,
    max_content_chars: int | None = 10000,
    ctx: Context | None = None,
) -> list[Document]:
    """Search the HoloViz and any user defined project documentation using semantic similarity.

    IMPORTANT: This is a general purpose search tool. Not just for searching the HoloViz documentation.

    DO use this tool to search the HoloViz project documentation
    DO use this tool to search any additional user-defined project documentation.
    DO use the holoviz_list_projects tool to list the available projects.

    BEST PRACTICES:
    - For initial exploration, use content=False to get an overview of available documents
    - Use content="chunk" for quick snippets, content="full" for complete documents
    - Adjust max_content_chars if you need more or less content per result
    - Set max_content_chars=None to get untruncated content (use with caution for large docs)

    QUERY OPTIMIZATION:
    The search uses context-aware truncation that centers returned content on query keywords.
    To get the most relevant excerpts:

    - Use SPECIFIC terms: "CheckboxEditor SelectEditor" > "editor dropdown"
    - Use UNIQUE identifiers: "background_gradient text_gradient" > "styling colors"
    - Avoid COMMON terms that appear everywhere: "pandas", "import", "data", "widget"
    - Include CLASS/FUNCTION names: "add_filter RangeSlider" > "filtering with widgets"
    - Use MULTIPLE specific terms: Helps the algorithm find the right section
    - Target FEATURE-SPECIFIC vocabulary: Terms unique to the feature you're looking for

    Example: Instead of "how to add pagination to a table", use "pagination page_size local remote"
    This ensures the truncated content focuses on the pagination section, not generic table info.

    Args:
        query (str): Search query using natural language or specific keywords.
            Natural language works for finding documents, but specific terms work better
            for content truncation. See QUERY OPTIMIZATION above.

            Good examples: "Button onClick on_click callback event", "hvPlot bar chart kind options"
            Okay examples: "how to style Material UI components", "interactive plotting with widgets"
        project (str, optional): Optional project filter. Defaults to None.
            Examples: "panel", "hvplot", "my-custom-project""
        content (str | bool, optional): Controls what content is returned. Defaults to "truncated".
            - "truncated": Full document content, smart-truncated around query keywords (default)
            - "chunk": Only the best-matching chunk from the document
            - "full": Full document content with no truncation (can be very large)
            - False: No content, metadata only (fastest)
            For backward compat, True maps to "truncated".
        max_results (int, optional): Maximum number of results to return. Defaults to 2.
            Increase if you need more options, but be mindful of response size.
        max_content_chars (int | None, optional): Maximum characters of content per result.
            Defaults to 10000. Set to None for untruncated content (may cause token limit errors).
            Content is truncated at word boundaries with an ellipsis indicator.

    Returns
    -------
        list[Document]: A list of relevant documents ordered by relevance.

    Examples
    --------
    >>> search("Button onClick on_click callback event", "panel-material-ui")  # Optimized: specific class and methods
    >>> search("hvPlot bar chart kind colormap", "hvplot")  # Optimized: feature-specific terms
    >>> search("FlexBox GridBox layout responsive sizing", "panel")  # Optimized: specific layout classes
    >>> search("Tabulator pagination page_size local remote", "panel", max_results=3)  # Optimized with more results
    >>> search("Param Parameter depends watch", content=False)  # Quick metadata search with specific terms
    >>> search("stream follow rollover patch", max_content_chars=5000)  # Streaming-specific methods
    >>> search("custom database connector SQL query", "my-custom-project")  # User project with specific terms
    >>> search("Tabulator formatters", content="full")  # Full document content, no truncation
    >>> search("Button widget", content="chunk")  # Only the best-matching chunk
    """
    indexer = get_indexer()
    return await indexer.search(query, project, content, max_results, max_content_chars, ctx=ctx)

update_index(ctx) async

Update the documentation index by re-cloning repositories and re-indexing content.

DO use this tool periodically (weekly) to ensure the documentation index is up-to-date with the latest changes in the HoloViz ecosystem.

Warning: This operation can take a long time (up to 5 minutes) depending on the number of repositories and their size!

Returns:

Type	Description
str: Status message indicating the result of the update operation.

Examples:

>>> update_index()  # Updates all documentation repositories and rebuilds index

Source code in src/holoviz_mcp/holoviz_mcp/server.py

async def update_index(ctx: Context) -> str:
    """Update the documentation index by re-cloning repositories and re-indexing content.

    DO use this tool periodically (weekly) to ensure the documentation index is up-to-date
    with the latest changes in the HoloViz ecosystem.

    Warning: This operation can take a long time (up to 5 minutes) depending on the number of
    repositories and their size!

    Returns
    -------
        str: Status message indicating the result of the update operation.

    Examples
    --------
    >>> update_index()  # Updates all documentation repositories and rebuilds index
    """
    try:
        indexer = get_indexer()

        # Use True as ctx to enable print statements for user feedback
        await indexer.index_documentation(ctx=ctx)

        return "Documentation index updated successfully."
    except Exception as e:
        logger.error(f"Failed to update documentation index: {e}")
        error_msg = f"Failed to update documentation index: {str(e)}"
        return error_msg

Models

Data models for the HoloViz Documentation MCP server.

Document

Bases: BaseModel

Represents a document.

Source code in src/holoviz_mcp/holoviz_mcp/models.py

class Document(BaseModel):
    """Represents a document."""

    title: str = Field(..., description="The title of the document.")
    url: HttpUrl = Field(..., description="The URL of the rendered, target document.")
    project: str = Field(..., description="The project to which the document belongs.")
    source_path: str = Field(..., description="The path to the document within the project.")
    source_url: HttpUrl = Field(..., description="The URL to the source document.")
    is_reference: bool = Field(..., description="Indicates if the document is a reference guide.")
    description: Optional[str] = Field(default=None, description="A brief description of the document.")
    content: Optional[str] = Field(default=None, description="The content of the documentation, if available. In Markdown format if possible.")
    relevance_score: Optional[float] = Field(default=None, description="Relevance score of the document, where 1 is the highest score indicating an exact match.")

content = Field(default=None, description='The content of the documentation, if available. In Markdown format if possible.') class-attribute instance-attribute

description = Field(default=None, description='A brief description of the document.') class-attribute instance-attribute

is_reference = Field(..., description='Indicates if the document is a reference guide.') class-attribute instance-attribute

project = Field(..., description='The project to which the document belongs.') class-attribute instance-attribute

relevance_score = Field(default=None, description='Relevance score of the document, where 1 is the highest score indicating an exact match.') class-attribute instance-attribute

source_path = Field(..., description='The path to the document within the project.') class-attribute instance-attribute

source_url = Field(..., description='The URL to the source document.') class-attribute instance-attribute

title = Field(..., description='The title of the document.') class-attribute instance-attribute

url = Field(..., description='The URL of the rendered, target document.') class-attribute instance-attribute

Data

Data handling for the HoloViz Documentation MCP server.

logger = logging.getLogger(__name__) module-attribute

DocumentationIndexer

Handles cloning, processing, and indexing of documentation.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

class DocumentationIndexer:
    """Handles cloning, processing, and indexing of documentation."""

    def __init__(self, *, data_dir: Optional[Path] = None, repos_dir: Optional[Path] = None, vector_dir: Optional[Path] = None):
        """Initialize the DocumentationIndexer.

        Args:
            data_dir: Directory to store index data. Defaults to user config directory.
            repos_dir: Directory to store cloned repositories. Defaults to HOLOVIZ_MCP_REPOS_DIR.
            vector_dir: Directory to store vector database. Defaults to config.vector_dir
        """
        # Use unified config for default paths
        config = self._holoviz_mcp_config = get_config()

        self.data_dir = data_dir or config.user_dir
        self.data_dir.mkdir(parents=True, exist_ok=True)

        # Use configurable repos directory for repository downloads
        self.repos_dir = repos_dir or config.repos_dir
        self.repos_dir.mkdir(parents=True, exist_ok=True)

        # Use configurable directory for vector database path
        self._vector_db_path = vector_dir or config.server.vector_db_path
        self._vector_db_path.parent.mkdir(parents=True, exist_ok=True)

        # Disable ChromaDB telemetry based on config
        if not config.server.anonymized_telemetry:
            os.environ["ANONYMIZED_TELEMETRY"] = "False"

        # Initialize ChromaDB with health check
        try:
            self.chroma_client = chromadb.PersistentClient(path=str(self._vector_db_path))
            self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
            self.collection.count()  # Probe read to verify health
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException as e:
            logger.error("ChromaDB initialization failed (%s: %s). Wiping and reinitializing.", type(e).__name__, e)
            shutil.rmtree(self._vector_db_path, ignore_errors=True)
            self._vector_db_path.mkdir(parents=True, exist_ok=True)
            SharedSystemClient.clear_system_cache()
            self.chroma_client = chromadb.PersistentClient(path=str(self._vector_db_path))
            self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)

        # Lazy-initialized async lock for database operations to prevent corruption from concurrent access
        self._db_lock: Optional[asyncio.Lock] = None

        # Thread-local storage for MarkdownExporter (not thread-safe due to Jinja2)
        self._thread_local = threading.local()

        # Load documentation config from the centralized config system
        self.config = get_config().docs

        # Wrap index_documentation to ensure all calls use the async DB lock
        # This prevents race conditions when indexing is called directly (e.g., from
        # update_index tool or run method) and concurrently with search operations
        if not hasattr(self, "_index_documentation_wrapped"):
            original_index_documentation = self.index_documentation

            async def _locked_index_documentation(*args: Any, **kwargs: Any) -> Any:
                async with self.db_lock:
                    return await original_index_documentation(*args, **kwargs)

            self.index_documentation = _locked_index_documentation  # type: ignore[method-assign]
            self._index_documentation_wrapped = True

    def _get_nb_exporter(self) -> MarkdownExporter:
        """Get or create a thread-local MarkdownExporter instance.

        MarkdownExporter uses Jinja2 templates internally which are not
        thread-safe, so each thread gets its own instance.
        """
        if not hasattr(self._thread_local, "nb_exporter"):
            self._thread_local.nb_exporter = MarkdownExporter()
        return self._thread_local.nb_exporter

    @property
    def db_lock(self) -> asyncio.Lock:
        """Lazy-initialize and return the database lock.

        This ensures the lock is created in the correct event loop context.
        """
        if self._db_lock is None:
            self._db_lock = asyncio.Lock()
        return self._db_lock

    @property
    def _backup_path(self) -> Path:
        """Path to the backup copy of the vector database directory."""
        return Path(str(self._vector_db_path) + ".bak")

    @property
    def _hash_file_path(self) -> Path:
        """Path to the hash sidecar file for incremental indexing."""
        return self._vector_db_path.parent / "index_hashes.json"

    async def _restore_from_backup(self, ctx: Context | None = None) -> None:
        """Restore vector database from backup after a write failure."""
        backup_path = self._backup_path
        if not backup_path.exists():
            await log_warning("No backup found to restore from.", ctx)
            return
        try:
            shutil.rmtree(self._vector_db_path, ignore_errors=True)
            shutil.copytree(backup_path, self._vector_db_path)
            SharedSystemClient.clear_system_cache()
            self.chroma_client = chromadb.PersistentClient(path=str(self._vector_db_path))
            self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
            # Restore hash file if backup exists
            hash_bak = backup_path.parent / "index_hashes.json.bak"
            if hash_bak.exists():
                shutil.copy2(hash_bak, self._hash_file_path)
            await log_info("Restored vector database from backup.", ctx)
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException as e:
            logger.error("Failed to restore from backup (%s: %s). Database may be degraded.", type(e).__name__, e)

    def _load_hashes(self) -> dict[str, str]:
        """Load document hashes from the sidecar file.

        Returns
        -------
        dict[str, str]
            Mapping of doc_id -> SHA-256 hex digest. Empty dict if file
            doesn't exist or is corrupt.
        """
        path = self._hash_file_path
        if not path.exists():
            return {}
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)
            if isinstance(data, dict):
                return data
            return {}
        except (json.JSONDecodeError, OSError):
            logger.warning("Failed to load hash file %s, starting fresh", path)
            return {}

    def _save_hashes(self, hashes: dict[str, str]) -> None:
        """Save document hashes to the sidecar file."""
        path = self._hash_file_path
        with open(path, "w", encoding="utf-8") as f:
            json.dump(hashes, f, indent=2)

    @staticmethod
    def _compute_file_hash(file_path: Path) -> str:
        """Compute SHA-256 hash of a file's contents."""
        return hashlib.sha256(file_path.read_bytes()).hexdigest()

    def _delete_doc_chunks(self, doc_id: str) -> int:
        """Delete all chunks for a document from ChromaDB.

        Parameters
        ----------
        doc_id : str
            The parent document ID (chunks have parent_id metadata matching this).

        Returns
        -------
        int
            Number of chunks deleted.
        """
        results = self.collection.get(
            where={"parent_id": doc_id},
            include=[],  # only need IDs
        )
        if results["ids"]:
            self.collection.delete(ids=results["ids"])
        return len(results["ids"]) if results["ids"] else 0

    def is_indexed(self) -> bool:
        """Check if documentation index exists and is valid."""
        try:
            count = self.collection.count()
            return count > 0
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException:
            return False

    async def ensure_indexed(self, ctx: Context | None = None):
        """Ensure documentation is indexed, creating if necessary."""
        if not self.is_indexed():
            await log_info("Documentation index not found. Creating initial index...", ctx)
            await self.index_documentation()

    async def clone_or_update_repo(self, repo_name: str, repo_config: "GitRepository", ctx: Context | None = None) -> Optional[Path]:
        """Clone or update a single repository.

        Delegates to the synchronous implementation via run_in_executor.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self._clone_or_update_repo_sync, repo_name, repo_config)

    def _clone_or_update_repo_sync(self, repo_name: str, repo_config: "GitRepository") -> Optional[Path]:
        """Clone or update a single repository (synchronous, for thread pool).

        Same logic as clone_or_update_repo but uses logger instead of async ctx.
        """
        repo_path = self.repos_dir / repo_name

        try:
            if repo_path.exists():
                repo = git.Repo(repo_path)

                if repo_config.tag:
                    # Tag checkouts leave the repo in detached HEAD; verify via tag→commit mapping.
                    matching_tag = next((t for t in repo.tags if t.name == repo_config.tag), None)
                    if matching_tag and matching_tag.commit == repo.head.commit:
                        # Already on the correct tag — tags are immutable, nothing to pull.
                        return repo_path
                    logger.info(f"Stale checkout for {repo_name}: expected tag '{repo_config.tag}'. Deleting and re-cloning...")
                    shutil.rmtree(repo_path)
                    # Fall through to clone below
                elif repo_config.branch:
                    try:
                        current_branch = repo.active_branch.name
                    except TypeError:
                        current_branch = None  # Detached HEAD

                    if current_branch != repo_config.branch:
                        logger.info(f"Stale checkout for {repo_name}: on '{current_branch}', expected '{repo_config.branch}'. Deleting and re-cloning...")
                        shutil.rmtree(repo_path)
                        # Fall through to clone below
                    else:
                        logger.info(f"Updating {repo_name} repository at {repo_path}...")
                        repo.remotes.origin.pull()
                        return repo_path
                else:
                    logger.info(f"Updating {repo_name} repository at {repo_path}...")
                    repo.remotes.origin.pull()
                    return repo_path

            # Repo does not exist (fresh or was just deleted above)
            logger.info(f"Cloning {repo_name} repository to {repo_path}...")
            clone_kwargs: dict[str, Any] = {"depth": 1}

            if repo_config.branch:
                clone_kwargs["branch"] = repo_config.branch
            elif repo_config.tag:
                clone_kwargs["branch"] = repo_config.tag
            elif repo_config.commit:
                git.Repo.clone_from(str(repo_config.url), repo_path, **clone_kwargs)
                repo = git.Repo(repo_path)
                repo.git.checkout(repo_config.commit)
                return repo_path

            git.Repo.clone_from(str(repo_config.url), repo_path, **clone_kwargs)
            return repo_path
        except Exception as e:
            logger.warning(f"Failed to clone/update {repo_name}: {e}")
            return None

    def _extract_docs_from_repo_sync(self, repo_path: Path, project: str, old_hashes: dict[str, str] | None = None) -> dict[str, Any]:
        """Extract documentation files from a repository (synchronous, for thread pool).

        When old_hashes is provided, files whose hash matches are skipped (not
        read or converted), making incremental runs much faster.

        Returns
        -------
        dict
            ``{"docs": [...], "skipped_hashes": {doc_id: hash, ...}}``
            where ``docs`` contains only new/changed documents and
            ``skipped_hashes`` maps unchanged doc_ids to their hashes.
        """
        docs = []
        skipped_hashes: dict[str, str] = {}
        repo_config = self.config.repositories[project]

        if isinstance(repo_config.folders, dict):
            folders = repo_config.folders
        else:
            folders = {name: FolderConfig() for name in repo_config.folders}

        files: set = set()
        logger.info(f"Processing {project} documentation files in {','.join(folders.keys())}")

        for folder_name in folders.keys():
            docs_folder: Path = repo_path / folder_name
            if docs_folder.exists():
                for pattern in self.config.index_patterns:
                    files.update(docs_folder.glob(pattern))

        skipped_count = 0
        for file in files:
            if file.exists() and not file.is_dir():
                # Early hash check: skip unchanged files before expensive content extraction
                if old_hashes:
                    relative_path = file.relative_to(repo_path)
                    doc_id = self._generate_doc_id(project, relative_path)
                    file_hash = self._compute_file_hash(file)
                    if old_hashes.get(doc_id) == file_hash:
                        skipped_hashes[doc_id] = file_hash
                        skipped_count += 1
                        continue

                folder_name = ""
                for fname in folders.keys():
                    folder_path = repo_path / fname
                    try:
                        file.relative_to(folder_path)
                        folder_name = fname
                        break
                    except ValueError:
                        continue

                doc_data = self.process_file(file, project, repo_config, folder_name)
                if doc_data:
                    docs.append(doc_data)

        reference_count = sum(1 for doc in docs if doc["is_reference"])
        regular_count = len(docs) - reference_count
        if skipped_count:
            logger.info(f"  {project}: {len(docs)} changed + {skipped_count} unchanged ({regular_count} regular, {reference_count} reference guides)")
        else:
            logger.info(f"  {project}: {len(docs)} total documents ({regular_count} regular, {reference_count} reference guides)")
        return {"docs": docs, "skipped_hashes": skipped_hashes}

    def _clone_and_extract_repo_sync(self, repo_name: str, repo_config: "GitRepository", old_hashes: dict[str, str] | None = None) -> dict[str, Any]:
        """Clone/update a repo and extract its documents (single unit of work for thread pool)."""
        t0 = time.monotonic()
        repo_path = self._clone_or_update_repo_sync(repo_name, repo_config)
        if not repo_path:
            return {"docs": [], "skipped_hashes": {}}
        result = self._extract_docs_from_repo_sync(repo_path, repo_name, old_hashes)
        logger.info(f"Completed {repo_name}: {len(result['docs'])} docs in {time.monotonic() - t0:.1f}s")
        return result

    def _is_reference_document(self, file_path: Path, project: str, folder_name: str = "") -> bool:
        """Check if the document is a reference document using configurable patterns.

        Args:
            file_path: Full path to the file
            project: Project name
            folder_name: Name of the folder this file belongs to

        Returns
        -------
            bool: True if this is a reference document
        """
        repo_config = self.config.repositories[project]
        repo_path = self.repos_dir / project

        try:
            relative_path = file_path.relative_to(repo_path)

            # Check against configured reference patterns
            for pattern in repo_config.reference_patterns:
                if relative_path.match(pattern):
                    return True

            # Fallback to simple "reference" in path check
            return "reference" in relative_path.parts
        except (ValueError, KeyError):
            # If we can't determine relative path or no patterns configured, use simple fallback
            return "reference" in file_path.parts

    def _generate_doc_id(self, project: str, path: Path) -> str:
        """Generate a unique document ID from project and path."""
        readable_path = str(path).replace("/", "___").replace(".", "_")
        readable_id = f"{project}___{readable_path}"

        return readable_id

    def _generate_doc_url(self, project: str, path: Path, folder_name: str = "") -> str:
        """Generate documentation URL for a file.

        This method creates the final URL where the documentation can be accessed online.
        It handles folder URL mapping to ensure proper URL structure for different documentation layouts.

        Args:
            project: Name of the project/repository (e.g., "panel", "hvplot")
            path: Relative path to the file within the repository
            folder_name: Name of the folder containing the file (e.g., "examples/reference", "doc")
                       Used for URL path mapping when folders have custom URL structures

        Returns
        -------
            Complete URL to the documentation file

        Examples
        --------
            For Panel reference guides:
            - Input: project="panel", path="examples/reference/widgets/Button.ipynb", folder_name="examples/reference"
            - Output: "https://panel.holoviz.org/reference/widgets/Button.html"

            For regular documentation:
            - Input: project="panel", path="doc/getting_started.md", folder_name="doc"
            - Output: "https://panel.holoviz.org/getting_started.html"
        """
        repo_config = self.config.repositories[project]
        base_url = str(repo_config.base_url).rstrip("/")

        # Get the URL path mapping for this folder
        folder_url_path = repo_config.get_folder_url_path(folder_name)

        # If there's a folder URL mapping, we need to adjust the path
        if folder_url_path and folder_name:
            # Remove the folder name from the beginning of the path
            path_str = str(path)

            # Check if path starts with the folder name
            if path_str.startswith(folder_name + "/"):
                # Remove the folder prefix and leading slash
                remaining_path = path_str[len(folder_name) + 1 :]
                adjusted_path = Path(remaining_path) if remaining_path else Path(".")
            elif path_str == folder_name:
                # The path is exactly the folder name
                adjusted_path = Path(".")
            else:
                # Fallback: try to remove folder parts from the beginning
                path_parts = list(path.parts)
                folder_parts = folder_name.split("/")
                for folder_part in folder_parts:
                    if path_parts and path_parts[0] == folder_part:
                        path_parts = path_parts[1:]
                adjusted_path = Path(*path_parts) if path_parts else Path(".")

            # Don't remove first part since we already adjusted the path
            doc_path = convert_path_to_url(adjusted_path, remove_first_part=False, url_transform=repo_config.url_transform)
        else:
            # Convert file path to URL format normally (remove first part for legacy compatibility)
            doc_path = convert_path_to_url(path, remove_first_part=True, url_transform=repo_config.url_transform)

        # Combine base URL, folder URL path, and document path
        if folder_url_path:
            full_url = f"{base_url}{folder_url_path}/{doc_path}"
        else:
            full_url = f"{base_url}/{doc_path}"

        return full_url.replace("//", "/").replace(":/", "://")  # Fix double slashes

    @staticmethod
    def _to_title(fallback_filename: str = "") -> str:
        """Extract title from a filename or return a default title."""
        title = Path(fallback_filename).stem
        if "_" in title and title.split("_")[0].isdigit():
            title = title.split("_", 1)[-1]
        title = title.replace("_", " ").replace("-", " ").title()
        return title

    @classmethod
    def _extract_title_from_markdown(cls, content: str, fallback_filename: str = "") -> str:
        """Extract title from markdown content, with filename fallback."""
        lines = content.split("\n")
        for line in lines:
            line = line.strip()
            if line.startswith("# "):
                # Return just the title text without the "# " prefix
                return line[2:].strip()
            if line.startswith("##"):
                break

        if fallback_filename:
            return cls._to_title(fallback_filename)

        return "No Title"

    @staticmethod
    def _extract_description_from_markdown(content: str, max_length=200) -> str:
        """Extract description from markdown content."""
        content = content.strip()

        # Plotly documents start with --- ... --- section. Skip the section
        if content.startswith("---"):
            content = content.split("---", 2)[-1].strip()

        lines = content.split("\n")
        clean_lines = []
        in_code_block = False

        for line in lines:
            if line.strip().startswith("```"):
                in_code_block = not in_code_block
                continue

            if in_code_block or line.startswith(("#", "    ", "\t", "---", "___")):
                continue

            clean_lines.append(line)

        # Join lines and clean up
        clean_content = "\n".join(clean_lines).strip()

        # Remove extra whitespace and limit length
        clean_content = " ".join(clean_content.split())

        if len(clean_content) > max_length:
            clean_content = clean_content[:max_length].rsplit(" ", 1)[0]
        if not clean_content.endswith("."):
            clean_content += " ..."

        return clean_content

    def convert_notebook_to_markdown(self, notebook_path: Path) -> str:
        """Convert a Jupyter notebook to markdown."""
        try:
            with open(notebook_path, "r", encoding="utf-8") as f:
                notebook = nbread(f, as_version=4)

            (body, resources) = self._get_nb_exporter().from_notebook_node(notebook)
            return body
        except Exception as e:
            logger.error(f"Failed to convert notebook {notebook_path}: {e}")
            return str(e)

    @staticmethod
    def _to_source_url(file_path: Path, repo_config: GitRepository, raw: bool = False) -> str:
        """Generate source URL for a file based on repository configuration."""
        url = str(repo_config.url)
        branch = repo_config.branch or "main"
        if url.startswith("https://github.com") and url.endswith(".git"):
            url = url.replace("https://github.com/", "").replace(".git", "")
            project, repository = url.split("/")
            if raw:
                return f"https://raw.githubusercontent.com/{project}/{repository}/refs/heads/{branch}/{file_path}"

            return f"https://github.com/{project}/{repository}/blob/{branch}/{file_path}"
        if "dev.azure.com" in url:
            organisation = url.split("/")[3].split("@")[0]
            project = url.split("/")[-3]
            repo_name = url.split("/")[-1]
            if raw:
                return f"https://dev.azure.com/{organisation}/{project}/_apis/sourceProviders/TfsGit/filecontents?repository={repo_name}&path=/{file_path}&commitOrBranch={branch}&api-version=7.0"

            return f"https://dev.azure.com/{organisation}/{project}/_git/{repo_name}?path=/{file_path}&version=GB{branch}"

        raise ValueError(f"Unsupported repository URL format: {url}. Please provide a valid GitHub or Azure DevOps URL.")

    def process_file(self, file_path: Path, project: str, repo_config: GitRepository, folder_name: str = "") -> Optional[dict[str, Any]]:
        """Process a file and extract metadata."""
        try:
            if file_path.suffix == ".ipynb":
                content = self.convert_notebook_to_markdown(file_path)
            elif file_path.suffix in [".md", ".rst", ".txt"]:
                with open(file_path, "r", encoding="utf-8") as f:
                    content = f.read()
            else:
                logger.debug(f"Skipping unsupported file type: {file_path}")
                return None

            title = self._extract_title_from_markdown(content, file_path.name)
            if not title:
                title = file_path.stem.replace("_", " ").title()

            description = self._extract_description_from_markdown(content)

            repo_path = self.repos_dir / project
            relative_path = file_path.relative_to(repo_path)

            doc_id = self._generate_doc_id(project, relative_path)

            is_reference = self._is_reference_document(file_path, project, folder_name)

            source_url = self._to_source_url(relative_path, repo_config)

            return {
                "id": doc_id,
                "title": title,
                "url": self._generate_doc_url(project, relative_path, folder_name),
                "project": project,
                "source_path": str(relative_path),
                "source_path_stem": file_path.stem,
                "source_url": source_url,
                "description": description,
                "content": content,
                "is_reference": is_reference,
                "file_hash": self._compute_file_hash(file_path),
            }
        except Exception as e:
            logger.error(f"Failed to process file {file_path}: {e}")
            return None

    async def extract_docs_from_repo(self, repo_path: Path, project: str, ctx: Context | None = None) -> dict[str, Any]:
        """Extract documentation files from a repository.

        Delegates to the synchronous implementation via run_in_executor.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self._extract_docs_from_repo_sync, repo_path, project)

    async def index_documentation(
        self,
        ctx: Context | None = None,
        projects: list[str] | None = None,
        full_rebuild: bool = False,
    ) -> None:
        """Index documentation, incrementally when possible.

        Parameters
        ----------
        ctx : Context | None
            FastMCP context for logging.
        projects : list[str] | None
            Only process these projects. None means all.
        full_rebuild : bool
            Force full rebuild, ignoring cached hashes.
        """
        overall_t0 = time.monotonic()
        await log_info("Starting documentation indexing...", ctx)

        # Validate project names
        if projects:
            available = set(self.config.repositories.keys())
            invalid = [p for p in projects if p not in available]
            if invalid:
                raise ValueError(f"Unknown project(s): {', '.join(invalid)}. Available: {', '.join(sorted(available))}")

        # Determine target repos
        if projects:
            target_repos = {name: cfg for name, cfg in self.config.repositories.items() if name in projects}
        else:
            target_repos = dict(self.config.repositories.items())

        # Load existing hashes BEFORE extraction so workers can skip unchanged files
        old_hashes = self._load_hashes()
        is_full_rebuild = full_rebuild or not old_hashes
        worker_hashes = {} if is_full_rebuild else old_hashes

        all_docs: list[dict[str, Any]] = []
        all_skipped_hashes: dict[str, str] = {}

        # Clone/update repositories and extract documentation in parallel
        # Workers receive old_hashes so they can skip unchanged files early
        # (avoiding expensive notebook conversion for files whose hash matches)
        num_repos = len(target_repos)
        if num_repos > 0:
            max_workers = min(4, num_repos)
            await log_info(f"Processing {num_repos} repositories with {max_workers} workers...", ctx)

            loop = asyncio.get_running_loop()
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = {
                    loop.run_in_executor(executor, self._clone_and_extract_repo_sync, repo_name, repo_config, worker_hashes): repo_name
                    for repo_name, repo_config in target_repos.items()
                }
                gather_results = await asyncio.gather(*futures.keys(), return_exceptions=True)
                for result, repo_name in zip(gather_results, futures.values(), strict=False):
                    if isinstance(result, BaseException):
                        await log_warning(f"Repository {repo_name} failed: {result}", ctx)
                    elif isinstance(result, dict):
                        all_docs.extend(result["docs"])
                        all_skipped_hashes.update(result["skipped_hashes"])

        clone_extract_elapsed = time.monotonic() - overall_t0
        await log_info(
            f"Clone + extract completed in {clone_extract_elapsed:.1f}s ({len(all_docs)} to index, {len(all_skipped_hashes)} unchanged)",
            ctx,
        )

        # Detect deleted docs: in old hashes for target projects but not extracted or skipped
        target_project_names = set(target_repos.keys())
        all_known_ids = {doc["id"] for doc in all_docs} | set(all_skipped_hashes.keys())
        deleted_doc_ids: list[str] = [
            doc_id for doc_id in old_hashes if doc_id not in all_known_ids and any(doc_id.startswith(f"{proj}___") for proj in target_project_names)
        ]

        changed_docs: list[dict[str, Any]] = []

        if is_full_rebuild:
            if not all_docs:
                await log_warning("No documentation found to index", ctx)
                return
            docs_to_index = all_docs
            await log_info(f"Full rebuild: indexing all {len(docs_to_index)} documents", ctx)
        else:
            # Incremental: all_docs contains only new/changed files (skipped were filtered by workers)
            new_docs = [doc for doc in all_docs if doc["id"] not in old_hashes]
            changed_docs = [doc for doc in all_docs if doc["id"] in old_hashes]

            docs_to_index = all_docs  # all are new or changed (unchanged were skipped by workers)
            await log_info(
                f"Incremental: {len(new_docs)} new, {len(changed_docs)} changed, {len(all_skipped_hashes)} unchanged, {len(deleted_doc_ids)} deleted",
                ctx,
            )

            if not docs_to_index and not deleted_doc_ids:
                total_elapsed = time.monotonic() - overall_t0
                await log_info(f"Index up to date, nothing to do ({total_elapsed:.1f}s)", ctx)
                # Still save hashes (preserves skipped + old non-target hashes)
                new_hashes = dict(old_hashes)
                for doc_id in deleted_doc_ids:
                    new_hashes.pop(doc_id, None)
                self._save_hashes(new_hashes)
                return

        if docs_to_index:
            # Validate for duplicate IDs and log details
            await self._validate_unique_ids(docs_to_index)

        # Split documents into chunks at H1/H2 headers for better embedding quality
        all_chunks: list[dict[str, Any]] = []
        for doc in docs_to_index:
            all_chunks.extend(chunk_document(doc))
        await log_info(f"Chunked {len(docs_to_index)} documents into {len(all_chunks)} chunks", ctx)

        # Create pre-write backup of vector database and hash file
        backup_path = self._backup_path
        try:
            if self._vector_db_path.exists():
                t0 = time.monotonic()
                shutil.copytree(self._vector_db_path, backup_path, dirs_exist_ok=True)
                await log_info(f"Pre-write backup created at {backup_path} ({time.monotonic() - t0:.1f}s)", ctx)
            hash_src = self._hash_file_path
            if hash_src.exists():
                shutil.copy2(hash_src, backup_path.parent / "index_hashes.json.bak")
        except Exception as e:
            await log_warning(f"Failed to create pre-write backup: {e}. Continuing without backup.", ctx)

        if is_full_rebuild:
            # Full rebuild: clear existing collection and re-add everything
            await log_info("Clearing existing index...", ctx)
            try:
                count = self.collection.count()
                if count > 0:
                    results = self.collection.get()
                    if results["ids"]:
                        delete_batch_size = self.chroma_client.get_max_batch_size()
                        for i in range(0, len(results["ids"]), delete_batch_size):
                            self.collection.delete(ids=results["ids"][i : i + delete_batch_size])
            except Exception as e:
                logger.warning(f"Failed to clear existing collection: {e}")
                try:
                    self.chroma_client.delete_collection("holoviz_docs")
                    self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
                except Exception as e2:
                    await log_exception(f"Failed to recreate collection: {e2}", ctx)
                    raise
        else:
            # Incremental: delete chunks for changed and deleted docs
            for doc_id in deleted_doc_ids:
                deleted_count = self._delete_doc_chunks(doc_id)
                if deleted_count > 0:
                    logger.debug("Deleted %d chunks for removed doc %s", deleted_count, doc_id)
            for doc in changed_docs:
                deleted_count = self._delete_doc_chunks(doc["id"])
                if deleted_count > 0:
                    logger.debug("Deleted %d old chunks for changed doc %s", deleted_count, doc["id"])

        # Add chunks to ChromaDB in batches (ChromaDB enforces a max batch size)
        if all_chunks:
            batch_size = self.chroma_client.get_max_batch_size()
            await log_info(f"Adding {len(all_chunks)} chunks from {len(docs_to_index)} documents to index (batch size {batch_size})...", ctx)

            try:
                for batch_start in range(0, len(all_chunks), batch_size):
                    batch = all_chunks[batch_start : batch_start + batch_size]
                    self.collection.add(
                        documents=[doc["content"] for doc in batch],
                        metadatas=[
                            {
                                "title": doc["title"],
                                "url": doc["url"],
                                "project": doc["project"],
                                "source_path": doc["source_path"],
                                "source_path_stem": doc["source_path_stem"],
                                "source_url": doc["source_url"],
                                "description": doc["description"],
                                "is_reference": doc["is_reference"],
                                "chunk_index": doc["chunk_index"],
                                "parent_id": doc["parent_id"],
                            }
                            for doc in batch
                        ],
                        ids=[doc["id"] for doc in batch],
                    )
            except (KeyboardInterrupt, SystemExit):
                raise
            except BaseException as e:
                logger.error("ChromaDB write failed (%s: %s). Attempting restore from backup.", type(e).__name__, e)
                await self._restore_from_backup(ctx)
                raise

        # Save updated hashes (after successful ChromaDB write)
        new_hashes = dict(old_hashes)  # start from existing (preserves non-target projects)
        # Remove deleted docs
        for doc_id in deleted_doc_ids:
            new_hashes.pop(doc_id, None)
        # Add/update hashes for indexed docs
        for doc in docs_to_index:
            file_hash = doc.get("file_hash")
            if file_hash:
                new_hashes[doc["id"]] = file_hash
        # Include hashes for files that were skipped (unchanged)
        new_hashes.update(all_skipped_hashes)
        self._save_hashes(new_hashes)

        total_elapsed = time.monotonic() - overall_t0
        await log_info(f"Successfully indexed {len(all_chunks)} chunks from {len(docs_to_index)} documents in {total_elapsed:.1f}s", ctx)
        await log_info(f"Vector database stored at: {self._vector_db_path}", ctx)
        await log_info(f"Index contains {self.collection.count()} total documents", ctx)

        # Show detailed summary table
        await self._log_summary_table(ctx)

    async def _validate_unique_ids(self, all_docs: list[dict[str, Any]], ctx: Context | None = None) -> None:
        """Validate that all document IDs are unique and log duplicates."""
        seen_ids: dict = {}
        duplicates = []

        for doc in all_docs:
            doc_id = doc["id"]
            if doc_id in seen_ids:
                duplicates.append(
                    {
                        "id": doc_id,
                        "first_doc": seen_ids[doc_id],
                        "duplicate_doc": {"project": doc["project"], "source_path": doc["source_path"], "title": doc["title"]},
                    }
                )

                await log_warning(f"DUPLICATE ID FOUND: {doc_id}", ctx)
                await log_warning(f"  First document: {seen_ids[doc_id]['project']}/{seen_ids[doc_id]['source_path']} - {seen_ids[doc_id]['title']}", ctx)
                await log_warning(f"  Duplicate document: {doc['project']}/{doc['source_path']} - {doc['title']}", ctx)
            else:
                seen_ids[doc_id] = {"project": doc["project"], "source_path": doc["source_path"], "title": doc["title"]}

        if duplicates:
            error_msg = f"Found {len(duplicates)} duplicate document IDs"
            await log_exception(error_msg, ctx)

            # Log all duplicates for debugging
            for dup in duplicates:
                await log_exception(
                    f"Duplicate ID '{dup['id']}': {dup['first_doc']['project']}/{dup['first_doc']['source_path']} vs {dup['duplicate_doc']['project']}/{dup['duplicate_doc']['source_path']}",  # noqa: D401, E501
                    ctx,
                )

            raise ValueError(f"Document ID collision detected. {len(duplicates)} duplicate IDs found. Check logs for details.")

    async def search_get_reference_guide(
        self,
        component: str,
        project: str | None = None,
        content: bool = True,
        ctx: Context | None = None,
    ) -> list[Document]:
        """Search for reference guides for a specific component."""
        async with self.db_lock:
            await self.ensure_indexed()

            # Build search strategies
            filters: list[dict[str, Any]] = []
            if project:
                filters.append({"project": str(project)})
            filters.append({"source_path_stem": str(component)})
            filters.append({"is_reference": True})
            where_clause: dict[str, Any] = {"$and": filters} if len(filters) > 1 else filters[0]

            filename_results = self.collection.query(query_texts=[component], n_results=1000, where=where_clause)

            # Group chunks by source_path, then merge content per document
            grouped: dict[str, list[tuple[int, str, Any]]] = {}
            if filename_results["ids"] and filename_results["ids"][0]:
                for i, _ in enumerate(filename_results["ids"][0]):
                    if filename_results["metadatas"] and filename_results["metadatas"][0]:
                        metadata = filename_results["metadatas"][0][i]
                        source_path = str(metadata["source_path"])

                        # Validate filters
                        if project and str(metadata["project"]) != project:
                            await log_exception(f"Project mismatch for component '{component}': expected '{project}', got '{metadata['project']}'", ctx)
                            continue
                        if metadata["source_path_stem"] != component:
                            await log_exception(f"Path stem mismatch for component '{component}': expected '{component}', got '{metadata['source_path_stem']}'", ctx)
                            continue

                        content_text = filename_results["documents"][0][i] if (content and filename_results["documents"]) else None
                        chunk_index = int(metadata.get("chunk_index", 0) or 0)

                        if source_path not in grouped:
                            grouped[source_path] = []
                        grouped[source_path].append((chunk_index, content_text or "", metadata))

            # Build one Document per unique source_path
            all_results: list[Document] = []
            for source_path, chunks in grouped.items():
                chunks.sort(key=lambda c: c[0])
                metadata = chunks[0][2]

                doc_title = str(metadata["title"])

                # Merge content from all chunks if content was requested.
                # Strip the title prefix that chunk_document() prepends for embedding.
                if content:
                    merged_content: str | None = "\n".join(
                        _strip_title_prefix(
                            c[1],
                            doc_title,
                            project=str(metadata["project"]),
                            source_path=source_path,
                            is_reference=bool(metadata["is_reference"]),
                        )
                        for c in chunks
                    )
                else:
                    merged_content = None

                # Safe URL construction
                url_value = metadata.get("url", "https://example.com")
                if not url_value or url_value == "None" or not isinstance(url_value, str):
                    url_value = "https://example.com"

                document = Document(
                    title=doc_title,
                    url=HttpUrl(url_value),
                    project=str(metadata["project"]),
                    source_path=source_path,
                    source_url=HttpUrl(str(metadata.get("source_url", ""))),
                    description=str(metadata["description"]),
                    is_reference=bool(metadata["is_reference"]),
                    content=merged_content,
                    relevance_score=1.0,
                )
                all_results.append(document)

            return all_results

    def _reconstruct_document_content(self, source_path: str, project: str) -> str:
        """Reconstruct full document content from its chunks in ChromaDB.

        Uses collection.get() (metadata filter, no embedding computation)
        to fetch all chunks for a document, sorts by chunk_index, strips
        title prefixes, and joins into a single string.

        Must be called while holding db_lock.

        Parameters
        ----------
        source_path : str
            The source path of the document.
        project : str
            The project name.

        Returns
        -------
        str
            The reconstructed document content, or empty string if not found.
        """
        results = self.collection.get(
            where={"$and": [{"project": project}, {"source_path": source_path}]},
            include=["documents", "metadatas"],
        )
        if not results["ids"]:
            return ""

        chunks: list[tuple[int, str, str]] = []
        for i, _ in enumerate(results["ids"]):
            metadata = results["metadatas"][i] if results["metadatas"] else {}
            content_text = results["documents"][i] if results["documents"] else ""
            chunk_index = int(metadata.get("chunk_index", 0) or 0)
            title = str(metadata.get("title", ""))
            chunks.append((chunk_index, content_text or "", title))

        chunks.sort(key=lambda c: c[0])
        title = chunks[0][2]
        is_ref = bool(results["metadatas"][0].get("is_reference", False)) if results["metadatas"] else False
        return "\n".join(_strip_title_prefix(c[1], title, project=project, source_path=source_path, is_reference=is_ref) for c in chunks)

    def _extract_documents_from_results(
        self,
        results: dict[str, Any],
        documents: list[Document],
        seen_paths: set[str],
        max_results: int,
        content_mode: str | None,
        max_content_chars: int | None,
        query: str,
    ) -> None:
        """Extract :class:`Document` objects from a ChromaDB query result.

        Appends to *documents* in place, deduplicating by ``source_path``
        via the shared *seen_paths* set.  Stops once *documents* reaches
        *max_results*.

        Must be called while holding ``db_lock``.

        Parameters
        ----------
        results : dict[str, Any]
            Raw ChromaDB ``collection.query()`` return value.
        documents : list[Document]
            Accumulator list — new documents are appended here.
        seen_paths : set[str]
            Already-seen ``source_path`` values for deduplication.
        max_results : int
            Stop after this many documents have been collected.
        content_mode : str | None
            One of ``"chunk"``, ``"truncated"``, ``"full"``, or ``None``.
        max_content_chars : int | None
            Passed through to :func:`truncate_content`.
        query : str
            Original search query (used for context-aware truncation).
        """
        if not (results["ids"] and results["ids"][0]):
            return

        for i, _ in enumerate(results["ids"][0]):
            if len(documents) >= max_results:
                break

            if not (results["metadatas"] and results["metadatas"][0]):
                continue

            metadata = results["metadatas"][0][i]

            # Deduplicate by source_path — keep only the best-scoring chunk per document
            source_path = str(metadata["source_path"])
            if source_path in seen_paths:
                continue
            seen_paths.add(source_path)

            # Resolve content based on content mode
            if content_mode is None:
                content_text = None
            elif content_mode == "chunk":
                content_text = results["documents"][0][i] if results["documents"] else None
                if content_text:
                    content_text = _strip_title_prefix(
                        content_text,
                        str(metadata["title"]),
                        project=str(metadata.get("project", "")),
                        source_path=source_path,
                        is_reference=bool(metadata.get("is_reference", False)),
                    )
                content_text = truncate_content(content_text, max_content_chars, query=query)
            else:
                # "truncated", "full", or unknown mode — reconstruct full document
                content_text = self._reconstruct_document_content(source_path, str(metadata["project"]))
                if content_mode != "full":
                    content_text = truncate_content(content_text, max_content_chars, query=query)

            # Safe URL construction
            url_value = metadata.get("url", "https://example.com")
            if not url_value or url_value == "None" or not isinstance(url_value, str):
                url_value = "https://example.com"

            # Safe relevance score calculation
            relevance_score = None
            if (
                results.get("distances")
                and isinstance(results["distances"], list)
                and len(results["distances"]) > 0
                and isinstance(results["distances"][0], list)
                and len(results["distances"][0]) > i
            ):
                try:
                    relevance_score = (2.0 - float(results["distances"][0][i])) / 2.0
                except (ValueError, TypeError):
                    relevance_score = None

            document = Document(
                title=str(metadata["title"]),
                url=HttpUrl(url_value),
                project=str(metadata["project"]),
                source_path=source_path,
                source_url=HttpUrl(str(metadata.get("source_url", ""))),
                description=str(metadata["description"]),
                is_reference=bool(metadata["is_reference"]),
                content=content_text,
                relevance_score=relevance_score,
            )
            documents.append(document)

    def _merge_search_results(
        self,
        metadata_results: dict[str, Any] | None,
        keyword_results: dict[str, Any] | None,
        semantic_results: dict[str, Any],
        max_results: int,
        content_mode: str | None,
        max_content_chars: int | None,
        query: str,
    ) -> list[Document]:
        """Merge metadata-boosted, keyword-filtered and semantic search results.

        Results are merged in priority order: metadata boost (exact stem
        match) > keyword pre-filter (content ``$contains``) > pure semantic
        similarity.  Deduplication by ``source_path`` is maintained across
        all passes.

        Must be called while holding ``db_lock``.

        Parameters
        ----------
        metadata_results : dict[str, Any] | None
            ChromaDB query result with ``source_path_stem`` metadata filter,
            or ``None`` when no PascalCase terms were found.
        keyword_results : dict[str, Any] | None
            ChromaDB query result with ``where_document`` pre-filter, or
            ``None`` when no technical terms were found.
        semantic_results : dict[str, Any]
            ChromaDB query result from pure semantic similarity.
        max_results : int
            Maximum number of documents to return.
        content_mode : str | None
            Content resolution mode.
        max_content_chars : int | None
            Maximum content characters.
        query : str
            Original search query.

        Returns
        -------
        list[Document]
            Merged, deduplicated document list.
        """
        documents: list[Document] = []
        seen_paths: set[str] = set()

        # Pass 0: metadata-boosted results (exact stem match)
        if metadata_results is not None:
            self._extract_documents_from_results(
                metadata_results,
                documents,
                seen_paths,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

        # Pass 1: keyword-filtered results (content $contains)
        if keyword_results is not None and len(documents) < max_results:
            self._extract_documents_from_results(
                keyword_results,
                documents,
                seen_paths,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

        # Pass 2: fill remaining slots from semantic results
        if len(documents) < max_results:
            self._extract_documents_from_results(
                semantic_results,
                documents,
                seen_paths,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

        return documents

    async def search(
        self,
        query: str,
        project: str | None = None,
        content: str | bool = "truncated",
        max_results: int = 5,
        max_content_chars: int | None = 10000,
        ctx: Context | None = None,
    ) -> list[Document]:
        """Search the documentation using semantic similarity."""
        async with self.db_lock:
            await self.ensure_indexed(ctx=ctx)

            # Normalize content parameter for backward compatibility
            if content is True:
                content_mode = "truncated"
            elif content is False or content is None:
                content_mode = None
            else:
                content_mode = str(content)

            # Build where clause for filtering
            where_clause = {"project": str(project)} if project else None

            # Over-query to allow deduplication across chunks of the same document
            n_results = max_results * 3

            # Extract technical terms for keyword pre-filtering
            tech_terms = extract_tech_terms(query)
            # Extract PascalCase terms for metadata boost + content matching
            pascal_terms = extract_pascal_terms(query)

            # When a project filter is active, drop terms that match the
            # project name itself — every document in the project trivially
            # contains the project name, so the pre-filter adds no selectivity
            # and fills merge slots with irrelevant docs.
            if project and (tech_terms or pascal_terms):
                project_lower = project.lower().replace("-", "")
                tech_terms = [t for t in tech_terms if t.lower().replace("-", "") != project_lower]
                pascal_terms = [t for t in pascal_terms if t.lower().replace("-", "") != project_lower]

            # Combine for content pre-filter: tech_terms + pascal_terms (deduplicated)
            all_content_terms = list(dict.fromkeys(tech_terms + pascal_terms))
            where_doc = _build_where_document_clause(all_content_terms)

            # Query 0: metadata boost on source_path_stem
            metadata_results = None
            if pascal_terms:
                stem_clause = _build_stem_boost_clause(pascal_terms, project)
                if stem_clause:
                    metadata_results = self.collection.query(
                        query_texts=[query],
                        n_results=n_results,
                        where=stem_clause,  # type: ignore[arg-type]
                    )

            # Query 1: keyword-filtered (only when content terms are found)
            keyword_results = None
            if where_doc:
                keyword_results = self.collection.query(
                    query_texts=[query],
                    n_results=n_results,
                    where=where_clause,
                    where_document=where_doc,  # type: ignore[arg-type]
                )

            # Query 2: pure semantic similarity (always)
            semantic_results = self.collection.query(query_texts=[query], n_results=n_results, where=where_clause)  # type: ignore[arg-type]

            return self._merge_search_results(
                metadata_results,
                keyword_results,
                semantic_results,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

    async def get_document(self, path: str, project: str, ctx: Context | None = None) -> Document:
        """Get a specific document, reconstructing from chunks if needed."""
        async with self.db_lock:
            await self.ensure_indexed(ctx=ctx)

            # Reconstruct full content from chunks
            merged_content = self._reconstruct_document_content(path, project)
            if not merged_content:
                raise ValueError(f"No document found for path '{path}' in project '{project}'.")

            # Get metadata from a single chunk (for Document fields)
            results = self.collection.get(
                where={"$and": [{"project": project}, {"source_path": path}]},
                include=["metadatas"],
                limit=1,
            )
            metadata = results["metadatas"][0] if results["metadatas"] else {}

            # Safe URL construction
            url_value = metadata.get("url", "https://example.com")
            if not url_value or url_value == "None" or not isinstance(url_value, str):
                url_value = "https://example.com"

            return Document(
                title=str(metadata.get("title", "")),
                url=HttpUrl(url_value),
                project=str(metadata.get("project", "")),
                source_path=str(metadata.get("source_path", "")),
                source_url=HttpUrl(str(metadata.get("source_url", ""))),
                description=str(metadata.get("description", "")),
                is_reference=bool(metadata.get("is_reference", False)),
                content=merged_content,
                relevance_score=None,
            )

    async def list_projects(self) -> list[str]:
        """List all available projects with documentation in the index.

        Returns
        -------
        list[str]: A list of project names that have documentation available.
                   Names are returned in hyphenated format (e.g., "panel-material-ui").
        """
        async with self.db_lock:
            await self.ensure_indexed()

            try:
                # Get all documents from the collection to extract unique project names
                results = self.collection.get()

                if not results["metadatas"]:
                    return []

                # Extract unique project names
                projects = set()
                for metadata in results["metadatas"]:
                    project = metadata.get("project")
                    if project:
                        # Convert underscored names to hyphenated format for consistency
                        project_name = str(project).replace("_", "-")
                        projects.add(project_name)

                # Return sorted list
                return sorted(projects)

            except Exception as e:
                logger.error(f"Failed to list projects: {e}")
                return []

    async def _log_summary_table(self, ctx: Context | None = None):
        """Log a summary table showing document counts by repository."""
        try:
            # Get all documents from the collection
            results = self.collection.get()

            if not results["metadatas"]:
                await log_info("No documents found in index", ctx)
                return

            # Count unique documents (not chunks) by project and type
            project_stats: dict[str, dict[str, int]] = {}
            seen_docs: set[str] = set()
            total_chunks = 0

            for metadata in results["metadatas"]:
                total_chunks += 1
                project = str(metadata.get("project", "unknown"))
                source_path = str(metadata.get("source_path", ""))
                is_reference = metadata.get("is_reference", False)

                # Only count each unique document once (not each chunk)
                doc_key = f"{project}::{source_path}"
                if doc_key in seen_docs:
                    continue
                seen_docs.add(doc_key)

                if project not in project_stats:
                    project_stats[project] = {"total": 0, "regular": 0, "reference": 0}

                project_stats[project]["total"] += 1
                if is_reference:
                    project_stats[project]["reference"] += 1
                else:
                    project_stats[project]["regular"] += 1

            # Log summary table
            await log_info("", ctx)
            await log_info("📊 Document Summary by Repository:", ctx)
            await log_info("=" * 60, ctx)
            await log_info(f"{'Repository':<20} {'Total':<8} {'Regular':<8} {'Reference':<10}", ctx)
            await log_info("-" * 60, ctx)

            total_docs = 0
            total_regular = 0
            total_reference = 0

            for project in sorted(project_stats.keys()):
                stats = project_stats[project]
                await log_info(f"{project:<20} {stats['total']:<8} {stats['regular']:<8} {stats['reference']:<10}", ctx)
                total_docs += stats["total"]
                total_regular += stats["regular"]
                total_reference += stats["reference"]

            await log_info("-" * 60, ctx)
            await log_info(f"{'TOTAL':<20} {total_docs:<8} {total_regular:<8} {total_reference:<10}", ctx)
            await log_info("=" * 60, ctx)
            await log_info(f"Total chunks in vector database: {total_chunks}", ctx)

        except Exception as e:
            await log_warning(f"Failed to generate summary table: {e}", ctx)

    def run(self, projects: list[str] | None = None, full_rebuild: bool = False):
        """Update the DocumentationIndexer.

        Parameters
        ----------
        projects : list[str] | None
            Only process these projects. None means all.
        full_rebuild : bool
            Force full rebuild, ignoring cached hashes.
        """
        # Configure logging for the CLI
        logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()])

        logger.info("HoloViz MCP Documentation Indexer")
        logger.info("=" * 50)

        async def run_indexer(indexer=self):
            logger.info(f"Default config: {indexer._holoviz_mcp_config.config_file_path(location='default')}")
            logger.info(f"User config: {indexer._holoviz_mcp_config.config_file_path(location='user')}")
            logger.info(f"Repository directory: {indexer.repos_dir}")
            logger.info(f"Vector database: {self._vector_db_path}")
            logger.info(f"Hash cache: {self._hash_file_path}")
            logger.info(f"Configured repositories: {len(indexer.config.repositories)}")
            if projects:
                logger.info(f"Target projects: {', '.join(projects)}")
            if full_rebuild:
                logger.info("Mode: full rebuild (ignoring hashes)")
            logger.info("")

            await indexer.index_documentation(projects=projects, full_rebuild=full_rebuild)

            # Final summary
            count = indexer.collection.count()
            logger.info("")
            logger.info("=" * 50)
            logger.info("Indexing completed successfully!")
            logger.info(f"Total documents in database: {count}")
            logger.info("=" * 50)

        asyncio.run(run_indexer())

chroma_client = chromadb.PersistentClient(path=(str(self._vector_db_path))) instance-attribute

collection = self.chroma_client.get_or_create_collection('holoviz_docs', configuration=_CROMA_CONFIGURATION) instance-attribute

config = get_config().docs instance-attribute

data_dir = data_dir or config.user_dir instance-attribute

db_lock property

Lazy-initialize and return the database lock.

This ensures the lock is created in the correct event loop context.

repos_dir = repos_dir or config.repos_dir instance-attribute

clone_or_update_repo(repo_name, repo_config, ctx=None) async

Clone or update a single repository.

Delegates to the synchronous implementation via run_in_executor.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def clone_or_update_repo(self, repo_name: str, repo_config: "GitRepository", ctx: Context | None = None) -> Optional[Path]:
    """Clone or update a single repository.

    Delegates to the synchronous implementation via run_in_executor.
    """
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, self._clone_or_update_repo_sync, repo_name, repo_config)

convert_notebook_to_markdown(notebook_path)

Convert a Jupyter notebook to markdown.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def convert_notebook_to_markdown(self, notebook_path: Path) -> str:
    """Convert a Jupyter notebook to markdown."""
    try:
        with open(notebook_path, "r", encoding="utf-8") as f:
            notebook = nbread(f, as_version=4)

        (body, resources) = self._get_nb_exporter().from_notebook_node(notebook)
        return body
    except Exception as e:
        logger.error(f"Failed to convert notebook {notebook_path}: {e}")
        return str(e)

ensure_indexed(ctx=None) async

Ensure documentation is indexed, creating if necessary.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def ensure_indexed(self, ctx: Context | None = None):
    """Ensure documentation is indexed, creating if necessary."""
    if not self.is_indexed():
        await log_info("Documentation index not found. Creating initial index...", ctx)
        await self.index_documentation()

extract_docs_from_repo(repo_path, project, ctx=None) async

Extract documentation files from a repository.

Delegates to the synchronous implementation via run_in_executor.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def extract_docs_from_repo(self, repo_path: Path, project: str, ctx: Context | None = None) -> dict[str, Any]:
    """Extract documentation files from a repository.

    Delegates to the synchronous implementation via run_in_executor.
    """
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, self._extract_docs_from_repo_sync, repo_path, project)

get_document(path, project, ctx=None) async

Get a specific document, reconstructing from chunks if needed.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def get_document(self, path: str, project: str, ctx: Context | None = None) -> Document:
    """Get a specific document, reconstructing from chunks if needed."""
    async with self.db_lock:
        await self.ensure_indexed(ctx=ctx)

        # Reconstruct full content from chunks
        merged_content = self._reconstruct_document_content(path, project)
        if not merged_content:
            raise ValueError(f"No document found for path '{path}' in project '{project}'.")

        # Get metadata from a single chunk (for Document fields)
        results = self.collection.get(
            where={"$and": [{"project": project}, {"source_path": path}]},
            include=["metadatas"],
            limit=1,
        )
        metadata = results["metadatas"][0] if results["metadatas"] else {}

        # Safe URL construction
        url_value = metadata.get("url", "https://example.com")
        if not url_value or url_value == "None" or not isinstance(url_value, str):
            url_value = "https://example.com"

        return Document(
            title=str(metadata.get("title", "")),
            url=HttpUrl(url_value),
            project=str(metadata.get("project", "")),
            source_path=str(metadata.get("source_path", "")),
            source_url=HttpUrl(str(metadata.get("source_url", ""))),
            description=str(metadata.get("description", "")),
            is_reference=bool(metadata.get("is_reference", False)),
            content=merged_content,
            relevance_score=None,
        )

index_documentation(ctx=None, projects=None, full_rebuild=False) async

Index documentation, incrementally when possible.

Parameters:

Name	Type	Description	Default
ctx	Context | None	FastMCP context for logging.	None
projects	list[str] | None	Only process these projects. None means all.	None
full_rebuild	bool	Force full rebuild, ignoring cached hashes.	False

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def index_documentation(
    self,
    ctx: Context | None = None,
    projects: list[str] | None = None,
    full_rebuild: bool = False,
) -> None:
    """Index documentation, incrementally when possible.

    Parameters
    ----------
    ctx : Context | None
        FastMCP context for logging.
    projects : list[str] | None
        Only process these projects. None means all.
    full_rebuild : bool
        Force full rebuild, ignoring cached hashes.
    """
    overall_t0 = time.monotonic()
    await log_info("Starting documentation indexing...", ctx)

    # Validate project names
    if projects:
        available = set(self.config.repositories.keys())
        invalid = [p for p in projects if p not in available]
        if invalid:
            raise ValueError(f"Unknown project(s): {', '.join(invalid)}. Available: {', '.join(sorted(available))}")

    # Determine target repos
    if projects:
        target_repos = {name: cfg for name, cfg in self.config.repositories.items() if name in projects}
    else:
        target_repos = dict(self.config.repositories.items())

    # Load existing hashes BEFORE extraction so workers can skip unchanged files
    old_hashes = self._load_hashes()
    is_full_rebuild = full_rebuild or not old_hashes
    worker_hashes = {} if is_full_rebuild else old_hashes

    all_docs: list[dict[str, Any]] = []
    all_skipped_hashes: dict[str, str] = {}

    # Clone/update repositories and extract documentation in parallel
    # Workers receive old_hashes so they can skip unchanged files early
    # (avoiding expensive notebook conversion for files whose hash matches)
    num_repos = len(target_repos)
    if num_repos > 0:
        max_workers = min(4, num_repos)
        await log_info(f"Processing {num_repos} repositories with {max_workers} workers...", ctx)

        loop = asyncio.get_running_loop()
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                loop.run_in_executor(executor, self._clone_and_extract_repo_sync, repo_name, repo_config, worker_hashes): repo_name
                for repo_name, repo_config in target_repos.items()
            }
            gather_results = await asyncio.gather(*futures.keys(), return_exceptions=True)
            for result, repo_name in zip(gather_results, futures.values(), strict=False):
                if isinstance(result, BaseException):
                    await log_warning(f"Repository {repo_name} failed: {result}", ctx)
                elif isinstance(result, dict):
                    all_docs.extend(result["docs"])
                    all_skipped_hashes.update(result["skipped_hashes"])

    clone_extract_elapsed = time.monotonic() - overall_t0
    await log_info(
        f"Clone + extract completed in {clone_extract_elapsed:.1f}s ({len(all_docs)} to index, {len(all_skipped_hashes)} unchanged)",
        ctx,
    )

    # Detect deleted docs: in old hashes for target projects but not extracted or skipped
    target_project_names = set(target_repos.keys())
    all_known_ids = {doc["id"] for doc in all_docs} | set(all_skipped_hashes.keys())
    deleted_doc_ids: list[str] = [
        doc_id for doc_id in old_hashes if doc_id not in all_known_ids and any(doc_id.startswith(f"{proj}___") for proj in target_project_names)
    ]

    changed_docs: list[dict[str, Any]] = []

    if is_full_rebuild:
        if not all_docs:
            await log_warning("No documentation found to index", ctx)
            return
        docs_to_index = all_docs
        await log_info(f"Full rebuild: indexing all {len(docs_to_index)} documents", ctx)
    else:
        # Incremental: all_docs contains only new/changed files (skipped were filtered by workers)
        new_docs = [doc for doc in all_docs if doc["id"] not in old_hashes]
        changed_docs = [doc for doc in all_docs if doc["id"] in old_hashes]

        docs_to_index = all_docs  # all are new or changed (unchanged were skipped by workers)
        await log_info(
            f"Incremental: {len(new_docs)} new, {len(changed_docs)} changed, {len(all_skipped_hashes)} unchanged, {len(deleted_doc_ids)} deleted",
            ctx,
        )

        if not docs_to_index and not deleted_doc_ids:
            total_elapsed = time.monotonic() - overall_t0
            await log_info(f"Index up to date, nothing to do ({total_elapsed:.1f}s)", ctx)
            # Still save hashes (preserves skipped + old non-target hashes)
            new_hashes = dict(old_hashes)
            for doc_id in deleted_doc_ids:
                new_hashes.pop(doc_id, None)
            self._save_hashes(new_hashes)
            return

    if docs_to_index:
        # Validate for duplicate IDs and log details
        await self._validate_unique_ids(docs_to_index)

    # Split documents into chunks at H1/H2 headers for better embedding quality
    all_chunks: list[dict[str, Any]] = []
    for doc in docs_to_index:
        all_chunks.extend(chunk_document(doc))
    await log_info(f"Chunked {len(docs_to_index)} documents into {len(all_chunks)} chunks", ctx)

    # Create pre-write backup of vector database and hash file
    backup_path = self._backup_path
    try:
        if self._vector_db_path.exists():
            t0 = time.monotonic()
            shutil.copytree(self._vector_db_path, backup_path, dirs_exist_ok=True)
            await log_info(f"Pre-write backup created at {backup_path} ({time.monotonic() - t0:.1f}s)", ctx)
        hash_src = self._hash_file_path
        if hash_src.exists():
            shutil.copy2(hash_src, backup_path.parent / "index_hashes.json.bak")
    except Exception as e:
        await log_warning(f"Failed to create pre-write backup: {e}. Continuing without backup.", ctx)

    if is_full_rebuild:
        # Full rebuild: clear existing collection and re-add everything
        await log_info("Clearing existing index...", ctx)
        try:
            count = self.collection.count()
            if count > 0:
                results = self.collection.get()
                if results["ids"]:
                    delete_batch_size = self.chroma_client.get_max_batch_size()
                    for i in range(0, len(results["ids"]), delete_batch_size):
                        self.collection.delete(ids=results["ids"][i : i + delete_batch_size])
        except Exception as e:
            logger.warning(f"Failed to clear existing collection: {e}")
            try:
                self.chroma_client.delete_collection("holoviz_docs")
                self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
            except Exception as e2:
                await log_exception(f"Failed to recreate collection: {e2}", ctx)
                raise
    else:
        # Incremental: delete chunks for changed and deleted docs
        for doc_id in deleted_doc_ids:
            deleted_count = self._delete_doc_chunks(doc_id)
            if deleted_count > 0:
                logger.debug("Deleted %d chunks for removed doc %s", deleted_count, doc_id)
        for doc in changed_docs:
            deleted_count = self._delete_doc_chunks(doc["id"])
            if deleted_count > 0:
                logger.debug("Deleted %d old chunks for changed doc %s", deleted_count, doc["id"])

    # Add chunks to ChromaDB in batches (ChromaDB enforces a max batch size)
    if all_chunks:
        batch_size = self.chroma_client.get_max_batch_size()
        await log_info(f"Adding {len(all_chunks)} chunks from {len(docs_to_index)} documents to index (batch size {batch_size})...", ctx)

        try:
            for batch_start in range(0, len(all_chunks), batch_size):
                batch = all_chunks[batch_start : batch_start + batch_size]
                self.collection.add(
                    documents=[doc["content"] for doc in batch],
                    metadatas=[
                        {
                            "title": doc["title"],
                            "url": doc["url"],
                            "project": doc["project"],
                            "source_path": doc["source_path"],
                            "source_path_stem": doc["source_path_stem"],
                            "source_url": doc["source_url"],
                            "description": doc["description"],
                            "is_reference": doc["is_reference"],
                            "chunk_index": doc["chunk_index"],
                            "parent_id": doc["parent_id"],
                        }
                        for doc in batch
                    ],
                    ids=[doc["id"] for doc in batch],
                )
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException as e:
            logger.error("ChromaDB write failed (%s: %s). Attempting restore from backup.", type(e).__name__, e)
            await self._restore_from_backup(ctx)
            raise

    # Save updated hashes (after successful ChromaDB write)
    new_hashes = dict(old_hashes)  # start from existing (preserves non-target projects)
    # Remove deleted docs
    for doc_id in deleted_doc_ids:
        new_hashes.pop(doc_id, None)
    # Add/update hashes for indexed docs
    for doc in docs_to_index:
        file_hash = doc.get("file_hash")
        if file_hash:
            new_hashes[doc["id"]] = file_hash
    # Include hashes for files that were skipped (unchanged)
    new_hashes.update(all_skipped_hashes)
    self._save_hashes(new_hashes)

    total_elapsed = time.monotonic() - overall_t0
    await log_info(f"Successfully indexed {len(all_chunks)} chunks from {len(docs_to_index)} documents in {total_elapsed:.1f}s", ctx)
    await log_info(f"Vector database stored at: {self._vector_db_path}", ctx)
    await log_info(f"Index contains {self.collection.count()} total documents", ctx)

    # Show detailed summary table
    await self._log_summary_table(ctx)

is_indexed()

Check if documentation index exists and is valid.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def is_indexed(self) -> bool:
    """Check if documentation index exists and is valid."""
    try:
        count = self.collection.count()
        return count > 0
    except (KeyboardInterrupt, SystemExit):
        raise
    except BaseException:
        return False

list_projects() async

List all available projects with documentation in the index.

Returns:

Type	Description
list[str]: A list of project names that have documentation available.	Names are returned in hyphenated format (e.g., "panel-material-ui").

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def list_projects(self) -> list[str]:
    """List all available projects with documentation in the index.

    Returns
    -------
    list[str]: A list of project names that have documentation available.
               Names are returned in hyphenated format (e.g., "panel-material-ui").
    """
    async with self.db_lock:
        await self.ensure_indexed()

        try:
            # Get all documents from the collection to extract unique project names
            results = self.collection.get()

            if not results["metadatas"]:
                return []

            # Extract unique project names
            projects = set()
            for metadata in results["metadatas"]:
                project = metadata.get("project")
                if project:
                    # Convert underscored names to hyphenated format for consistency
                    project_name = str(project).replace("_", "-")
                    projects.add(project_name)

            # Return sorted list
            return sorted(projects)

        except Exception as e:
            logger.error(f"Failed to list projects: {e}")
            return []

process_file(file_path, project, repo_config, folder_name='')

Process a file and extract metadata.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def process_file(self, file_path: Path, project: str, repo_config: GitRepository, folder_name: str = "") -> Optional[dict[str, Any]]:
    """Process a file and extract metadata."""
    try:
        if file_path.suffix == ".ipynb":
            content = self.convert_notebook_to_markdown(file_path)
        elif file_path.suffix in [".md", ".rst", ".txt"]:
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
        else:
            logger.debug(f"Skipping unsupported file type: {file_path}")
            return None

        title = self._extract_title_from_markdown(content, file_path.name)
        if not title:
            title = file_path.stem.replace("_", " ").title()

        description = self._extract_description_from_markdown(content)

        repo_path = self.repos_dir / project
        relative_path = file_path.relative_to(repo_path)

        doc_id = self._generate_doc_id(project, relative_path)

        is_reference = self._is_reference_document(file_path, project, folder_name)

        source_url = self._to_source_url(relative_path, repo_config)

        return {
            "id": doc_id,
            "title": title,
            "url": self._generate_doc_url(project, relative_path, folder_name),
            "project": project,
            "source_path": str(relative_path),
            "source_path_stem": file_path.stem,
            "source_url": source_url,
            "description": description,
            "content": content,
            "is_reference": is_reference,
            "file_hash": self._compute_file_hash(file_path),
        }
    except Exception as e:
        logger.error(f"Failed to process file {file_path}: {e}")
        return None

run(projects=None, full_rebuild=False)

Update the DocumentationIndexer.

Parameters:

Name	Type	Description	Default
projects	list[str] | None	Only process these projects. None means all.	None
full_rebuild	bool	Force full rebuild, ignoring cached hashes.	False

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def run(self, projects: list[str] | None = None, full_rebuild: bool = False):
    """Update the DocumentationIndexer.

    Parameters
    ----------
    projects : list[str] | None
        Only process these projects. None means all.
    full_rebuild : bool
        Force full rebuild, ignoring cached hashes.
    """
    # Configure logging for the CLI
    logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()])

    logger.info("HoloViz MCP Documentation Indexer")
    logger.info("=" * 50)

    async def run_indexer(indexer=self):
        logger.info(f"Default config: {indexer._holoviz_mcp_config.config_file_path(location='default')}")
        logger.info(f"User config: {indexer._holoviz_mcp_config.config_file_path(location='user')}")
        logger.info(f"Repository directory: {indexer.repos_dir}")
        logger.info(f"Vector database: {self._vector_db_path}")
        logger.info(f"Hash cache: {self._hash_file_path}")
        logger.info(f"Configured repositories: {len(indexer.config.repositories)}")
        if projects:
            logger.info(f"Target projects: {', '.join(projects)}")
        if full_rebuild:
            logger.info("Mode: full rebuild (ignoring hashes)")
        logger.info("")

        await indexer.index_documentation(projects=projects, full_rebuild=full_rebuild)

        # Final summary
        count = indexer.collection.count()
        logger.info("")
        logger.info("=" * 50)
        logger.info("Indexing completed successfully!")
        logger.info(f"Total documents in database: {count}")
        logger.info("=" * 50)

    asyncio.run(run_indexer())

search(query, project=None, content='truncated', max_results=5, max_content_chars=10000, ctx=None) async

Search the documentation using semantic similarity.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def search(
    self,
    query: str,
    project: str | None = None,
    content: str | bool = "truncated",
    max_results: int = 5,
    max_content_chars: int | None = 10000,
    ctx: Context | None = None,
) -> list[Document]:
    """Search the documentation using semantic similarity."""
    async with self.db_lock:
        await self.ensure_indexed(ctx=ctx)

        # Normalize content parameter for backward compatibility
        if content is True:
            content_mode = "truncated"
        elif content is False or content is None:
            content_mode = None
        else:
            content_mode = str(content)

        # Build where clause for filtering
        where_clause = {"project": str(project)} if project else None

        # Over-query to allow deduplication across chunks of the same document
        n_results = max_results * 3

        # Extract technical terms for keyword pre-filtering
        tech_terms = extract_tech_terms(query)
        # Extract PascalCase terms for metadata boost + content matching
        pascal_terms = extract_pascal_terms(query)

        # When a project filter is active, drop terms that match the
        # project name itself — every document in the project trivially
        # contains the project name, so the pre-filter adds no selectivity
        # and fills merge slots with irrelevant docs.
        if project and (tech_terms or pascal_terms):
            project_lower = project.lower().replace("-", "")
            tech_terms = [t for t in tech_terms if t.lower().replace("-", "") != project_lower]
            pascal_terms = [t for t in pascal_terms if t.lower().replace("-", "") != project_lower]

        # Combine for content pre-filter: tech_terms + pascal_terms (deduplicated)
        all_content_terms = list(dict.fromkeys(tech_terms + pascal_terms))
        where_doc = _build_where_document_clause(all_content_terms)

        # Query 0: metadata boost on source_path_stem
        metadata_results = None
        if pascal_terms:
            stem_clause = _build_stem_boost_clause(pascal_terms, project)
            if stem_clause:
                metadata_results = self.collection.query(
                    query_texts=[query],
                    n_results=n_results,
                    where=stem_clause,  # type: ignore[arg-type]
                )

        # Query 1: keyword-filtered (only when content terms are found)
        keyword_results = None
        if where_doc:
            keyword_results = self.collection.query(
                query_texts=[query],
                n_results=n_results,
                where=where_clause,
                where_document=where_doc,  # type: ignore[arg-type]
            )

        # Query 2: pure semantic similarity (always)
        semantic_results = self.collection.query(query_texts=[query], n_results=n_results, where=where_clause)  # type: ignore[arg-type]

        return self._merge_search_results(
            metadata_results,
            keyword_results,
            semantic_results,
            max_results,
            content_mode,
            max_content_chars,
            query,
        )

search_get_reference_guide(component, project=None, content=True, ctx=None) async

Search for reference guides for a specific component.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def search_get_reference_guide(
    self,
    component: str,
    project: str | None = None,
    content: bool = True,
    ctx: Context | None = None,
) -> list[Document]:
    """Search for reference guides for a specific component."""
    async with self.db_lock:
        await self.ensure_indexed()

        # Build search strategies
        filters: list[dict[str, Any]] = []
        if project:
            filters.append({"project": str(project)})
        filters.append({"source_path_stem": str(component)})
        filters.append({"is_reference": True})
        where_clause: dict[str, Any] = {"$and": filters} if len(filters) > 1 else filters[0]

        filename_results = self.collection.query(query_texts=[component], n_results=1000, where=where_clause)

        # Group chunks by source_path, then merge content per document
        grouped: dict[str, list[tuple[int, str, Any]]] = {}
        if filename_results["ids"] and filename_results["ids"][0]:
            for i, _ in enumerate(filename_results["ids"][0]):
                if filename_results["metadatas"] and filename_results["metadatas"][0]:
                    metadata = filename_results["metadatas"][0][i]
                    source_path = str(metadata["source_path"])

                    # Validate filters
                    if project and str(metadata["project"]) != project:
                        await log_exception(f"Project mismatch for component '{component}': expected '{project}', got '{metadata['project']}'", ctx)
                        continue
                    if metadata["source_path_stem"] != component:
                        await log_exception(f"Path stem mismatch for component '{component}': expected '{component}', got '{metadata['source_path_stem']}'", ctx)
                        continue

                    content_text = filename_results["documents"][0][i] if (content and filename_results["documents"]) else None
                    chunk_index = int(metadata.get("chunk_index", 0) or 0)

                    if source_path not in grouped:
                        grouped[source_path] = []
                    grouped[source_path].append((chunk_index, content_text or "", metadata))

        # Build one Document per unique source_path
        all_results: list[Document] = []
        for source_path, chunks in grouped.items():
            chunks.sort(key=lambda c: c[0])
            metadata = chunks[0][2]

            doc_title = str(metadata["title"])

            # Merge content from all chunks if content was requested.
            # Strip the title prefix that chunk_document() prepends for embedding.
            if content:
                merged_content: str | None = "\n".join(
                    _strip_title_prefix(
                        c[1],
                        doc_title,
                        project=str(metadata["project"]),
                        source_path=source_path,
                        is_reference=bool(metadata["is_reference"]),
                    )
                    for c in chunks
                )
            else:
                merged_content = None

            # Safe URL construction
            url_value = metadata.get("url", "https://example.com")
            if not url_value or url_value == "None" or not isinstance(url_value, str):
                url_value = "https://example.com"

            document = Document(
                title=doc_title,
                url=HttpUrl(url_value),
                project=str(metadata["project"]),
                source_path=source_path,
                source_url=HttpUrl(str(metadata.get("source_url", ""))),
                description=str(metadata["description"]),
                is_reference=bool(metadata["is_reference"]),
                content=merged_content,
                relevance_score=1.0,
            )
            all_results.append(document)

        return all_results

build_excerpts(content, matches, max_chars, context_chars)

Build excerpt string from matches with context windows.

Combines nearby matches, adds separators for distant sections.

Args: content: Full document content matches: List of (start_pos, end_pos, keyword) tuples max_chars: Maximum total characters to return context_chars: Characters to include before/after each match

Returns:

Type	Description
Excerpt(s) with [...] separators and truncation indicators

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def build_excerpts(content: str, matches: list[tuple[int, int, str]], max_chars: int, context_chars: int) -> str:
    """Build excerpt string from matches with context windows.

    Combines nearby matches, adds separators for distant sections.

    Args:
        content: Full document content
        matches: List of (start_pos, end_pos, keyword) tuples
        max_chars: Maximum total characters to return
        context_chars: Characters to include before/after each match

    Returns
    -------
        Excerpt(s) with [...] separators and truncation indicators
    """
    if not matches:
        # Fallback to beginning truncation
        truncated = content[:max_chars]
        last_space = truncated.rfind(" ")
        if last_space > max_chars * 0.8:
            truncated = truncated[:last_space]
        return truncated + "\n\n[... content truncated, use content='full' for complete content ...]"

    # Cluster nearby matches
    clusters = []
    current_cluster = [matches[0]]

    for match in matches[1:]:
        # If within 2x context window, add to current cluster
        if match[0] - current_cluster[-1][1] < 2 * context_chars:
            current_cluster.append(match)
        else:
            clusters.append(current_cluster)
            current_cluster = [match]
    clusters.append(current_cluster)

    # Build excerpts from clusters
    excerpts: list[str] = []
    total_chars = 0

    for cluster in clusters:
        # Get range for this cluster
        start_pos = max(0, cluster[0][0] - context_chars)
        end_pos = min(len(content), cluster[-1][1] + context_chars)

        # Extract excerpt, break at word boundaries
        excerpt = content[start_pos:end_pos]

        # Trim to word boundaries
        if start_pos > 0:
            # Find first space to start at word boundary
            first_space = excerpt.find(" ")
            if first_space != -1 and first_space < context_chars * 0.3:
                excerpt = excerpt[first_space + 1 :]
                start_pos += first_space + 1

        if end_pos < len(content):
            # Find last space to end at word boundary
            last_space = excerpt.rfind(" ")
            if last_space > len(excerpt) * 0.7:
                excerpt = excerpt[:last_space]

        # Check if we have room
        separator = "\n\n[...]\n\n" if excerpts else ""
        if total_chars + len(excerpt) + len(separator) > max_chars:
            # Try to fit partial excerpt
            remaining = max_chars - total_chars - len(separator)
            if remaining > 200:  # Only add if we have reasonable space
                excerpt = excerpt[:remaining]
                last_space = excerpt.rfind(" ")
                if last_space > remaining * 0.7:
                    excerpt = excerpt[:last_space]
                excerpts.append(excerpt)
            break

        excerpts.append(excerpt)
        total_chars += len(excerpt) + len(separator)

    if not excerpts:
        # Fallback if nothing fits
        return build_excerpts(content, [], max_chars, context_chars)

    # Combine excerpts
    result = "\n\n[...]\n\n".join(excerpts)

    # Add indicators at start/end if content was truncated
    first_match_pos = matches[0][0] if matches else 0
    last_match_pos = matches[-1][1] if matches else len(content)

    if first_match_pos > context_chars:
        result = "[...]\n\n" + result
    if last_match_pos < len(content) - context_chars:
        result = result + "\n\n[...]"

    return result

chunk_document(doc, min_chunk_chars=100)

Split a document into chunks at H1/H2 markdown headers.

Only headers outside fenced code blocks are used as split points, so Python comments (# ...) and decorative dividers inside code blocks are left intact.

Each chunk stores two content fields:

• content: the document title prepended to the raw section text ("Title\\n\\n## Section ...") so that ChromaDB's embedding model associates every chunk with its parent document context.
• raw_content: the original section text without the title prefix, used by get_document() and search_get_reference_guide() to reconstruct the full document without duplicating the title.

Parameters:

Name	Type	Description	Default
doc	dict[str, Any]	Document dict with at least 'id', 'title', and 'content' keys, plus other metadata fields.	required
min_chunk_chars	int	Minimum character count for a chunk to be kept. Chunks below this threshold are discarded (e.g. empty sections). Default: 100.	100

Returns:

Type	Description
list[dict[str, Any]]	List of chunk dicts. If no H1/H2 headers are found, returns a single chunk with chunk_index=0.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def chunk_document(doc: dict[str, Any], min_chunk_chars: int = 100) -> list[dict[str, Any]]:
    r"""Split a document into chunks at H1/H2 markdown headers.

    Only headers **outside** fenced code blocks are used as split points,
    so Python comments (``# ...``) and decorative dividers inside code
    blocks are left intact.

    Each chunk stores two content fields:

    - ``content``: the document title prepended to the raw section text
      (``"Title\\n\\n## Section ..."``) so that ChromaDB's embedding model
      associates every chunk with its parent document context.
    - ``raw_content``: the original section text without the title prefix,
      used by ``get_document()`` and ``search_get_reference_guide()`` to
      reconstruct the full document without duplicating the title.

    Parameters
    ----------
    doc : dict[str, Any]
        Document dict with at least 'id', 'title', and 'content' keys,
        plus other metadata fields.
    min_chunk_chars : int
        Minimum character count for a chunk to be kept. Chunks below this
        threshold are discarded (e.g. empty sections). Default: 100.

    Returns
    -------
    list[dict[str, Any]]
        List of chunk dicts. If no H1/H2 headers are found, returns a single
        chunk with chunk_index=0.
    """
    content = doc.get("content", "") or ""
    parent_id = doc["id"]
    title = doc.get("title", "")

    # Find H1/H2 header lines that are outside code fences
    lines = content.split("\n")
    header_indices = _find_markdown_header_lines(content)

    # Build text parts by splitting at header boundaries
    if header_indices:
        parts: list[str] = []
        prev = 0
        for idx in header_indices:
            if idx > prev:
                parts.append("\n".join(lines[prev:idx]))
            prev = idx
        # Last chunk: from last header to end
        parts.append("\n".join(lines[prev:]))
    else:
        parts = [content]

    # Filter out tiny/empty chunks
    parts = [p for p in parts if len(p.strip()) >= min_chunk_chars]

    # If nothing survived filtering, keep the whole content as one chunk
    if not parts:
        parts = [content]

    # Metadata keys to copy from the parent document to each chunk
    metadata_keys = ("title", "url", "project", "source_path", "source_path_stem", "source_url", "description", "is_reference")

    chunks: list[dict[str, Any]] = []
    for idx, part in enumerate(parts):
        chunk: dict[str, Any] = {}
        for key in metadata_keys:
            if key in doc:
                chunk[key] = doc[key]
        chunk["id"] = f"{parent_id}___chunk_{idx}"
        chunk["chunk_index"] = idx
        chunk["parent_id"] = parent_id
        # raw_content: original section text for faithful document reconstruction
        chunk["raw_content"] = part
        # content: context-prefixed text stored in ChromaDB for better embeddings
        context_prefix = _build_context_prefix(doc.get("project", ""), doc.get("source_path", ""), doc.get("is_reference", False))
        if title:
            chunk["content"] = f"{context_prefix}{title}\n\n{part}"
        elif context_prefix:
            chunk["content"] = f"{context_prefix}{part}"
        else:
            chunk["content"] = part
        chunks.append(chunk)

    return chunks

convert_path_to_url(path, remove_first_part=True, url_transform='holoviz')

Convert a relative file path to a URL path.

Converts file paths to web URLs by replacing file extensions with .html and optionally removing the first path component for legacy compatibility.

Args: path: The file path to convert remove_first_part: Whether to remove the first path component (legacy compatibility) url_transform: How to transform the file path into a URL:

    - "holoviz": Replace file extension with .html (default)
    - "plotly": Replace file extension with / (e.g., filename.md -> filename/)
    - "datashader": Remove leading index and replace file extension with .html (e.g., 01_filename.md -> filename.html)

Returns:

Type	Description
URL path with .html extension

Examples:

>>> convert_path_to_url(Path("doc/getting_started.md"))
"getting_started.html"
>>> convert_path_to_url(Path("examples/reference/Button.ipynb"), False)
"examples/reference/Button.html"
>>> convert_path_to_url(Path("/doc/python/3d-axes.md"), False, "plotly")
"/doc/python/3d-axes/"
>>> convert_path_to_url(Path("/examples/user_guide/10_Performance.ipynb"), False, "datashader")
"/examples/user_guide/Performance.html"

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def convert_path_to_url(path: Path, remove_first_part: bool = True, url_transform: Literal["holoviz", "plotly", "datashader"] = "holoviz") -> str:
    """Convert a relative file path to a URL path.

    Converts file paths to web URLs by replacing file extensions with .html
    and optionally removing the first path component for legacy compatibility.

    Args:
        path: The file path to convert
        remove_first_part: Whether to remove the first path component (legacy compatibility)
        url_transform: How to transform the file path into a URL:

            - "holoviz": Replace file extension with .html (default)
            - "plotly": Replace file extension with / (e.g., filename.md -> filename/)
            - "datashader": Remove leading index and replace file extension with .html (e.g., 01_filename.md -> filename.html)

    Returns
    -------
        URL path with .html extension

    Examples
    --------
        >>> convert_path_to_url(Path("doc/getting_started.md"))
        "getting_started.html"
        >>> convert_path_to_url(Path("examples/reference/Button.ipynb"), False)
        "examples/reference/Button.html"
        >>> convert_path_to_url(Path("/doc/python/3d-axes.md"), False, "plotly")
        "/doc/python/3d-axes/"
        >>> convert_path_to_url(Path("/examples/user_guide/10_Performance.ipynb"), False, "datashader")
        "/examples/user_guide/Performance.html"
    """
    if url_transform in ["holoviz", "datashader"]:
        path = remove_leading_number_sep_from_path(path)

    # Convert path to URL format
    parts = list(path.parts)

    # Only remove first part if requested (for legacy compatibility)
    if remove_first_part and parts:
        parts.pop(0)

    # Reconstruct path and convert to string
    if parts:
        url_path = str(Path(*parts))
    else:
        url_path = ""

    # Replace file extensions with suffix
    if url_path:
        path_obj = Path(url_path)
        if url_transform == "plotly":
            url_path = str(path_obj.with_suffix(suffix="")) + "/"
            if url_path.endswith("index/"):
                url_path = url_path[: -len("index/")] + "/"
        else:
            url_path = str(path_obj.with_suffix(suffix=".html"))

    return url_path

extract_keywords(query)

Extract meaningful keywords from search query.

Removes common stopwords and splits into terms.

Args: query: Search query string

Returns:

Type	Description
List of meaningful keywords (lowercase)

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def extract_keywords(query: str) -> list[str]:
    """Extract meaningful keywords from search query.

    Removes common stopwords and splits into terms.

    Args:
        query: Search query string

    Returns
    -------
        List of meaningful keywords (lowercase)
    """
    stopwords = {
        "the",
        "a",
        "an",
        "and",
        "or",
        "but",
        "in",
        "on",
        "at",
        "to",
        "for",
        "of",
        "with",
        "from",
        "by",
        "about",
        "how",
        "what",
        "where",
        "when",
        "why",
        "which",
        "who",
        "is",
        "are",
        "was",
        "were",
        "be",
        "been",
        "being",
        "have",
        "has",
        "had",
        "do",
        "does",
        "did",
        "will",
        "would",
        "should",
        "could",
        "can",
        "may",
        "might",
        "must",
        "shall",
    }

    # Split and clean
    keywords = query.lower().split()
    # Remove stopwords, keep meaningful terms (> 2 chars)
    keywords = [k for k in keywords if k not in stopwords and len(k) > 2]

    return keywords

extract_pascal_terms(query)

Extract single PascalCase words from a query, excluding stopwords.

Captures words like Scatter, Button, Tabulator that start with an uppercase letter followed by at least one lowercase letter. Compound CamelCase words (SelectEditor) are also captured — they overlap with :func:extract_tech_terms and deduplication happens at the call site.

Parameters:

Name	Type	Description	Default
query	str	Search query string.	required

Returns:

Type	Description
list[str]	Deduplicated list of PascalCase terms preserving discovery order. Empty list when no terms are found.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def extract_pascal_terms(query: str) -> list[str]:
    """Extract single PascalCase words from a query, excluding stopwords.

    Captures words like Scatter, Button, Tabulator that start with an
    uppercase letter followed by at least one lowercase letter.  Compound
    CamelCase words (SelectEditor) are also captured — they overlap with
    :func:`extract_tech_terms` and deduplication happens at the call site.

    Parameters
    ----------
    query : str
        Search query string.

    Returns
    -------
    list[str]
        Deduplicated list of PascalCase terms preserving discovery order.
        Empty list when no terms are found.
    """
    terms: list[str] = []
    seen: set[str] = set()
    for m in re.finditer(r"\b[A-Z][a-z][a-zA-Z]*\b", query):
        t = m.group()
        if t not in seen and t not in _PASCAL_STOPWORDS:
            terms.append(t)
            seen.add(t)
    return terms

extract_relevant_excerpt(content, query, max_chars, context_chars=500)

Extract relevant excerpt from content based on query keywords.

Args: content: Full document content query: Search query string max_chars: Maximum total characters to return context_chars: Characters to include before/after each match (default: 500)

Returns:

Type	Description
Excerpt(s) centered around query matches, or beginning if no matches

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def extract_relevant_excerpt(content: str, query: str, max_chars: int, context_chars: int = 500) -> str:
    """Extract relevant excerpt from content based on query keywords.

    Args:
        content: Full document content
        query: Search query string
        max_chars: Maximum total characters to return
        context_chars: Characters to include before/after each match (default: 500)

    Returns
    -------
        Excerpt(s) centered around query matches, or beginning if no matches
    """
    # Extract keywords from query
    keywords = extract_keywords(query)

    if not keywords:
        # No meaningful keywords, fall back to simple truncation
        return build_excerpts(content, [], max_chars, context_chars)

    # Find keyword matches in content
    matches = find_keyword_matches(content, keywords)

    if not matches:
        # No matches found, fall back to simple truncation
        return build_excerpts(content, [], max_chars, context_chars)

    # Build excerpts from matches
    return build_excerpts(content, matches, max_chars, context_chars)

extract_tech_terms(query)

Extract technical identifiers from a search query.

Identifies three categories of terms that benefit from exact substring matching rather than pure semantic similarity:

• Compound CamelCase (requires internal case transition): SelectEditor, ReactiveHTML, TextInput — but NOT single-word PascalCase like Button, Panel, Python.
• snake_case: add_filter, page_size
• Dot-separated qualified names: param.watch, pn.widgets.Button — excludes common abbreviations like e.g, i.e via a blocklist and minimum length filter.

Parameters:

Name	Type	Description	Default
query	str	Search query string.	required

Returns:

Type	Description
list[str]	Deduplicated list of technical terms preserving original case and discovery order. Empty list when no technical terms are found.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def extract_tech_terms(query: str) -> list[str]:
    """Extract technical identifiers from a search query.

    Identifies three categories of terms that benefit from exact substring
    matching rather than pure semantic similarity:

    - **Compound CamelCase** (requires internal case transition):
      ``SelectEditor``, ``ReactiveHTML``, ``TextInput`` — but NOT
      single-word PascalCase like ``Button``, ``Panel``, ``Python``.
    - **snake_case**: ``add_filter``, ``page_size``
    - **Dot-separated qualified names**: ``param.watch``,
      ``pn.widgets.Button`` — excludes common abbreviations like
      ``e.g``, ``i.e`` via a blocklist and minimum length filter.

    Parameters
    ----------
    query : str
        Search query string.

    Returns
    -------
    list[str]
        Deduplicated list of technical terms preserving original case
        and discovery order.  Empty list when no technical terms are found.
    """
    terms: list[str] = []
    seen: set[str] = set()

    # Compound CamelCase: requires an internal lower→upper transition
    # e.g. SelectEditor, ReactiveHTML, TextInput — NOT Button, Panel
    for m in re.finditer(r"\b[A-Z][a-z]+[A-Z][a-zA-Z]*\b", query):
        t = m.group()
        if t not in seen:
            terms.append(t)
            seen.add(t)

    # snake_case identifiers
    for m in re.finditer(r"\b[a-z][a-z0-9]*_[a-z][a-z0-9_]*\b", query):
        t = m.group()
        if t not in seen:
            terms.append(t)
            seen.add(t)

    # Dot-separated qualified names (param.watch, pn.widgets.Button)
    dot_blocklist = {"e.g", "i.e", "vs.", "etc."}
    for m in re.finditer(r"\b[a-z][a-z0-9]*(?:\.[a-zA-Z][a-zA-Z0-9_]*)+\b", query):
        t = m.group()
        if len(t) > 3 and t not in dot_blocklist and t not in seen:
            terms.append(t)
            seen.add(t)

    return terms

find_keyword_matches(content, keywords)

Find all positions where keywords appear in content.

Args: content: Document content to search keywords: List of keywords to find

Returns:

Type	Description
List of (start_pos, end_pos, matched_keyword) tuples, sorted by position

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def find_keyword_matches(content: str, keywords: list[str]) -> list[tuple[int, int, str]]:
    """Find all positions where keywords appear in content.

    Args:
        content: Document content to search
        keywords: List of keywords to find

    Returns
    -------
        List of (start_pos, end_pos, matched_keyword) tuples, sorted by position
    """
    matches = []
    content_lower = content.lower()

    for keyword in keywords:
        start = 0
        while True:
            pos = content_lower.find(keyword, start)
            if pos == -1:
                break
            matches.append((pos, pos + len(keyword), keyword))
            start = pos + 1

    # Sort by position
    matches.sort(key=lambda x: x[0])
    return matches

get_skill(name)

Get skill for using a project with LLMs.

This function searches for skill resources in user and default directories, with user resources taking precedence over default ones.

Args: name (str): The name of the skill to get.

Returns:

Type	Description
str: A string containing the skill in Markdown format.

Raises:

Type	Description
FileNotFoundError: If the specified skill is not found in either directory.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def get_skill(name: str) -> str:
    """Get skill for using a project with LLMs.

    This function searches for skill resources in user and default directories,
    with user resources taking precedence over default ones.

    Args:
        name (str): The name of the skill to get.

    Returns
    -------
        str: A string containing the skill in Markdown format.

    Raises
    ------
        FileNotFoundError: If the specified skill is not found in either directory.
    """
    config = get_config()

    # Convert underscored names to hyphenated for file lookup
    skill_filename = name.replace("_", "-") + ".md"

    # Search in user directory first, then default directory
    search_paths = [
        config.skills_dir("user"),
        config.skills_dir("default"),
    ]

    for search_dir in search_paths:
        skills_file = search_dir / skill_filename
        if skills_file.exists():
            return skills_file.read_text(encoding="utf-8")

    # If not found, raise error with helpful message
    available_files = []
    for search_dir in search_paths:
        if search_dir.exists():
            available_files.extend([f.name for f in search_dir.glob("*.md")])

    available_str = ", ".join(set(available_files)) if available_files else "None"
    raise FileNotFoundError(f"Skill file {name} not found. Available skills: {available_str}. Searched in: {[str(p) for p in search_paths]}")

list_skills()

List all available skills.

This function discovers available skills from both user and default directories, with user resources taking precedence over default ones.

Returns:

Type	Description
list[str]: A list of the skills available.	Names are returned in hyphenated format (e.g., "panel-material-ui").

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def list_skills() -> list[str]:
    """List all available skills.

    This function discovers available skills from both user and default directories,
    with user resources taking precedence over default ones.

    Returns
    -------
        list[str]: A list of the skills available.
            Names are returned in hyphenated format (e.g., "panel-material-ui").
    """
    config = get_config()

    # Collect available projects from both directories
    available_projects = set()

    search_paths = [
        config.skills_dir("user"),
        config.skills_dir("default"),
    ]

    for search_dir in search_paths:
        if search_dir.exists():
            for md_file in search_dir.glob("*.md"):
                available_projects.add(md_file.stem)

    return sorted(list(available_projects))

log_exception(message, ctx=None) async

Log an error message to the context or logger.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def log_exception(message: str, ctx: Context | None = None):
    """Log an error message to the context or logger."""
    if ctx:
        await ctx.error(message)
    else:
        logger.error(message)
        raise Exception(message)

log_info(message, ctx=None) async

Log an info message to the context or logger.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def log_info(message: str, ctx: Context | None = None):
    """Log an info message to the context or logger."""
    if ctx:
        await ctx.info(message)
    else:
        logger.info(message)

log_warning(message, ctx=None) async

Log a warning message to the context or logger.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

async def log_warning(message: str, ctx: Context | None = None):
    """Log a warning message to the context or logger."""
    if ctx:
        await ctx.warning(message)
    else:
        logger.warning(message)

main()

Run the documentation indexer (full rebuild, called from legacy entry points).

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def main():
    """Run the documentation indexer (full rebuild, called from legacy entry points)."""
    DocumentationIndexer().run()

remove_leading_number_sep_from_path(p)

Remove a leading number + underscore or hyphen from the last path component.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def remove_leading_number_sep_from_path(p: Path) -> Path:
    """Remove a leading number + underscore or hyphen from the last path component."""
    new_name = re.sub(r"^\d+[_-]", "", p.name)
    return p.with_name(new_name)

truncate_content(content, max_chars, query=None)

Truncate content, optionally centering on query matches.

Args: content: The content to truncate max_chars: Maximum characters allowed. If None, no truncation is performed. query: Optional search query for context-aware truncation

Returns:

Type	Description
The original content if under limit, truncated content with ellipsis if over limit,	or None if content is None.

Source code in src/holoviz_mcp/holoviz_mcp/data.py

def truncate_content(content: str | None, max_chars: int | None, query: str | None = None) -> str | None:
    """Truncate content, optionally centering on query matches.

    Args:
        content: The content to truncate
        max_chars: Maximum characters allowed. If None, no truncation is performed.
        query: Optional search query for context-aware truncation

    Returns
    -------
        The original content if under limit, truncated content with ellipsis if over limit,
        or None if content is None.
    """
    if content is None or max_chars is None:
        return content

    if len(content) <= max_chars:
        return content

    # If query provided, use smart excerpt extraction
    if query:
        # Adjust context_chars based on max_chars to ensure keywords fit
        # Use at most 40% of max_chars for context on each side
        context_chars = min(500, int(max_chars * 0.4))
        return extract_relevant_excerpt(content, query, max_chars, context_chars)

    # Otherwise, use simple truncation (existing logic)
    truncated = content[:max_chars]
    last_space = truncated.rfind(" ")

    # Don't cut off too much - if last space is in the last 20% of max_chars, use it
    if last_space > max_chars * 0.8:
        truncated = truncated[:last_space]

    # Add indicator
    return truncated + "\n\n[... content truncated, use content='full' for complete content ...]"

Configuration

Configuration package for HoloViz MCP server.

ConfigLoader

Loads and manages HoloViz MCP configuration.

Source code in src/holoviz_mcp/config/loader.py

class ConfigLoader:
    """Loads and manages HoloViz MCP configuration."""

    def __init__(self, config: Optional[HoloVizMCPConfig] = None):
        """Initialize configuration loader.

        Args:
            config: Pre-configured HoloVizMCPConfig with environment paths.
                   If None, loads paths from environment. Configuration will
                   still be loaded from files even if this is provided.
        """
        self._env_config = config
        self._loaded_config: Optional[HoloVizMCPConfig] = None

    def load_config(self) -> HoloVizMCPConfig:
        """Load configuration from files and environment.

        Returns
        -------
            Loaded configuration.

        Raises
        ------
            ConfigurationError: If configuration cannot be loaded or is invalid.
        """
        if self._loaded_config is not None:
            return self._loaded_config

        # Get environment config (from parameter or environment)
        if self._env_config is not None:
            env_config = self._env_config
        else:
            env_config = HoloVizMCPConfig()

        # Start with default configuration dict
        config_dict = self._get_default_config()

        # Load from default config file if it exists
        default_config_file = env_config.default_dir / "config.yaml"
        if default_config_file.exists():
            try:
                default_config = self._load_yaml_file(default_config_file)
                config_dict = self._merge_configs(config_dict, default_config)
                logger.info(f"Loaded default configuration from {default_config_file}")
            except Exception as e:
                logger.warning(f"Failed to load default config from {default_config_file}: {e}")

        # Load from user config file if it exists
        user_config_file = env_config.config_file_path()
        if user_config_file.exists():
            user_config = self._load_yaml_file(user_config_file)
            # Filter out any unknown fields to prevent validation errors
            user_config = self._filter_known_fields(user_config)
            config_dict = self._merge_configs(config_dict, user_config)
            logger.info(f"Loaded user configuration from {user_config_file}")

        # Apply environment variable overrides
        config_dict = self._apply_env_overrides(config_dict)

        # Add the environment paths to the config dict
        config_dict.update(
            {
                "user_dir": env_config.user_dir,
                "default_dir": env_config.default_dir,
                "repos_dir": env_config.repos_dir,
            }
        )

        # Create the final configuration
        try:
            self._loaded_config = HoloVizMCPConfig(**config_dict)
        except ValidationError as e:
            raise ConfigurationError(f"Invalid configuration: {e}") from e

        return self._loaded_config

    def _filter_known_fields(self, config_dict: dict[str, Any]) -> dict[str, Any]:
        """Filter out unknown fields that aren't part of the HoloVizMCPConfig schema.

        This prevents validation errors when loading user config files that might
        contain extra fields.
        """
        known_fields = {"server", "docs", "resources", "prompts", "user_dir", "default_dir", "repos_dir"}
        return {k: v for k, v in config_dict.items() if k in known_fields}

    def _get_default_config(self) -> dict[str, Any]:
        """Get default configuration dictionary."""
        return {
            "server": {
                "name": "holoviz-mcp",
                "version": "1.0.0",
                "description": "Model Context Protocol server for HoloViz ecosystem",
                "log_level": "INFO",
            },
            "docs": {
                "repositories": {},  # No more Python-side defaults!
                "index_patterns": ["**/*.md", "**/*.rst", "**/*.txt"],
                "exclude_patterns": ["**/node_modules/**", "**/.git/**", "**/build/**"],
                "max_file_size": 1024 * 1024,  # 1MB
                "update_interval": 86400,  # 24 hours
            },
            "resources": {"search_paths": []},
            "prompts": {"search_paths": []},
        }

    def _load_yaml_file(self, file_path: Path) -> dict[str, Any]:
        """Load YAML file safely.

        Args:
            file_path: Path to YAML file.

        Returns
        -------
            Parsed YAML content.

        Raises
        ------
            ConfigurationError: If file cannot be loaded or parsed.
        """
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                content = yaml.safe_load(f)
                if content is None:
                    return {}
                if not isinstance(content, dict):
                    raise ConfigurationError(f"Configuration file must contain a YAML dictionary: {file_path}")
                return content
        except yaml.YAMLError as e:
            raise ConfigurationError(f"Invalid YAML in {file_path}: {e}") from e
        except Exception as e:
            raise ConfigurationError(f"Failed to load {file_path}: {e}") from e

    def _merge_configs(self, base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
        """Merge two configuration dictionaries recursively.

        Args:
            base: Base configuration.
            override: Override configuration.

        Returns
        -------
            Merged configuration.
        """
        result = base.copy()

        for key, value in override.items():
            if key in result and isinstance(result[key], dict) and isinstance(value, dict):
                result[key] = self._merge_configs(result[key], value)
            else:
                result[key] = value

        return result

    def _apply_env_overrides(self, config: dict[str, Any]) -> dict[str, Any]:
        """Apply environment variable overrides to configuration.

        Args:
            config: Configuration dictionary.

        Returns
        -------
            Configuration with environment overrides applied.
        """
        # Log level override
        if "HOLOVIZ_MCP_LOG_LEVEL" in os.environ:
            config.setdefault("server", {})["log_level"] = os.environ["HOLOVIZ_MCP_LOG_LEVEL"]

        # Server name override
        if "HOLOVIZ_MCP_SERVER_NAME" in os.environ:
            config.setdefault("server", {})["name"] = os.environ["HOLOVIZ_MCP_SERVER_NAME"]

        # Transport override
        if "HOLOVIZ_MCP_TRANSPORT" in os.environ:
            config.setdefault("server", {})["transport"] = os.environ["HOLOVIZ_MCP_TRANSPORT"]

        # Host override (for HTTP transport)
        if "HOLOVIZ_MCP_HOST" in os.environ:
            config.setdefault("server", {})["host"] = os.environ["HOLOVIZ_MCP_HOST"]

        # Port override (for HTTP transport)
        if "HOLOVIZ_MCP_PORT" in os.environ:
            port_str = os.environ["HOLOVIZ_MCP_PORT"]
            try:
                port = int(port_str)
                if not (1 <= port <= 65535):
                    raise ValueError(f"Port must be between 1 and 65535, got {port}")
                config.setdefault("server", {})["port"] = port
            except ValueError as e:
                raise ConfigurationError(f"Invalid HOLOVIZ_MCP_PORT: {port_str}") from e

        # Telemetry override
        if "ANONYMIZED_TELEMETRY" in os.environ:
            config.setdefault("server", {})["anonymized_telemetry"] = os.environ["ANONYMIZED_TELEMETRY"].lower() in ("true", "1", "yes", "on")

        # Jupyter proxy URL override
        if "JUPYTER_SERVER_PROXY_URL" in os.environ:
            config.setdefault("server", {})["jupyter_server_proxy_url"] = os.environ["JUPYTER_SERVER_PROXY_URL"]

        return config

    def get_repos_dir(self) -> Path:
        """Get the repository download directory."""
        config = self.load_config()
        return config.repos_dir

    def get_resources_dir(self) -> Path:
        """Get the resources directory."""
        config = self.load_config()
        return config.resources_dir()

    def get_agents_dir(self) -> Path:
        """Get the agents directory."""
        config = self.load_config()
        return config.agents_dir()

    def get_skills_dir(self) -> Path:
        """Get the skills directory."""
        config = self.load_config()
        return config.skills_dir()

    def create_default_user_config(self) -> None:
        """Create a default user configuration file."""
        config = self.load_config()
        config_file = config.config_file_path()

        # Create directories if they don't exist
        config_file.parent.mkdir(parents=True, exist_ok=True)

        # Don't overwrite existing config
        if config_file.exists():
            logger.info(f"Configuration file already exists: {config_file}")
            return

        # Create default configuration
        template = {
            "server": {
                "name": "holoviz-mcp",
                "log_level": "INFO",
            },
            "docs": {
                "repositories": {
                    "example-repo": {
                        "url": "https://github.com/example/repo.git",
                        "branch": "main",
                        "folders": {"doc": {"url_path": ""}},
                        "base_url": "https://example.readthedocs.io",
                        "reference_patterns": ["doc/reference/**/*.md", "examples/reference/**/*.ipynb"],
                    }
                }
            },
            "resources": {"search_paths": []},
            "prompts": {"search_paths": []},
        }

        with open(config_file, "w", encoding="utf-8") as f:
            yaml.dump(template, f, default_flow_style=False, sort_keys=False)

        logger.info(f"Created default user configuration file: {config_file}")

    def reload_config(self) -> HoloVizMCPConfig:
        """Reload configuration from files.

        Returns
        -------
            Reloaded configuration.
        """
        self._loaded_config = None
        return self.load_config()

    def clear_cache(self) -> None:
        """Clear the cached configuration to force reload on next access."""
        self._loaded_config = None

clear_cache()

Clear the cached configuration to force reload on next access.

Source code in src/holoviz_mcp/config/loader.py

def clear_cache(self) -> None:
    """Clear the cached configuration to force reload on next access."""
    self._loaded_config = None

create_default_user_config()

Create a default user configuration file.

Source code in src/holoviz_mcp/config/loader.py

def create_default_user_config(self) -> None:
    """Create a default user configuration file."""
    config = self.load_config()
    config_file = config.config_file_path()

    # Create directories if they don't exist
    config_file.parent.mkdir(parents=True, exist_ok=True)

    # Don't overwrite existing config
    if config_file.exists():
        logger.info(f"Configuration file already exists: {config_file}")
        return

    # Create default configuration
    template = {
        "server": {
            "name": "holoviz-mcp",
            "log_level": "INFO",
        },
        "docs": {
            "repositories": {
                "example-repo": {
                    "url": "https://github.com/example/repo.git",
                    "branch": "main",
                    "folders": {"doc": {"url_path": ""}},
                    "base_url": "https://example.readthedocs.io",
                    "reference_patterns": ["doc/reference/**/*.md", "examples/reference/**/*.ipynb"],
                }
            }
        },
        "resources": {"search_paths": []},
        "prompts": {"search_paths": []},
    }

    with open(config_file, "w", encoding="utf-8") as f:
        yaml.dump(template, f, default_flow_style=False, sort_keys=False)

    logger.info(f"Created default user configuration file: {config_file}")

get_agents_dir()

Get the agents directory.

Source code in src/holoviz_mcp/config/loader.py

def get_agents_dir(self) -> Path:
    """Get the agents directory."""
    config = self.load_config()
    return config.agents_dir()

get_repos_dir()

Get the repository download directory.

Source code in src/holoviz_mcp/config/loader.py

def get_repos_dir(self) -> Path:
    """Get the repository download directory."""
    config = self.load_config()
    return config.repos_dir

get_resources_dir()

Get the resources directory.

Source code in src/holoviz_mcp/config/loader.py

def get_resources_dir(self) -> Path:
    """Get the resources directory."""
    config = self.load_config()
    return config.resources_dir()

get_skills_dir()

Get the skills directory.

Source code in src/holoviz_mcp/config/loader.py

def get_skills_dir(self) -> Path:
    """Get the skills directory."""
    config = self.load_config()
    return config.skills_dir()

load_config()

Load configuration from files and environment.

Returns:

Type	Description
Loaded configuration.

Raises:

Type	Description
ConfigurationError: If configuration cannot be loaded or is invalid.

Source code in src/holoviz_mcp/config/loader.py

def load_config(self) -> HoloVizMCPConfig:
    """Load configuration from files and environment.

    Returns
    -------
        Loaded configuration.

    Raises
    ------
        ConfigurationError: If configuration cannot be loaded or is invalid.
    """
    if self._loaded_config is not None:
        return self._loaded_config

    # Get environment config (from parameter or environment)
    if self._env_config is not None:
        env_config = self._env_config
    else:
        env_config = HoloVizMCPConfig()

    # Start with default configuration dict
    config_dict = self._get_default_config()

    # Load from default config file if it exists
    default_config_file = env_config.default_dir / "config.yaml"
    if default_config_file.exists():
        try:
            default_config = self._load_yaml_file(default_config_file)
            config_dict = self._merge_configs(config_dict, default_config)
            logger.info(f"Loaded default configuration from {default_config_file}")
        except Exception as e:
            logger.warning(f"Failed to load default config from {default_config_file}: {e}")

    # Load from user config file if it exists
    user_config_file = env_config.config_file_path()
    if user_config_file.exists():
        user_config = self._load_yaml_file(user_config_file)
        # Filter out any unknown fields to prevent validation errors
        user_config = self._filter_known_fields(user_config)
        config_dict = self._merge_configs(config_dict, user_config)
        logger.info(f"Loaded user configuration from {user_config_file}")

    # Apply environment variable overrides
    config_dict = self._apply_env_overrides(config_dict)

    # Add the environment paths to the config dict
    config_dict.update(
        {
            "user_dir": env_config.user_dir,
            "default_dir": env_config.default_dir,
            "repos_dir": env_config.repos_dir,
        }
    )

    # Create the final configuration
    try:
        self._loaded_config = HoloVizMCPConfig(**config_dict)
    except ValidationError as e:
        raise ConfigurationError(f"Invalid configuration: {e}") from e

    return self._loaded_config

reload_config()

Reload configuration from files.

Returns:

Type	Description
Reloaded configuration.

Source code in src/holoviz_mcp/config/loader.py

def reload_config(self) -> HoloVizMCPConfig:
    """Reload configuration from files.

    Returns
    -------
        Reloaded configuration.
    """
    self._loaded_config = None
    return self.load_config()

ConfigurationError

Bases: Exception

Raised when configuration cannot be loaded or is invalid.

Source code in src/holoviz_mcp/config/loader.py

class ConfigurationError(Exception):
    """Raised when configuration cannot be loaded or is invalid."""

DocsConfig

Bases: BaseModel

Configuration for documentation repositories.

Source code in src/holoviz_mcp/config/models.py

class DocsConfig(BaseModel):
    """Configuration for documentation repositories."""

    repositories: dict[str, GitRepository] = Field(default_factory=dict, description="Dictionary mapping package names to repository configurations")
    index_patterns: list[str] = Field(
        default_factory=lambda: ["**/*.md", "**/*.rst", "**/*.txt"], description="File patterns to include when indexing documentation"
    )
    exclude_patterns: list[str] = Field(
        default_factory=lambda: ["**/node_modules/**", "**/.git/**", "**/build/**"], description="File patterns to exclude when indexing documentation"
    )
    max_file_size: PositiveInt = Field(
        default=1024 * 1024,  # 1MB
        description="Maximum file size in bytes to index",
    )
    update_interval: PositiveInt = Field(
        default=86400,  # 24 hours
        description="How often to check for updates in seconds",
    )

exclude_patterns = Field(default_factory=(lambda: ['**/node_modules/**', '**/.git/**', '**/build/**']), description='File patterns to exclude when indexing documentation') class-attribute instance-attribute

index_patterns = Field(default_factory=(lambda: ['**/*.md', '**/*.rst', '**/*.txt']), description='File patterns to include when indexing documentation') class-attribute instance-attribute

max_file_size = Field(default=(1024 * 1024), description='Maximum file size in bytes to index') class-attribute instance-attribute

repositories = Field(default_factory=dict, description='Dictionary mapping package names to repository configurations') class-attribute instance-attribute

update_interval = Field(default=86400, description='How often to check for updates in seconds') class-attribute instance-attribute

GitRepository

Bases: BaseModel

Configuration for a Git repository.

Source code in src/holoviz_mcp/config/models.py

class GitRepository(BaseModel):
    """Configuration for a Git repository."""

    url: AnyHttpUrl = Field(..., description="Git repository URL")
    branch: Optional[str] = Field(default=None, description="Git branch to use")
    tag: Optional[str] = Field(default=None, description="Git tag to use (e.g., '1.7.2')")
    commit: Optional[str] = Field(default=None, description="Git commit hash to use")
    folders: Union[list[str], dict[str, FolderConfig]] = Field(
        default_factory=lambda: {"doc": FolderConfig()},
        description="Folders to index within the repository. Can be a list of folder names or a dict mapping folder names to FolderConfig objects",
    )
    base_url: AnyHttpUrl = Field(..., description="Base URL for documentation links")
    url_transform: Literal["holoviz", "plotly", "datashader"] = Field(
        default="holoviz",
        description="""How to transform file path into URL:

        - holoViz transform suffix to .html: filename.md -> filename.html
        - plotly transform suffix to /: filename.md -> filename/
        - datashader removes leading index and transform suffix to .html: 01_filename.md -> filename.html
        """,
    )
    reference_patterns: list[str] = Field(
        default_factory=lambda: ["examples/reference/**/*.md", "examples/reference/**/*.ipynb"], description="Glob patterns for reference documentation files"
    )

    @field_validator("tag")
    @classmethod
    def validate_tag(cls, v):
        """Validate git tag format, allowing both 'v1.2.3' and '1.2.3' formats."""
        if v is not None and v.startswith("v"):
            # Allow tags like 'v1.7.2' but also suggest plain version numbers
            pass
        return v

    @field_validator("folders")
    @classmethod
    def validate_folders(cls, v):
        """Validate and normalize folders configuration."""
        if isinstance(v, list):
            # Convert list to dict format for backward compatibility
            return {folder: FolderConfig() for folder in v}
        elif isinstance(v, dict):
            # Ensure all values are FolderConfig objects
            result = {}
            for folder, config in v.items():
                if isinstance(config, dict):
                    result[folder] = FolderConfig(**config)
                elif isinstance(config, FolderConfig):
                    result[folder] = config
                else:
                    raise ValueError(f"Invalid folder config for '{folder}': must be dict or FolderConfig")
            return result
        else:
            raise ValueError("folders must be a list or dict")

    def get_folder_names(self) -> list[str]:
        """Get list of folder names for backward compatibility."""
        if isinstance(self.folders, dict):
            return list(self.folders.keys())
        return self.folders

    def get_folder_url_path(self, folder_name: str) -> str:
        """Get URL path for a specific folder."""
        if isinstance(self.folders, dict):
            folder_config = self.folders.get(folder_name)
            if folder_config:
                return folder_config.url_path
        return ""

base_url = Field(..., description='Base URL for documentation links') class-attribute instance-attribute

branch = Field(default=None, description='Git branch to use') class-attribute instance-attribute

commit = Field(default=None, description='Git commit hash to use') class-attribute instance-attribute

folders = Field(default_factory=(lambda: {'doc': FolderConfig()}), description='Folders to index within the repository. Can be a list of folder names or a dict mapping folder names to FolderConfig objects') class-attribute instance-attribute

reference_patterns = Field(default_factory=(lambda: ['examples/reference/**/*.md', 'examples/reference/**/*.ipynb']), description='Glob patterns for reference documentation files') class-attribute instance-attribute

tag = Field(default=None, description="Git tag to use (e.g., '1.7.2')") class-attribute instance-attribute

url = Field(..., description='Git repository URL') class-attribute instance-attribute

url_transform = Field(default='holoviz', description='How to transform file path into URL:\n\n - holoViz transform suffix to .html: filename.md -> filename.html\n - plotly transform suffix to /: filename.md -> filename/\n - datashader removes leading index and transform suffix to .html: 01_filename.md -> filename.html\n ') class-attribute instance-attribute

get_folder_names()

Get list of folder names for backward compatibility.

Source code in src/holoviz_mcp/config/models.py

def get_folder_names(self) -> list[str]:
    """Get list of folder names for backward compatibility."""
    if isinstance(self.folders, dict):
        return list(self.folders.keys())
    return self.folders

get_folder_url_path(folder_name)

Get URL path for a specific folder.

Source code in src/holoviz_mcp/config/models.py

def get_folder_url_path(self, folder_name: str) -> str:
    """Get URL path for a specific folder."""
    if isinstance(self.folders, dict):
        folder_config = self.folders.get(folder_name)
        if folder_config:
            return folder_config.url_path
    return ""

validate_folders(v) classmethod

Validate and normalize folders configuration.

Source code in src/holoviz_mcp/config/models.py

@field_validator("folders")
@classmethod
def validate_folders(cls, v):
    """Validate and normalize folders configuration."""
    if isinstance(v, list):
        # Convert list to dict format for backward compatibility
        return {folder: FolderConfig() for folder in v}
    elif isinstance(v, dict):
        # Ensure all values are FolderConfig objects
        result = {}
        for folder, config in v.items():
            if isinstance(config, dict):
                result[folder] = FolderConfig(**config)
            elif isinstance(config, FolderConfig):
                result[folder] = config
            else:
                raise ValueError(f"Invalid folder config for '{folder}': must be dict or FolderConfig")
        return result
    else:
        raise ValueError("folders must be a list or dict")

validate_tag(v) classmethod

Validate git tag format, allowing both 'v1.2.3' and '1.2.3' formats.

Source code in src/holoviz_mcp/config/models.py

@field_validator("tag")
@classmethod
def validate_tag(cls, v):
    """Validate git tag format, allowing both 'v1.2.3' and '1.2.3' formats."""
    if v is not None and v.startswith("v"):
        # Allow tags like 'v1.7.2' but also suggest plain version numbers
        pass
    return v

HoloVizMCPConfig

Bases: BaseModel

Main configuration for HoloViz MCP server.

Source code in src/holoviz_mcp/config/models.py

class HoloVizMCPConfig(BaseModel):
    """Main configuration for HoloViz MCP server."""

    server: ServerConfig = Field(default_factory=ServerConfig)
    docs: DocsConfig = Field(default_factory=DocsConfig)
    resources: ResourceConfig = Field(default_factory=ResourceConfig)
    prompts: PromptConfig = Field(default_factory=PromptConfig)
    display: DisplayConfig = Field(default_factory=DisplayConfig)

    # Environment paths - merged from EnvironmentConfig with defaults
    user_dir: Path = Field(default_factory=_holoviz_mcp_user_dir, description="User configuration directory")
    default_dir: Path = Field(default_factory=_holoviz_mcp_default_dir, description="Default configuration directory")
    repos_dir: Path = Field(default_factory=lambda: _holoviz_mcp_user_dir() / "repos", description="Repository download directory")

    model_config = ConfigDict(extra="forbid", validate_assignment=True)

    def config_file_path(self, location: Literal["user", "default"] = "user") -> Path:
        """Get the path to the configuration file.

        Args:
            location: Whether to get user or default config file path
        """
        if location == "user":
            return self.user_dir / "config.yaml"
        else:
            return self.default_dir / "config.yaml"

    def resources_dir(self, location: Literal["user", "default"] = "user") -> Path:
        """Get the path to the resources directory.

        Args:
            location: Whether to get user or default resources directory
        """
        if location == "user":
            return self.user_dir / "resources"
        else:
            return self.default_dir / "resources"

    def agents_dir(self, location: Literal["user", "default"] = "user", tool: Literal["copilot", "claude"] | None = None) -> Path:
        """Get the path to the agents directory.

        Args:
            location: Whether to get user or default agents directory
            tool: Optional tool-specific subdirectory (e.g., "copilot", "claude")

        Returns
        -------
            Path to agents directory, optionally scoped to a specific tool
        """
        base_dir = self.resources_dir(location) / "agents"
        if tool:
            return base_dir / tool
        return base_dir

    def skills_dir(self, location: Literal["user", "default"] = "user") -> Path:
        """Get the path to the skills directory.

        Args:
            location: Whether to get user or default skills directory
        """
        return self.resources_dir(location) / "skills"

default_dir = Field(default_factory=_holoviz_mcp_default_dir, description='Default configuration directory') class-attribute instance-attribute

display = Field(default_factory=DisplayConfig) class-attribute instance-attribute

docs = Field(default_factory=DocsConfig) class-attribute instance-attribute

model_config = ConfigDict(extra='forbid', validate_assignment=True) class-attribute instance-attribute

prompts = Field(default_factory=PromptConfig) class-attribute instance-attribute

repos_dir = Field(default_factory=(lambda: _holoviz_mcp_user_dir() / 'repos'), description='Repository download directory') class-attribute instance-attribute

resources = Field(default_factory=ResourceConfig) class-attribute instance-attribute

server = Field(default_factory=ServerConfig) class-attribute instance-attribute

user_dir = Field(default_factory=_holoviz_mcp_user_dir, description='User configuration directory') class-attribute instance-attribute

agents_dir(location='user', tool=None)

Get the path to the agents directory.

Args: location: Whether to get user or default agents directory tool: Optional tool-specific subdirectory (e.g., "copilot", "claude")

Returns:

Type	Description
Path to agents directory, optionally scoped to a specific tool

Source code in src/holoviz_mcp/config/models.py

def agents_dir(self, location: Literal["user", "default"] = "user", tool: Literal["copilot", "claude"] | None = None) -> Path:
    """Get the path to the agents directory.

    Args:
        location: Whether to get user or default agents directory
        tool: Optional tool-specific subdirectory (e.g., "copilot", "claude")

    Returns
    -------
        Path to agents directory, optionally scoped to a specific tool
    """
    base_dir = self.resources_dir(location) / "agents"
    if tool:
        return base_dir / tool
    return base_dir

config_file_path(location='user')

Get the path to the configuration file.

Args: location: Whether to get user or default config file path

Source code in src/holoviz_mcp/config/models.py

def config_file_path(self, location: Literal["user", "default"] = "user") -> Path:
    """Get the path to the configuration file.

    Args:
        location: Whether to get user or default config file path
    """
    if location == "user":
        return self.user_dir / "config.yaml"
    else:
        return self.default_dir / "config.yaml"

resources_dir(location='user')

Get the path to the resources directory.

Args: location: Whether to get user or default resources directory

Source code in src/holoviz_mcp/config/models.py

def resources_dir(self, location: Literal["user", "default"] = "user") -> Path:
    """Get the path to the resources directory.

    Args:
        location: Whether to get user or default resources directory
    """
    if location == "user":
        return self.user_dir / "resources"
    else:
        return self.default_dir / "resources"

skills_dir(location='user')

Get the path to the skills directory.

Args: location: Whether to get user or default skills directory

Source code in src/holoviz_mcp/config/models.py

def skills_dir(self, location: Literal["user", "default"] = "user") -> Path:
    """Get the path to the skills directory.

    Args:
        location: Whether to get user or default skills directory
    """
    return self.resources_dir(location) / "skills"

PromptConfig

Bases: BaseModel

Configuration for prompts.

Source code in src/holoviz_mcp/config/models.py

class PromptConfig(BaseModel):
    """Configuration for prompts."""

    search_paths: list[Path] = Field(default_factory=list, description="Additional paths to search for prompts")

search_paths = Field(default_factory=list, description='Additional paths to search for prompts') class-attribute instance-attribute

ResourceConfig

Bases: BaseModel

Configuration for resources (agents, skills, etc.).

Source code in src/holoviz_mcp/config/models.py

class ResourceConfig(BaseModel):
    """Configuration for resources (agents, skills, etc.)."""

    search_paths: list[Path] = Field(default_factory=list, description="Additional paths to search for resources")

search_paths = Field(default_factory=list, description='Additional paths to search for resources') class-attribute instance-attribute

ServerConfig

Bases: BaseModel

Configuration for the MCP server.

Source code in src/holoviz_mcp/config/models.py

class ServerConfig(BaseModel):
    """Configuration for the MCP server."""

    name: str = Field(default="holoviz-mcp", description="Server name")
    version: str = Field(default="1.0.0", description="Server version")
    description: str = Field(default="Model Context Protocol server for HoloViz ecosystem", description="Server description")
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(default="INFO", description="Logging level")
    transport: Literal["stdio", "http"] = Field(default="stdio", description="Transport protocol for MCP communication")
    host: str = Field(default="127.0.0.1", description="Host address to bind to when using HTTP transport (use 0.0.0.0 for Docker)")
    port: int = Field(default=8000, description="Port to bind to when using HTTP transport")
    anonymized_telemetry: bool = Field(default=False, description="Enable anonymized telemetry")
    jupyter_server_proxy_url: str = Field(default="", description="Jupyter server proxy URL for Panel app integration")
    vector_db_path: Path = Field(
        default_factory=lambda: (_holoviz_mcp_user_dir() / "vector_db" / "chroma").expanduser(), description="Path to the Chroma vector database."
    )
    screenshots_dir: Path = Field(
        default_factory=lambda: (_holoviz_mcp_user_dir() / "screenshots").expanduser(), description="Directory for saving screenshots from panel_inspect_app tool."
    )

anonymized_telemetry = Field(default=False, description='Enable anonymized telemetry') class-attribute instance-attribute

description = Field(default='Model Context Protocol server for HoloViz ecosystem', description='Server description') class-attribute instance-attribute

host = Field(default='127.0.0.1', description='Host address to bind to when using HTTP transport (use 0.0.0.0 for Docker)') class-attribute instance-attribute

jupyter_server_proxy_url = Field(default='', description='Jupyter server proxy URL for Panel app integration') class-attribute instance-attribute

log_level = Field(default='INFO', description='Logging level') class-attribute instance-attribute

name = Field(default='holoviz-mcp', description='Server name') class-attribute instance-attribute

port = Field(default=8000, description='Port to bind to when using HTTP transport') class-attribute instance-attribute

screenshots_dir = Field(default_factory=(lambda: (_holoviz_mcp_user_dir() / 'screenshots').expanduser()), description='Directory for saving screenshots from panel_inspect_app tool.') class-attribute instance-attribute

transport = Field(default='stdio', description='Transport protocol for MCP communication') class-attribute instance-attribute

vector_db_path = Field(default_factory=(lambda: (_holoviz_mcp_user_dir() / 'vector_db' / 'chroma').expanduser()), description='Path to the Chroma vector database.') class-attribute instance-attribute

version = Field(default='1.0.0', description='Server version') class-attribute instance-attribute

get_config()

Get the current configuration.

Source code in src/holoviz_mcp/config/loader.py

def get_config() -> HoloVizMCPConfig:
    """Get the current configuration."""
    return get_config_loader().load_config()

get_config_loader()

Get the global configuration loader instance.

Source code in src/holoviz_mcp/config/loader.py

def get_config_loader() -> ConfigLoader:
    """Get the global configuration loader instance."""
    global _config_loader
    if _config_loader is None:
        _config_loader = ConfigLoader()
    return _config_loader

reload_config()

Reload configuration from files.

Source code in src/holoviz_mcp/config/loader.py

def reload_config() -> HoloVizMCPConfig:
    """Reload configuration from files."""
    return get_config_loader().reload_config()

Loader

Configuration loader for HoloViz MCP server.

logger = logging.getLogger(__name__) module-attribute

ConfigLoader

Loads and manages HoloViz MCP configuration.

Source code in src/holoviz_mcp/config/loader.py

class ConfigLoader:
    """Loads and manages HoloViz MCP configuration."""

    def __init__(self, config: Optional[HoloVizMCPConfig] = None):
        """Initialize configuration loader.

        Args:
            config: Pre-configured HoloVizMCPConfig with environment paths.
                   If None, loads paths from environment. Configuration will
                   still be loaded from files even if this is provided.
        """
        self._env_config = config
        self._loaded_config: Optional[HoloVizMCPConfig] = None

    def load_config(self) -> HoloVizMCPConfig:
        """Load configuration from files and environment.

        Returns
        -------
            Loaded configuration.

        Raises
        ------
            ConfigurationError: If configuration cannot be loaded or is invalid.
        """
        if self._loaded_config is not None:
            return self._loaded_config

        # Get environment config (from parameter or environment)
        if self._env_config is not None:
            env_config = self._env_config
        else:
            env_config = HoloVizMCPConfig()

        # Start with default configuration dict
        config_dict = self._get_default_config()

        # Load from default config file if it exists
        default_config_file = env_config.default_dir / "config.yaml"
        if default_config_file.exists():
            try:
                default_config = self._load_yaml_file(default_config_file)
                config_dict = self._merge_configs(config_dict, default_config)
                logger.info(f"Loaded default configuration from {default_config_file}")
            except Exception as e:
                logger.warning(f"Failed to load default config from {default_config_file}: {e}")

        # Load from user config file if it exists
        user_config_file = env_config.config_file_path()
        if user_config_file.exists():
            user_config = self._load_yaml_file(user_config_file)
            # Filter out any unknown fields to prevent validation errors
            user_config = self._filter_known_fields(user_config)
            config_dict = self._merge_configs(config_dict, user_config)
            logger.info(f"Loaded user configuration from {user_config_file}")

        # Apply environment variable overrides
        config_dict = self._apply_env_overrides(config_dict)

        # Add the environment paths to the config dict
        config_dict.update(
            {
                "user_dir": env_config.user_dir,
                "default_dir": env_config.default_dir,
                "repos_dir": env_config.repos_dir,
            }
        )

        # Create the final configuration
        try:
            self._loaded_config = HoloVizMCPConfig(**config_dict)
        except ValidationError as e:
            raise ConfigurationError(f"Invalid configuration: {e}") from e

        return self._loaded_config

    def _filter_known_fields(self, config_dict: dict[str, Any]) -> dict[str, Any]:
        """Filter out unknown fields that aren't part of the HoloVizMCPConfig schema.

        This prevents validation errors when loading user config files that might
        contain extra fields.
        """
        known_fields = {"server", "docs", "resources", "prompts", "user_dir", "default_dir", "repos_dir"}
        return {k: v for k, v in config_dict.items() if k in known_fields}

    def _get_default_config(self) -> dict[str, Any]:
        """Get default configuration dictionary."""
        return {
            "server": {
                "name": "holoviz-mcp",
                "version": "1.0.0",
                "description": "Model Context Protocol server for HoloViz ecosystem",
                "log_level": "INFO",
            },
            "docs": {
                "repositories": {},  # No more Python-side defaults!
                "index_patterns": ["**/*.md", "**/*.rst", "**/*.txt"],
                "exclude_patterns": ["**/node_modules/**", "**/.git/**", "**/build/**"],
                "max_file_size": 1024 * 1024,  # 1MB
                "update_interval": 86400,  # 24 hours
            },
            "resources": {"search_paths": []},
            "prompts": {"search_paths": []},
        }

    def _load_yaml_file(self, file_path: Path) -> dict[str, Any]:
        """Load YAML file safely.

        Args:
            file_path: Path to YAML file.

        Returns
        -------
            Parsed YAML content.

        Raises
        ------
            ConfigurationError: If file cannot be loaded or parsed.
        """
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                content = yaml.safe_load(f)
                if content is None:
                    return {}
                if not isinstance(content, dict):
                    raise ConfigurationError(f"Configuration file must contain a YAML dictionary: {file_path}")
                return content
        except yaml.YAMLError as e:
            raise ConfigurationError(f"Invalid YAML in {file_path}: {e}") from e
        except Exception as e:
            raise ConfigurationError(f"Failed to load {file_path}: {e}") from e

    def _merge_configs(self, base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
        """Merge two configuration dictionaries recursively.

        Args:
            base: Base configuration.
            override: Override configuration.

        Returns
        -------
            Merged configuration.
        """
        result = base.copy()

        for key, value in override.items():
            if key in result and isinstance(result[key], dict) and isinstance(value, dict):
                result[key] = self._merge_configs(result[key], value)
            else:
                result[key] = value

        return result

    def _apply_env_overrides(self, config: dict[str, Any]) -> dict[str, Any]:
        """Apply environment variable overrides to configuration.

        Args:
            config: Configuration dictionary.

        Returns
        -------
            Configuration with environment overrides applied.
        """
        # Log level override
        if "HOLOVIZ_MCP_LOG_LEVEL" in os.environ:
            config.setdefault("server", {})["log_level"] = os.environ["HOLOVIZ_MCP_LOG_LEVEL"]

        # Server name override
        if "HOLOVIZ_MCP_SERVER_NAME" in os.environ:
            config.setdefault("server", {})["name"] = os.environ["HOLOVIZ_MCP_SERVER_NAME"]

        # Transport override
        if "HOLOVIZ_MCP_TRANSPORT" in os.environ:
            config.setdefault("server", {})["transport"] = os.environ["HOLOVIZ_MCP_TRANSPORT"]

        # Host override (for HTTP transport)
        if "HOLOVIZ_MCP_HOST" in os.environ:
            config.setdefault("server", {})["host"] = os.environ["HOLOVIZ_MCP_HOST"]

        # Port override (for HTTP transport)
        if "HOLOVIZ_MCP_PORT" in os.environ:
            port_str = os.environ["HOLOVIZ_MCP_PORT"]
            try:
                port = int(port_str)
                if not (1 <= port <= 65535):
                    raise ValueError(f"Port must be between 1 and 65535, got {port}")
                config.setdefault("server", {})["port"] = port
            except ValueError as e:
                raise ConfigurationError(f"Invalid HOLOVIZ_MCP_PORT: {port_str}") from e

        # Telemetry override
        if "ANONYMIZED_TELEMETRY" in os.environ:
            config.setdefault("server", {})["anonymized_telemetry"] = os.environ["ANONYMIZED_TELEMETRY"].lower() in ("true", "1", "yes", "on")

        # Jupyter proxy URL override
        if "JUPYTER_SERVER_PROXY_URL" in os.environ:
            config.setdefault("server", {})["jupyter_server_proxy_url"] = os.environ["JUPYTER_SERVER_PROXY_URL"]

        return config

    def get_repos_dir(self) -> Path:
        """Get the repository download directory."""
        config = self.load_config()
        return config.repos_dir

    def get_resources_dir(self) -> Path:
        """Get the resources directory."""
        config = self.load_config()
        return config.resources_dir()

    def get_agents_dir(self) -> Path:
        """Get the agents directory."""
        config = self.load_config()
        return config.agents_dir()

    def get_skills_dir(self) -> Path:
        """Get the skills directory."""
        config = self.load_config()
        return config.skills_dir()

    def create_default_user_config(self) -> None:
        """Create a default user configuration file."""
        config = self.load_config()
        config_file = config.config_file_path()

        # Create directories if they don't exist
        config_file.parent.mkdir(parents=True, exist_ok=True)

        # Don't overwrite existing config
        if config_file.exists():
            logger.info(f"Configuration file already exists: {config_file}")
            return

        # Create default configuration
        template = {
            "server": {
                "name": "holoviz-mcp",
                "log_level": "INFO",
            },
            "docs": {
                "repositories": {
                    "example-repo": {
                        "url": "https://github.com/example/repo.git",
                        "branch": "main",
                        "folders": {"doc": {"url_path": ""}},
                        "base_url": "https://example.readthedocs.io",
                        "reference_patterns": ["doc/reference/**/*.md", "examples/reference/**/*.ipynb"],
                    }
                }
            },
            "resources": {"search_paths": []},
            "prompts": {"search_paths": []},
        }

        with open(config_file, "w", encoding="utf-8") as f:
            yaml.dump(template, f, default_flow_style=False, sort_keys=False)

        logger.info(f"Created default user configuration file: {config_file}")

    def reload_config(self) -> HoloVizMCPConfig:
        """Reload configuration from files.

        Returns
        -------
            Reloaded configuration.
        """
        self._loaded_config = None
        return self.load_config()

    def clear_cache(self) -> None:
        """Clear the cached configuration to force reload on next access."""
        self._loaded_config = None

clear_cache()

Clear the cached configuration to force reload on next access.

Source code in src/holoviz_mcp/config/loader.py

def clear_cache(self) -> None:
    """Clear the cached configuration to force reload on next access."""
    self._loaded_config = None

create_default_user_config()

Create a default user configuration file.

Source code in src/holoviz_mcp/config/loader.py

def create_default_user_config(self) -> None:
    """Create a default user configuration file."""
    config = self.load_config()
    config_file = config.config_file_path()

    # Create directories if they don't exist
    config_file.parent.mkdir(parents=True, exist_ok=True)

    # Don't overwrite existing config
    if config_file.exists():
        logger.info(f"Configuration file already exists: {config_file}")
        return

    # Create default configuration
    template = {
        "server": {
            "name": "holoviz-mcp",
            "log_level": "INFO",
        },
        "docs": {
            "repositories": {
                "example-repo": {
                    "url": "https://github.com/example/repo.git",
                    "branch": "main",
                    "folders": {"doc": {"url_path": ""}},
                    "base_url": "https://example.readthedocs.io",
                    "reference_patterns": ["doc/reference/**/*.md", "examples/reference/**/*.ipynb"],
                }
            }
        },
        "resources": {"search_paths": []},
        "prompts": {"search_paths": []},
    }

    with open(config_file, "w", encoding="utf-8") as f:
        yaml.dump(template, f, default_flow_style=False, sort_keys=False)

    logger.info(f"Created default user configuration file: {config_file}")

get_agents_dir()

Get the agents directory.

Source code in src/holoviz_mcp/config/loader.py

def get_agents_dir(self) -> Path:
    """Get the agents directory."""
    config = self.load_config()
    return config.agents_dir()

get_repos_dir()

Get the repository download directory.

Source code in src/holoviz_mcp/config/loader.py

def get_repos_dir(self) -> Path:
    """Get the repository download directory."""
    config = self.load_config()
    return config.repos_dir

get_resources_dir()

Get the resources directory.

Source code in src/holoviz_mcp/config/loader.py

def get_resources_dir(self) -> Path:
    """Get the resources directory."""
    config = self.load_config()
    return config.resources_dir()

get_skills_dir()

Get the skills directory.

Source code in src/holoviz_mcp/config/loader.py

def get_skills_dir(self) -> Path:
    """Get the skills directory."""
    config = self.load_config()
    return config.skills_dir()

load_config()

Load configuration from files and environment.

Returns:

Type	Description
Loaded configuration.

Raises:

Type	Description
ConfigurationError: If configuration cannot be loaded or is invalid.

Source code in src/holoviz_mcp/config/loader.py

def load_config(self) -> HoloVizMCPConfig:
    """Load configuration from files and environment.

    Returns
    -------
        Loaded configuration.

    Raises
    ------
        ConfigurationError: If configuration cannot be loaded or is invalid.
    """
    if self._loaded_config is not None:
        return self._loaded_config

    # Get environment config (from parameter or environment)
    if self._env_config is not None:
        env_config = self._env_config
    else:
        env_config = HoloVizMCPConfig()

    # Start with default configuration dict
    config_dict = self._get_default_config()

    # Load from default config file if it exists
    default_config_file = env_config.default_dir / "config.yaml"
    if default_config_file.exists():
        try:
            default_config = self._load_yaml_file(default_config_file)
            config_dict = self._merge_configs(config_dict, default_config)
            logger.info(f"Loaded default configuration from {default_config_file}")
        except Exception as e:
            logger.warning(f"Failed to load default config from {default_config_file}: {e}")

    # Load from user config file if it exists
    user_config_file = env_config.config_file_path()
    if user_config_file.exists():
        user_config = self._load_yaml_file(user_config_file)
        # Filter out any unknown fields to prevent validation errors
        user_config = self._filter_known_fields(user_config)
        config_dict = self._merge_configs(config_dict, user_config)
        logger.info(f"Loaded user configuration from {user_config_file}")

    # Apply environment variable overrides
    config_dict = self._apply_env_overrides(config_dict)

    # Add the environment paths to the config dict
    config_dict.update(
        {
            "user_dir": env_config.user_dir,
            "default_dir": env_config.default_dir,
            "repos_dir": env_config.repos_dir,
        }
    )

    # Create the final configuration
    try:
        self._loaded_config = HoloVizMCPConfig(**config_dict)
    except ValidationError as e:
        raise ConfigurationError(f"Invalid configuration: {e}") from e

    return self._loaded_config

reload_config()

Reload configuration from files.

Returns:

Type	Description
Reloaded configuration.

Source code in src/holoviz_mcp/config/loader.py

def reload_config(self) -> HoloVizMCPConfig:
    """Reload configuration from files.

    Returns
    -------
        Reloaded configuration.
    """
    self._loaded_config = None
    return self.load_config()

ConfigurationError

Bases: Exception

Raised when configuration cannot be loaded or is invalid.

Source code in src/holoviz_mcp/config/loader.py

class ConfigurationError(Exception):
    """Raised when configuration cannot be loaded or is invalid."""

get_config()

Get the current configuration.

Source code in src/holoviz_mcp/config/loader.py

def get_config() -> HoloVizMCPConfig:
    """Get the current configuration."""
    return get_config_loader().load_config()

get_config_loader()

Get the global configuration loader instance.

Source code in src/holoviz_mcp/config/loader.py

def get_config_loader() -> ConfigLoader:
    """Get the global configuration loader instance."""
    global _config_loader
    if _config_loader is None:
        _config_loader = ConfigLoader()
    return _config_loader

reload_config()

Reload configuration from files.

Source code in src/holoviz_mcp/config/loader.py

def reload_config() -> HoloVizMCPConfig:
    """Reload configuration from files."""
    return get_config_loader().reload_config()

Models

Configuration models for HoloViz MCP server.

DisplayConfig

Bases: BaseModel

Configuration for the AI Visualizer display server.

Source code in src/holoviz_mcp/config/models.py

class DisplayConfig(BaseModel):
    """Configuration for the AI Visualizer display server."""

    enabled: bool = Field(default=True, description="Enable the display server")
    mode: Literal["subprocess", "remote"] = Field(
        default="subprocess",
        description="Display server mode: 'subprocess' to manage Panel server as subprocess, 'remote' to connect to existing server",
    )
    server_url: Optional[str] = Field(
        default=None,
        description="URL of remote display server (e.g., 'http://localhost:5005'). Only used in 'remote' mode.",
    )
    port: int = Field(default=5005, description="Port for the display Panel server (subprocess mode only)")
    host: str = Field(default="localhost", description="Host address for the display Panel server (subprocess mode only)")
    max_restarts: int = Field(default=3, description="Maximum number of restart attempts for Panel server (subprocess mode only)")
    health_check_interval: int = Field(default=60, description="Health check interval in seconds")
    db_path: Path = Field(
        default_factory=lambda: _holoviz_mcp_user_dir() / "snippets" / "snippets.db",
        description="Path to SQLite database for display requests",
    )

db_path = Field(default_factory=(lambda: _holoviz_mcp_user_dir() / 'snippets' / 'snippets.db'), description='Path to SQLite database for display requests') class-attribute instance-attribute

enabled = Field(default=True, description='Enable the display server') class-attribute instance-attribute

health_check_interval = Field(default=60, description='Health check interval in seconds') class-attribute instance-attribute

host = Field(default='localhost', description='Host address for the display Panel server (subprocess mode only)') class-attribute instance-attribute

max_restarts = Field(default=3, description='Maximum number of restart attempts for Panel server (subprocess mode only)') class-attribute instance-attribute

mode = Field(default='subprocess', description="Display server mode: 'subprocess' to manage Panel server as subprocess, 'remote' to connect to existing server") class-attribute instance-attribute

port = Field(default=5005, description='Port for the display Panel server (subprocess mode only)') class-attribute instance-attribute

server_url = Field(default=None, description="URL of remote display server (e.g., 'http://localhost:5005'). Only used in 'remote' mode.") class-attribute instance-attribute

DocsConfig

Bases: BaseModel

Configuration for documentation repositories.

Source code in src/holoviz_mcp/config/models.py

class DocsConfig(BaseModel):
    """Configuration for documentation repositories."""

    repositories: dict[str, GitRepository] = Field(default_factory=dict, description="Dictionary mapping package names to repository configurations")
    index_patterns: list[str] = Field(
        default_factory=lambda: ["**/*.md", "**/*.rst", "**/*.txt"], description="File patterns to include when indexing documentation"
    )
    exclude_patterns: list[str] = Field(
        default_factory=lambda: ["**/node_modules/**", "**/.git/**", "**/build/**"], description="File patterns to exclude when indexing documentation"
    )
    max_file_size: PositiveInt = Field(
        default=1024 * 1024,  # 1MB
        description="Maximum file size in bytes to index",
    )
    update_interval: PositiveInt = Field(
        default=86400,  # 24 hours
        description="How often to check for updates in seconds",
    )

exclude_patterns = Field(default_factory=(lambda: ['**/node_modules/**', '**/.git/**', '**/build/**']), description='File patterns to exclude when indexing documentation') class-attribute instance-attribute

index_patterns = Field(default_factory=(lambda: ['**/*.md', '**/*.rst', '**/*.txt']), description='File patterns to include when indexing documentation') class-attribute instance-attribute

max_file_size = Field(default=(1024 * 1024), description='Maximum file size in bytes to index') class-attribute instance-attribute

repositories = Field(default_factory=dict, description='Dictionary mapping package names to repository configurations') class-attribute instance-attribute

update_interval = Field(default=86400, description='How often to check for updates in seconds') class-attribute instance-attribute

FolderConfig

Bases: BaseModel

Configuration for a folder within a repository.

Source code in src/holoviz_mcp/config/models.py

class FolderConfig(BaseModel):
    """Configuration for a folder within a repository."""

    url_path: str = Field(default="", description="URL path mapping for this folder (e.g., '' for root, '/reference' for reference docs)")

url_path = Field(default='', description="URL path mapping for this folder (e.g., '' for root, '/reference' for reference docs)") class-attribute instance-attribute

GitRepository

Bases: BaseModel

Configuration for a Git repository.

Source code in src/holoviz_mcp/config/models.py

class GitRepository(BaseModel):
    """Configuration for a Git repository."""

    url: AnyHttpUrl = Field(..., description="Git repository URL")
    branch: Optional[str] = Field(default=None, description="Git branch to use")
    tag: Optional[str] = Field(default=None, description="Git tag to use (e.g., '1.7.2')")
    commit: Optional[str] = Field(default=None, description="Git commit hash to use")
    folders: Union[list[str], dict[str, FolderConfig]] = Field(
        default_factory=lambda: {"doc": FolderConfig()},
        description="Folders to index within the repository. Can be a list of folder names or a dict mapping folder names to FolderConfig objects",
    )
    base_url: AnyHttpUrl = Field(..., description="Base URL for documentation links")
    url_transform: Literal["holoviz", "plotly", "datashader"] = Field(
        default="holoviz",
        description="""How to transform file path into URL:

        - holoViz transform suffix to .html: filename.md -> filename.html
        - plotly transform suffix to /: filename.md -> filename/
        - datashader removes leading index and transform suffix to .html: 01_filename.md -> filename.html
        """,
    )
    reference_patterns: list[str] = Field(
        default_factory=lambda: ["examples/reference/**/*.md", "examples/reference/**/*.ipynb"], description="Glob patterns for reference documentation files"
    )

    @field_validator("tag")
    @classmethod
    def validate_tag(cls, v):
        """Validate git tag format, allowing both 'v1.2.3' and '1.2.3' formats."""
        if v is not None and v.startswith("v"):
            # Allow tags like 'v1.7.2' but also suggest plain version numbers
            pass
        return v

    @field_validator("folders")
    @classmethod
    def validate_folders(cls, v):
        """Validate and normalize folders configuration."""
        if isinstance(v, list):
            # Convert list to dict format for backward compatibility
            return {folder: FolderConfig() for folder in v}
        elif isinstance(v, dict):
            # Ensure all values are FolderConfig objects
            result = {}
            for folder, config in v.items():
                if isinstance(config, dict):
                    result[folder] = FolderConfig(**config)
                elif isinstance(config, FolderConfig):
                    result[folder] = config
                else:
                    raise ValueError(f"Invalid folder config for '{folder}': must be dict or FolderConfig")
            return result
        else:
            raise ValueError("folders must be a list or dict")

    def get_folder_names(self) -> list[str]:
        """Get list of folder names for backward compatibility."""
        if isinstance(self.folders, dict):
            return list(self.folders.keys())
        return self.folders

    def get_folder_url_path(self, folder_name: str) -> str:
        """Get URL path for a specific folder."""
        if isinstance(self.folders, dict):
            folder_config = self.folders.get(folder_name)
            if folder_config:
                return folder_config.url_path
        return ""

base_url = Field(..., description='Base URL for documentation links') class-attribute instance-attribute

branch = Field(default=None, description='Git branch to use') class-attribute instance-attribute

commit = Field(default=None, description='Git commit hash to use') class-attribute instance-attribute

folders = Field(default_factory=(lambda: {'doc': FolderConfig()}), description='Folders to index within the repository. Can be a list of folder names or a dict mapping folder names to FolderConfig objects') class-attribute instance-attribute

reference_patterns = Field(default_factory=(lambda: ['examples/reference/**/*.md', 'examples/reference/**/*.ipynb']), description='Glob patterns for reference documentation files') class-attribute instance-attribute

tag = Field(default=None, description="Git tag to use (e.g., '1.7.2')") class-attribute instance-attribute

url = Field(..., description='Git repository URL') class-attribute instance-attribute

url_transform = Field(default='holoviz', description='How to transform file path into URL:\n\n - holoViz transform suffix to .html: filename.md -> filename.html\n - plotly transform suffix to /: filename.md -> filename/\n - datashader removes leading index and transform suffix to .html: 01_filename.md -> filename.html\n ') class-attribute instance-attribute

get_folder_names()

Get list of folder names for backward compatibility.

Source code in src/holoviz_mcp/config/models.py

def get_folder_names(self) -> list[str]:
    """Get list of folder names for backward compatibility."""
    if isinstance(self.folders, dict):
        return list(self.folders.keys())
    return self.folders

get_folder_url_path(folder_name)

Get URL path for a specific folder.

Source code in src/holoviz_mcp/config/models.py

def get_folder_url_path(self, folder_name: str) -> str:
    """Get URL path for a specific folder."""
    if isinstance(self.folders, dict):
        folder_config = self.folders.get(folder_name)
        if folder_config:
            return folder_config.url_path
    return ""

validate_folders(v) classmethod

Validate and normalize folders configuration.

Source code in src/holoviz_mcp/config/models.py

@field_validator("folders")
@classmethod
def validate_folders(cls, v):
    """Validate and normalize folders configuration."""
    if isinstance(v, list):
        # Convert list to dict format for backward compatibility
        return {folder: FolderConfig() for folder in v}
    elif isinstance(v, dict):
        # Ensure all values are FolderConfig objects
        result = {}
        for folder, config in v.items():
            if isinstance(config, dict):
                result[folder] = FolderConfig(**config)
            elif isinstance(config, FolderConfig):
                result[folder] = config
            else:
                raise ValueError(f"Invalid folder config for '{folder}': must be dict or FolderConfig")
        return result
    else:
        raise ValueError("folders must be a list or dict")

validate_tag(v) classmethod

Validate git tag format, allowing both 'v1.2.3' and '1.2.3' formats.

Source code in src/holoviz_mcp/config/models.py

@field_validator("tag")
@classmethod
def validate_tag(cls, v):
    """Validate git tag format, allowing both 'v1.2.3' and '1.2.3' formats."""
    if v is not None and v.startswith("v"):
        # Allow tags like 'v1.7.2' but also suggest plain version numbers
        pass
    return v

HoloVizMCPConfig

Bases: BaseModel

Main configuration for HoloViz MCP server.

Source code in src/holoviz_mcp/config/models.py

class HoloVizMCPConfig(BaseModel):
    """Main configuration for HoloViz MCP server."""

    server: ServerConfig = Field(default_factory=ServerConfig)
    docs: DocsConfig = Field(default_factory=DocsConfig)
    resources: ResourceConfig = Field(default_factory=ResourceConfig)
    prompts: PromptConfig = Field(default_factory=PromptConfig)
    display: DisplayConfig = Field(default_factory=DisplayConfig)

    # Environment paths - merged from EnvironmentConfig with defaults
    user_dir: Path = Field(default_factory=_holoviz_mcp_user_dir, description="User configuration directory")
    default_dir: Path = Field(default_factory=_holoviz_mcp_default_dir, description="Default configuration directory")
    repos_dir: Path = Field(default_factory=lambda: _holoviz_mcp_user_dir() / "repos", description="Repository download directory")

    model_config = ConfigDict(extra="forbid", validate_assignment=True)

    def config_file_path(self, location: Literal["user", "default"] = "user") -> Path:
        """Get the path to the configuration file.

        Args:
            location: Whether to get user or default config file path
        """
        if location == "user":
            return self.user_dir / "config.yaml"
        else:
            return self.default_dir / "config.yaml"

    def resources_dir(self, location: Literal["user", "default"] = "user") -> Path:
        """Get the path to the resources directory.

        Args:
            location: Whether to get user or default resources directory
        """
        if location == "user":
            return self.user_dir / "resources"
        else:
            return self.default_dir / "resources"

    def agents_dir(self, location: Literal["user", "default"] = "user", tool: Literal["copilot", "claude"] | None = None) -> Path:
        """Get the path to the agents directory.

        Args:
            location: Whether to get user or default agents directory
            tool: Optional tool-specific subdirectory (e.g., "copilot", "claude")

        Returns
        -------
            Path to agents directory, optionally scoped to a specific tool
        """
        base_dir = self.resources_dir(location) / "agents"
        if tool:
            return base_dir / tool
        return base_dir

    def skills_dir(self, location: Literal["user", "default"] = "user") -> Path:
        """Get the path to the skills directory.

        Args:
            location: Whether to get user or default skills directory
        """
        return self.resources_dir(location) / "skills"

default_dir = Field(default_factory=_holoviz_mcp_default_dir, description='Default configuration directory') class-attribute instance-attribute

display = Field(default_factory=DisplayConfig) class-attribute instance-attribute

docs = Field(default_factory=DocsConfig) class-attribute instance-attribute

model_config = ConfigDict(extra='forbid', validate_assignment=True) class-attribute instance-attribute

prompts = Field(default_factory=PromptConfig) class-attribute instance-attribute

repos_dir = Field(default_factory=(lambda: _holoviz_mcp_user_dir() / 'repos'), description='Repository download directory') class-attribute instance-attribute

resources = Field(default_factory=ResourceConfig) class-attribute instance-attribute

server = Field(default_factory=ServerConfig) class-attribute instance-attribute

user_dir = Field(default_factory=_holoviz_mcp_user_dir, description='User configuration directory') class-attribute instance-attribute

agents_dir(location='user', tool=None)

Get the path to the agents directory.

Args: location: Whether to get user or default agents directory tool: Optional tool-specific subdirectory (e.g., "copilot", "claude")

Returns:

Type	Description
Path to agents directory, optionally scoped to a specific tool

Source code in src/holoviz_mcp/config/models.py

def agents_dir(self, location: Literal["user", "default"] = "user", tool: Literal["copilot", "claude"] | None = None) -> Path:
    """Get the path to the agents directory.

    Args:
        location: Whether to get user or default agents directory
        tool: Optional tool-specific subdirectory (e.g., "copilot", "claude")

    Returns
    -------
        Path to agents directory, optionally scoped to a specific tool
    """
    base_dir = self.resources_dir(location) / "agents"
    if tool:
        return base_dir / tool
    return base_dir

config_file_path(location='user')

Get the path to the configuration file.

Args: location: Whether to get user or default config file path

Source code in src/holoviz_mcp/config/models.py

def config_file_path(self, location: Literal["user", "default"] = "user") -> Path:
    """Get the path to the configuration file.

    Args:
        location: Whether to get user or default config file path
    """
    if location == "user":
        return self.user_dir / "config.yaml"
    else:
        return self.default_dir / "config.yaml"

resources_dir(location='user')

Get the path to the resources directory.

Args: location: Whether to get user or default resources directory

Source code in src/holoviz_mcp/config/models.py

def resources_dir(self, location: Literal["user", "default"] = "user") -> Path:
    """Get the path to the resources directory.

    Args:
        location: Whether to get user or default resources directory
    """
    if location == "user":
        return self.user_dir / "resources"
    else:
        return self.default_dir / "resources"

skills_dir(location='user')

Get the path to the skills directory.

Args: location: Whether to get user or default skills directory

Source code in src/holoviz_mcp/config/models.py

def skills_dir(self, location: Literal["user", "default"] = "user") -> Path:
    """Get the path to the skills directory.

    Args:
        location: Whether to get user or default skills directory
    """
    return self.resources_dir(location) / "skills"

PromptConfig

Bases: BaseModel

Configuration for prompts.

Source code in src/holoviz_mcp/config/models.py

class PromptConfig(BaseModel):
    """Configuration for prompts."""

    search_paths: list[Path] = Field(default_factory=list, description="Additional paths to search for prompts")

search_paths = Field(default_factory=list, description='Additional paths to search for prompts') class-attribute instance-attribute

ResourceConfig

Bases: BaseModel

Configuration for resources (agents, skills, etc.).

Source code in src/holoviz_mcp/config/models.py

class ResourceConfig(BaseModel):
    """Configuration for resources (agents, skills, etc.)."""

    search_paths: list[Path] = Field(default_factory=list, description="Additional paths to search for resources")

search_paths = Field(default_factory=list, description='Additional paths to search for resources') class-attribute instance-attribute

ServerConfig

Bases: BaseModel

Configuration for the MCP server.

Source code in src/holoviz_mcp/config/models.py

class ServerConfig(BaseModel):
    """Configuration for the MCP server."""

    name: str = Field(default="holoviz-mcp", description="Server name")
    version: str = Field(default="1.0.0", description="Server version")
    description: str = Field(default="Model Context Protocol server for HoloViz ecosystem", description="Server description")
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(default="INFO", description="Logging level")
    transport: Literal["stdio", "http"] = Field(default="stdio", description="Transport protocol for MCP communication")
    host: str = Field(default="127.0.0.1", description="Host address to bind to when using HTTP transport (use 0.0.0.0 for Docker)")
    port: int = Field(default=8000, description="Port to bind to when using HTTP transport")
    anonymized_telemetry: bool = Field(default=False, description="Enable anonymized telemetry")
    jupyter_server_proxy_url: str = Field(default="", description="Jupyter server proxy URL for Panel app integration")
    vector_db_path: Path = Field(
        default_factory=lambda: (_holoviz_mcp_user_dir() / "vector_db" / "chroma").expanduser(), description="Path to the Chroma vector database."
    )
    screenshots_dir: Path = Field(
        default_factory=lambda: (_holoviz_mcp_user_dir() / "screenshots").expanduser(), description="Directory for saving screenshots from panel_inspect_app tool."
    )

anonymized_telemetry = Field(default=False, description='Enable anonymized telemetry') class-attribute instance-attribute

description = Field(default='Model Context Protocol server for HoloViz ecosystem', description='Server description') class-attribute instance-attribute

host = Field(default='127.0.0.1', description='Host address to bind to when using HTTP transport (use 0.0.0.0 for Docker)') class-attribute instance-attribute

jupyter_server_proxy_url = Field(default='', description='Jupyter server proxy URL for Panel app integration') class-attribute instance-attribute

log_level = Field(default='INFO', description='Logging level') class-attribute instance-attribute

name = Field(default='holoviz-mcp', description='Server name') class-attribute instance-attribute

port = Field(default=8000, description='Port to bind to when using HTTP transport') class-attribute instance-attribute

screenshots_dir = Field(default_factory=(lambda: (_holoviz_mcp_user_dir() / 'screenshots').expanduser()), description='Directory for saving screenshots from panel_inspect_app tool.') class-attribute instance-attribute

transport = Field(default='stdio', description='Transport protocol for MCP communication') class-attribute instance-attribute

vector_db_path = Field(default_factory=(lambda: (_holoviz_mcp_user_dir() / 'vector_db' / 'chroma').expanduser()), description='Path to the Chroma vector database.') class-attribute instance-attribute

version = Field(default='1.0.0', description='Server version') class-attribute instance-attribute

Applications

Here for technical reasons.