feat: make workspace configurable (#106)

Author: mathislucka

src/deepset_mcp/main.py

@@ -0,0 +1,34 @@
+from .custom_components import get_latest_custom_component_installation_logs, list_custom_component_installations
+from .indexes import create_index, deploy_index, get_index, list_indexes, update_index
+from .pipeline_templates import get_pipeline_template, list_pipeline_templates, search_pipeline_templates
+from .pipelines import (
+    create_pipeline,
+    deploy_pipeline,
+    get_pipeline,
+    get_pipeline_logs,
+    list_pipelines,
+    search_pipeline,
+    update_pipeline,
+    validate_pipeline,
+)
+
+__all__ = [
+    "list_pipelines",
+    "get_pipeline",
+    "create_pipeline",
+    "update_pipeline",
+    "validate_pipeline",
+    "deploy_pipeline",
+    "get_pipeline_logs",
+    "search_pipeline",
+    "list_indexes",
+    "get_index",
+    "update_index",
+    "create_index",
+    "deploy_index",
+    "get_latest_custom_component_installation_logs",
+    "list_custom_component_installations",
+    "search_pipeline_templates",
+    "list_pipeline_templates",
+    "get_pipeline_template",
+]

src/deepset_mcp/mcp/__init__.py

@@ -0,0 +1,33 @@
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.tools.custom_components import (
+    get_latest_custom_component_installation_logs as get_latest_custom_component_installation_logs_tool,
+    list_custom_component_installations as list_custom_component_installations_tool,
+)
+
+
+async def list_custom_component_installations(workspace: str) -> str:
+    """Retrieves a list of recent custom component installations.
+
+    Use this to see the installation history of custom components, including status,
+    version information, and who installed them. This also includes installation logs.
+
+    :param workspace: The deepset workspace to operate on.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await list_custom_component_installations_tool(client, workspace)
+
+    return response
+
+
+async def get_latest_custom_component_installation_logs(workspace: str) -> str:
+    """Retrieves the logs from the latest custom component installation.
+
+    Use this to debug custom component installation issues or to understand
+    what happened during the most recent installation.
+
+    :param workspace: The deepset workspace to operate on.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await get_latest_custom_component_installation_logs_tool(client, workspace)
+
+    return response

src/deepset_mcp/mcp/custom_components.py

@@ -0,0 +1,95 @@
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.tools.indexes import (
+    create_index as create_index_tool,
+    deploy_index as deploy_index_tool,
+    get_index as get_index_tool,
+    list_indexes as list_indexes_tool,
+    update_index as update_index_tool,
+)
+
+
+async def list_indexes(workspace: str) -> str:
+    """Retrieves a list of all indexes available in the deepset workspace.
+
+    Use this to get an overview of existing indexes and their configurations.
+    The response includes basic information for each index.
+
+    :param workspace: The deepset workspace to operate on.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await list_indexes_tool(client=client, workspace=workspace)
+    return response
+
+
+async def get_index(workspace: str, index_name: str) -> str:
+    """Fetches detailed configuration information for a specific index.
+
+    Use this to get the full configuration and details of a single index.
+
+    :param workspace: The deepset workspace to operate on.
+    :param index_name: The name of the index to fetch.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await get_index_tool(client=client, workspace=workspace, index_name=index_name)
+    return response
+
+
+async def create_index(workspace: str, index_name: str, yaml_configuration: str, description: str | None = None) -> str:
+    """Creates a new index in the deepset workspace.
+
+    Use this to create a new index with the given configuration.
+    Make sure the YAML configuration is valid before creating the index.
+
+    :param workspace: The deepset workspace to operate on.
+    :param index_name: The name for the new index.
+    :param yaml_configuration: YAML configuration for the index.
+    :param description: Optional description for the index.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await create_index_tool(
+            client=client,
+            workspace=workspace,
+            index_name=index_name,
+            yaml_configuration=yaml_configuration,
+            description=description,
+        )
+    return response
+
+
+async def update_index(
+    workspace: str, index_name: str, updated_index_name: str | None = None, yaml_configuration: str | None = None
+) -> str:
+    """Updates an existing index in the deepset workspace.
+
+    Use this to update the name or configuration of an existing index.
+    You must provide at least one of updated_index_name or yaml_configuration.
+
+    :param workspace: The deepset workspace to operate on.
+    :param index_name: The name of the index to update.
+    :param updated_index_name: Optional new name for the index.
+    :param yaml_configuration: Optional new YAML configuration.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await update_index_tool(
+            client=client,
+            workspace=workspace,
+            index_name=index_name,
+            updated_index_name=updated_index_name,
+            yaml_configuration=yaml_configuration,
+        )
+    return response
+
+
+async def deploy_index(workspace: str, index_name: str) -> str:
+    """Deploys an index to production in the deepset workspace.
+
+    Use this to deploy an index that has been created and configured.
+    The deployment process will validate the index configuration and deploy it if valid.
+    If deployment fails due to validation errors, you will receive detailed error information.
+
+    :param workspace: The deepset workspace to operate on.
+    :param index_name: Name of the index to deploy.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await deploy_index_tool(client=client, workspace=workspace, index_name=index_name)
+    return response

src/deepset_mcp/mcp/indexes.py

@@ -0,0 +1,12 @@
+from functools import lru_cache
+
+from model2vec import StaticModel
+
+
+@lru_cache(maxsize=1)
+def get_initialized_model() -> StaticModel:
+    """Gets the initialized embedding model.
+
+    The model is cached to avoid reloading.
+    """
+    return StaticModel.from_pretrained("minishlab/potion-base-2M")

src/deepset_mcp/mcp/initialize_embedding_model.py

@@ -0,0 +1,54 @@
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.mcp.initialize_embedding_model import get_initialized_model
+from deepset_mcp.tools.pipeline_template import (
+    get_pipeline_template as get_pipeline_template_tool,
+    list_pipeline_templates as list_pipeline_templates_tool,
+    search_pipeline_templates as search_pipeline_templates_tool,
+)
+
+
+async def list_pipeline_templates(workspace: str) -> str:
+    """Retrieves a list of all pipeline templates available within the currently configured deepset workspace.
+
+    Use this when you need to know the available pipeline templates and their capabilities.
+
+    :param workspace: The name of the workspace where the pipeline templates are located.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await list_pipeline_templates_tool(client, workspace)
+
+    return response
+
+
+async def get_pipeline_template(workspace: str, template_name: str) -> str:
+    """Fetches detailed configuration information for a specific pipeline template.
+
+    This includes its YAML configuration, metadata, and recommended use cases.
+    Use this when you need to inspect a specific template's structure or settings.
+
+    :param workspace: The deepset workspace to operate on.
+    :param template_name: Name of the pipeline template to retrieve.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await get_pipeline_template_tool(client, workspace, template_name)
+
+    return response
+
+
+async def search_pipeline_templates(workspace: str, query: str) -> str:
+    """Use this to search for pipeline templates in deepset.
+
+    You can use full natural language queries to find templates.
+    You can also use simple keywords.
+    Use this if you want to find pipeline templates for specific use cases,
+    but you are not sure what the exact name of the template is.
+
+    :param workspace: The deepset workspace to operate on.
+    :param query: Search query for templates.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await search_pipeline_templates_tool(
+            client=client, query=query, model=get_initialized_model(), workspace=workspace
+        )
+
+    return response

src/deepset_mcp/mcp/pipeline_templates.py

@@ -0,0 +1,165 @@
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.api.pipeline.log_level import LogLevel
+from deepset_mcp.tools.pipeline import (
+    create_pipeline as create_pipeline_tool,
+    deploy_pipeline as deploy_pipeline_tool,
+    get_pipeline as get_pipeline_tool,
+    get_pipeline_logs as get_pipeline_logs_tool,
+    list_pipelines as list_pipelines_tool,
+    search_pipeline as search_pipeline_tool,
+    update_pipeline as update_pipeline_tool,
+    validate_pipeline as validate_pipeline_tool,
+)
+
+
+async def list_pipelines(workspace: str) -> str:
+    """Retrieves a list of all pipeline available within the currently configured deepset workspace.
+
+    Use this when you need to know the names or IDs of existing pipeline.
+    This does not return the pipeline configuration.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await list_pipelines_tool(client, workspace)
+
+    return response
+
+
+async def get_pipeline(workspace: str, pipeline_name: str) -> str:
+    """Fetches detailed configuration information for a specific pipeline, identified by its unique `pipeline_name`.
+
+    This includes its components, connections, and metadata.
+    Use this when you need to inspect the structure or settings of a known pipeline.
+
+    :param workspace: The deepset workspace to operate on.
+    :param pipeline_name: Name of the pipeline to retrieve.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await get_pipeline_tool(client, workspace, pipeline_name)
+
+    return response
+
+
+async def create_pipeline(workspace: str, pipeline_name: str, yaml_configuration: str) -> str:
+    """Creates a new pipeline in deepset.
+
+    :param workspace: The deepset workspace to operate on.
+    :param pipeline_name: Name of the pipeline to create
+    :param yaml_configuration: YAML configuration for the pipeline
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await create_pipeline_tool(client, workspace, pipeline_name, yaml_configuration)
+
+    return response
+
+
+async def update_pipeline(
+    workspace: str, pipeline_name: str, original_configuration_snippet: str, replacement_configuration_snippet: str
+) -> str:
+    """Updates an existing pipeline in deepset.
+
+    The update is performed by replacing the original configuration snippet with the new one.
+    Make sure that your original snippet only has a single exact match in the pipeline configuration.
+    Respect whitespace and formatting.
+
+    :param workspace: The deepset workspace to operate on.
+    :param pipeline_name: Name of the pipeline to update
+    :param original_configuration_snippet: The configuration snippet to replace
+    :param replacement_configuration_snippet: The new configuration snippet
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await update_pipeline_tool(
+            client=client,
+            workspace=workspace,
+            pipeline_name=pipeline_name,
+            original_config_snippet=original_configuration_snippet,
+            replacement_config_snippet=replacement_configuration_snippet,
+        )
+
+    return response
+
+
+async def validate_pipeline(workspace: str, yaml_configuration: str) -> str:
+    """
+    Validates the structure and syntax of a provided pipeline YAML configuration against the deepset API specifications.
+
+    Provide the YAML configuration as a string.
+    Returns a validation result, indicating success or detailing any errors or warnings found.
+    Use this *before* attempting to create or update a pipeline with new YAML.
+
+    :param workspace: The deepset workspace to operate on.
+    :param yaml_configuration: YAML configuration to validate.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await validate_pipeline_tool(client, workspace, yaml_configuration)
+
+    return response
+
+
+async def get_pipeline_logs(workspace: str, pipeline_name: str, limit: int = 30, level: str | None = None) -> str:
+    """Fetches logs for a specific pipeline in the deepset workspace.
+
+    Use this to debug pipeline issues, monitor pipeline execution, or understand what happened during pipeline runs.
+    The logs provide detailed information about pipeline operations, errors, and warnings.
+
+    :param workspace: The deepset workspace to operate on.
+    :param pipeline_name: Name of the pipeline to fetch logs for.
+    :param limit: Maximum number of log entries to return (default: 30, max: 100).
+    :param level: Filter logs by level. Valid values: 'info', 'warning', 'error'. If not specified, returns all levels.
+    """
+    # Convert string level to LogLevel enum if provided
+    log_level: LogLevel | None = None
+    if level is not None:
+        try:
+            log_level = LogLevel(level)
+        except ValueError:
+            return f"Invalid log level '{level}'. Valid values are: 'info', 'warning', 'error'."
+
+    async with AsyncDeepsetClient() as client:
+        response = await get_pipeline_logs_tool(
+            client=client,
+            workspace=workspace,
+            pipeline_name=pipeline_name,
+            limit=limit,
+            level=log_level,
+        )
+    return response
+
+
+async def deploy_pipeline(workspace: str, pipeline_name: str) -> str:
+    """Deploys a pipeline to production in the deepset workspace.
+
+    Use this to deploy a pipeline that has been created and validated.
+    The deployment process will validate the pipeline configuration and deploy it if valid.
+    If deployment fails due to validation errors, you will receive detailed error information.
+
+    :param workspace: The deepset workspace to operate on.
+    :param pipeline_name: Name of the pipeline to deploy.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await deploy_pipeline_tool(
+            client=client, workspace=workspace, pipeline_name=pipeline_name, wait_for_deployment=True
+        )
+    return response
+
+
+async def search_pipeline(workspace: str, pipeline_name: str, query: str) -> str:
+    """Search using a deployed pipeline in the deepset workspace.
+
+    This tool allows you to execute a search query using a specific pipeline.
+    The pipeline must already be deployed (status = DEPLOYED) for the search to work.
+    If the pipeline is not deployed, you will receive an error message instructing you to deploy it first.
+
+    Use this tool when you want to:
+    - Test a deployed pipeline with a specific query
+    - Get search results from a knowledge base using a specific pipeline
+    - Retrieve answers or documents based on a search query
+
+    :param workspace: The deepset workspace to operate on.
+    :param pipeline_name: Name of the deployed pipeline to use for search.
+    :param query: The search query to execute.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await search_pipeline_tool(
+            client=client, workspace=workspace, pipeline_name=pipeline_name, query=query
+        )
+    return response