refactor: make pipeline more consistent (#174)

Author: mathislucka

src/deepset_mcp/api/README.md

@@ -40,13 +40,13 @@ from deepset_mcp.api.client import AsyncDeepsetClient
 async with AsyncDeepsetClient() as client:
     # Access pipeline resource for a specific workspace
     pipelines = client.pipelines(workspace="your-workspace")
-    
+
     # List all pipelines
     pipeline_list = await pipelines.list(page_number=1, limit=10)
-    
+
     # Get a specific pipeline with its YAML configuration
     pipeline = await pipelines.get("my-pipeline", include_yaml=True)
-    
+
     # Create a new pipeline
     yaml_config = """
     components:
@@ -55,9 +55,9 @@ async with AsyncDeepsetClient() as client:
         params:
           model: "sentence-transformers/all-MiniLM-L6-v2"
     """
-    
+
     response = await pipelines.create(
-        name="my-new-pipeline",
+        pipeline_name="my-new-pipeline",
         yaml_config=yaml_config
     )
 ```
@@ -135,9 +135,9 @@ async with AsyncDeepsetClient() as client:
 Monitor pipeline performance and troubleshoot issues:
 
 ```python
-from deepset_mcp.api.pipeline.log_level import LogLevel
-from deepset_mcp.api.client import AsyncDeepsetClient
 
+from deepset_mcp.api.pipeline.models import LogLevel
+from deepset_mcp.api.client import AsyncDeepsetClient
 
 async with AsyncDeepsetClient() as client:
     pipelines = client.pipelines(workspace="your-workspace")
@@ -147,7 +147,7 @@ async with AsyncDeepsetClient() as client:
         limit=50,
         level=LogLevel.ERROR  # Filter by log level
     )
-    
+
     for log_entry in logs.data:
         print(f"[{log_entry.level}] {log_entry.message}")
 ```
@@ -212,23 +212,23 @@ from deepset_mcp.api.client import AsyncDeepsetClient
 
 async with AsyncDeepsetClient() as client:
     templates = client.pipeline_templates(workspace="your-workspace")
-    
+
     # List available templates
     template_list = await templates.list_templates(
         limit=20,
         field="created_at",
         order="DESC",
         filter="category eq 'RAG'"  # OData filter
     )
-    
+
     # Get a specific template with its YAML configuration
     template = await templates.get_template("template-name")
-    
+
     # Use the template YAML to create a new pipeline
     if template.yaml_config:
         pipelines = client.pipelines(workspace="your-workspace")
         await pipelines.create(
-            name="pipeline-from-template",
+            pipeline_name="pipeline-from-template",
             yaml_config=template.yaml_config
         )
 ```
@@ -451,11 +451,11 @@ from deepset_mcp.api.client import AsyncDeepsetClient
 async with AsyncDeepsetClient() as client:
     # Development workspace
     dev_pipelines = client.pipelines("development")
-    dev_pipeline = await dev_pipelines.create(name="test-pipeline", yaml_config=config)
-    
+    dev_pipeline = await dev_pipelines.create(pipeline_name="test-pipeline", yaml_config=config)
+
     # Production workspace
     prod_pipelines = client.pipelines("production")
-    prod_pipeline = await prod_pipelines.create(name="prod-pipeline", yaml_config=config)
+    prod_pipeline = await prod_pipelines.create(pipeline_name="prod-pipeline", yaml_config=config)
 ```
 
 ### Validation Before Deployment
@@ -465,18 +465,17 @@ Always validate pipeline configurations before deployment:
 ```python
 from deepset_mcp.api.client import AsyncDeepsetClient
 
-
 async with AsyncDeepsetClient() as client:
     pipelines = client.pipelines(workspace="your-workspace")
 
     # Validate configuration
     validation_result = await pipelines.validate(yaml_config)
-    
+
     if validation_result.valid:
         # Create and deploy pipeline
-        await pipelines.create(name="my-pipeline", yaml_config=yaml_config)
+        await pipelines.create(pipeline_name="my-pipeline", yaml_config=yaml_config)
         deployment_result = await pipelines.deploy("my-pipeline")
-        
+
         if deployment_result.valid:
             print("Pipeline deployed successfully")
         else:

src/deepset_mcp/api/pipeline/log_level.py

@@ -237,3 +237,11 @@ class PipelineOperationWithErrors(BaseModel):
     message: str
     validation_result: PipelineValidationResult
     pipeline: DeepsetPipeline
+
+
+class LogLevel(StrEnum):
+    """Log level filter options for pipeline logs."""
+
+    INFO = "info"
+    WARNING = "warning"
+    ERROR = "error"

src/deepset_mcp/api/pipeline/models.py

@@ -5,11 +5,11 @@
 from collections.abc import AsyncIterator
 from typing import Any, Protocol
 
-from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     DeepsetSearchResponse,
     DeepsetStreamEvent,
+    LogLevel,
     PipelineList,
     PipelineLogList,
     PipelineValidationResult,

src/deepset_mcp/api/pipeline/protocols.py

@@ -6,13 +6,14 @@
 import logging
 from collections.abc import AsyncIterator
 from typing import TYPE_CHECKING, Any
+from urllib.parse import quote
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
-from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     DeepsetSearchResponse,
     DeepsetStreamEvent,
+    LogLevel,
     PipelineList,
     PipelineLogList,
     PipelineValidationResult,
@@ -29,7 +30,7 @@
 
 
 class PipelineResource(PipelineResourceProtocol):
-    """Manages interactions with the deepset pipeline API."""
+    """Interact with pipelines on the deepset AI platform."""
 
     def __init__(
         self,
@@ -54,7 +55,7 @@ async def validate(self, yaml_config: str) -> PipelineValidationResult:
         data = {"query_yaml": yaml_config}
 
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipeline_validations",
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipeline_validations",
             method="POST",
             data=data,
         )
@@ -92,7 +93,7 @@ async def list(
         }
 
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines",
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines",
             method="GET",
             params=params,
         )
@@ -118,14 +119,18 @@ async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPip
         :param include_yaml: Whether to include YAML configuration in the response.
         :returns: DeepsetPipeline instance.
         """
-        resp = await self._client.request(endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}")
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}"
+        )
         raise_for_status(resp)
 
         pipeline = DeepsetPipeline.model_validate(resp.json)
 
         if include_yaml:
             yaml_response = await self._client.request(
-                endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/yaml"
+                endpoint=(
+                    f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}/yaml"
+                )
             )
 
             raise_for_status(yaml_response)
@@ -135,16 +140,16 @@ async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPip
 
         return pipeline
 
-    async def create(self, name: str, yaml_config: str) -> NoContentResponse:
+    async def create(self, pipeline_name: str, yaml_config: str) -> NoContentResponse:
         """Create a new pipeline with a name and YAML config.
 
-        :param name: Name of the new pipeline.
+        :param pipeline_name: Name of the new pipeline.
         :param yaml_config: YAML configuration for the pipeline.
         :returns: NoContentResponse indicating successful creation.
         """
-        data = {"name": name, "query_yaml": yaml_config}
+        data = {"name": pipeline_name, "query_yaml": yaml_config}
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines",
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines",
             method="POST",
             data=data,
         )
@@ -170,7 +175,7 @@ async def update(
         # Handle name update first if any
         if updated_pipeline_name is not None:
             name_resp = await self._client.request(
-                endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}",
+                endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}",
                 method="PATCH",
                 data={"name": updated_pipeline_name},
             )
@@ -184,7 +189,9 @@ async def update(
 
         if yaml_config is not None:
             yaml_resp = await self._client.request(
-                endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/yaml",
+                endpoint=(
+                    f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}/yaml"
+                ),
                 method="PUT",
                 data={"query_yaml": yaml_config},
             )
@@ -223,7 +230,7 @@ async def get_logs(
             params["filter"] = f"level eq '{level}' and origin eq 'querypipeline'"
 
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/logs",
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}/logs",
             method="GET",
             params=params,
         )
@@ -244,7 +251,9 @@ async def deploy(self, pipeline_name: str) -> PipelineValidationResult:
         :raises UnexpectedAPIError: If the API returns an unexpected status code.
         """
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/deploy",
+            endpoint=(
+                f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}/deploy"
+            ),
             method="POST",
         )
 
@@ -274,7 +283,7 @@ async def delete(self, pipeline_name: str) -> NoContentResponse:
         :raises UnexpectedAPIError: If the API returns an unexpected status code.
         """
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}",
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}",
             method="DELETE",
         )
 
@@ -315,7 +324,9 @@ async def search(
             data["filters"] = filters
 
         resp = await self._client.request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/search",
+            endpoint=(
+                f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}/search"
+            ),
             method="POST",
             data=data,
             response_type=dict[str, Any],
@@ -365,7 +376,10 @@ async def search_stream(
             data["filters"] = filters
 
         async with self._client.stream_request(
-            endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/search-stream",
+            endpoint=(
+                f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/"
+                f"{quote(pipeline_name, safe='')}/search-stream"
+            ),
             method="POST",
             data=data,
         ) as resp:

src/deepset_mcp/api/pipeline/resource.py

@@ -7,10 +7,10 @@
 import yaml
 
 from deepset_mcp.api.exceptions import BadRequestError, ResourceNotFoundError, UnexpectedAPIError
-from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     DeepsetSearchResponse,
+    LogLevel,
     PipelineList,
     PipelineLogList,
     PipelineOperationWithErrors,

src/deepset_mcp/tools/pipeline.py

@@ -125,7 +125,7 @@ async def test_get_logs_for_deployed_pipeline(
     pipeline_name = "test-logs-pipeline"
 
     # Step 1: Create a pipeline
-    await pipeline_resource.create(name=pipeline_name, yaml_config=simple_yaml_config)
+    await pipeline_resource.create(pipeline_name=pipeline_name, yaml_config=simple_yaml_config)
 
     # Step 2: Deploy the pipeline
     deploy_result = await pipeline_resource.deploy(pipeline_name=pipeline_name)
@@ -172,7 +172,7 @@ async def test_get_logs_for_non_deployed_pipeline(
     pipeline_name = "test-logs-non-deployed-pipeline"
 
     # Create pipeline but do not deploy it
-    await pipeline_resource.create(name=pipeline_name, yaml_config=simple_yaml_config)
+    await pipeline_resource.create(pipeline_name=pipeline_name, yaml_config=simple_yaml_config)
 
     # Try to get logs for the non-deployed pipeline
     logs = await pipeline_resource.get_logs(pipeline_name=pipeline_name)
@@ -197,7 +197,7 @@ async def test_deployment_timeout_handling(
     pipeline_name = "test-timeout-pipeline"
 
     # Create and deploy pipeline
-    await pipeline_resource.create(name=pipeline_name, yaml_config=simple_yaml_config)
+    await pipeline_resource.create(pipeline_name=pipeline_name, yaml_config=simple_yaml_config)
     deploy_result = await pipeline_resource.deploy(pipeline_name=pipeline_name)
     assert deploy_result.valid is True