API SDK

Main Client

deepset_mcp.api.client.AsyncDeepsetClient

Bases: AsyncClientProtocol

Async Client for interacting with the deepset API.

This client provides asynchronous access to most deepset API endpoints and resources through a convenient interface.

Example usage:

from deepset_mcp.api import AsyncDeepsetClient

async with AsyncDeepsetClient() as client:
    pipes = await client.pipelines(workspace="example-workspace")
    all_pipelines = await pipes.list()
    example_pipeline = await pipes.get(pipeline_name="example-pipeline")
    await pipes.create(pipeline_name="example-pipeline-v2", yaml_config="...")

    indexes = await client.indexes(workspace="example-workspace")
    all_indexes = await indexes.list()
    example_index = await indexes.get(index_name="example-index")

__init__

__init__(
    api_key: str | None = None,
    base_url: str = "https://api.cloud.deepset.ai/api",
    transport: TransportProtocol | None = None,
    transport_config: dict[str, Any] | None = None,
) -> None

Initialize an instance of the AsyncDeepsetClient.

Parameters:

Name	Type	Description	Default
api_key	str | None	API key or token. Falls back to DEEPSET_API_KEY env var	None
base_url	str	Base URL for the deepset API	'https://api.cloud.deepset.ai/api'
transport	TransportProtocol | None	Custom transport implementation	None
transport_config	dict[str, Any] | None	Configuration for default transport (e.g. timeout). All configuration parameters are forwarded to httpx.AsyncClient.	None

pipelines

pipelines(workspace: str) -> PipelineResource

Resource to interact with pipelines in the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	Workspace identifier	required

Returns:

Type	Description
PipelineResource	Pipeline resource instance

indexes

indexes(workspace: str) -> IndexResource

Resource to interact with indexes in the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	Workspace identifier	required

Returns:

Type	Description
IndexResource	Index resource instance

pipeline_templates

pipeline_templates(
    workspace: str,
) -> PipelineTemplateResource

Resource to interact with pipeline templates in the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	Workspace identifier	required

Returns:

Type	Description
PipelineTemplateResource	Pipeline template resource instance

haystack_service

haystack_service() -> HaystackServiceResource

Resource to interact with the Haystack service API.

Returns:

Type	Description
HaystackServiceResource	Haystack service resource instance

integrations

integrations() -> IntegrationResource

Resource to interact with integrations.

Returns:

Type	Description
IntegrationResource	Integration resource instance

custom_components

custom_components(
    workspace: str,
) -> CustomComponentsResource

Resource to interact with custom components in the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	Workspace identifier	required

Returns:

Type	Description
CustomComponentsResource	Custom components resource instance

secrets

secrets() -> SecretResource

Resource to interact with secrets.

Returns:

Type	Description
SecretResource	Secret resource instance

workspaces

workspaces() -> WorkspaceResource

Resource to interact with workspaces.

Returns:

Type	Description
WorkspaceResource	Workspace resource instance

users

users() -> UserResource

Resource to interact with users.

Returns:

Type	Description
UserResource	User resource instance

request async

request(
    endpoint: str,
    *,
    response_type: type[T],
    method: str = "GET",
    data: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[T]

request(
    endpoint: str,
    *,
    response_type: None = None,
    method: str = "GET",
    data: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[Any]

request(
    endpoint: str,
    *,
    method: str = "GET",
    data: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    response_type: type[T] | None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[Any]

Make a regular (non-streaming) request to the deepset API.

Parameters:

Name	Type	Description	Default
endpoint	str	API endpoint path	required
method	str	HTTP method	'GET'
data	dict[str, Any] | None	JSON data to send in request body	None
headers	dict[str, str] | None	Additional headers to include	None
response_type	type[T] | None	Expected response type for type checking	None
timeout	float | None | Literal['config']	Request timeout in seconds. If "config", uses transport config timeout. If None, disables timeout. If float, uses specific timeout	'config'
kwargs	Any	Additional arguments to pass to transport	{}

Returns:

Type	Description
TransportResponse[Any]	Response with parsed JSON if available

stream_request

stream_request(
    endpoint: str,
    *,
    method: str = "POST",
    data: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    **kwargs: Any,
) -> AbstractAsyncContextManager[StreamingResponse]

Make a streaming request to the deepset API.

Must be used as an async context manager to ensure proper cleanup.

Example::

async with client.stream_request("/pipelines/search-stream", data={"query": "AI"}) as response:
    if response.success:
        async for line in response.iter_lines():
            # Process each line of the stream
            data = json.loads(line)
            print(data)
    else:
        # Handle error
        error_body = await response.read_body()
        print(f"Error {response.status_code}: {error_body}")

Parameters:

Name	Type	Description	Default
endpoint	str	API endpoint path	required
method	str	HTTP method (usually POST for streaming)	'POST'
data	dict[str, Any] | None	JSON data to send in request body	None
headers	dict[str, str] | None	Additional headers to include	None
kwargs	Any	Additional arguments to pass to transport	{}

Returns:

Type	Description
AbstractAsyncContextManager[StreamingResponse]	Response object with streaming capabilities

close async

close() -> None

Close underlying transport resources.

Resources

deepset_mcp.api.pipeline.PipelineResource

Bases: PipelineResourceProtocol

Interact with pipelines on the deepset AI platform.

__init__

__init__(
    client: AsyncClientProtocol, workspace: str
) -> None

Initializes a PipelineResource instance.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The async client protocol instance.	required
workspace	str	The workspace identifier.	required

validate async

validate(yaml_config: str) -> PipelineValidationResult

Validate a pipeline's YAML configuration against the API.

Parameters:

Name	Type	Description	Default
yaml_config	str	The YAML configuration string to validate.	required

Returns:

Type	Description
PipelineValidationResult	PipelineValidationResult containing validation status and any errors.

Raises:

Type	Description
ValueError	If the YAML is not valid (422 error) or contains syntax errors.

list async

list(
    limit: int = 10, after: str | None = None
) -> PaginatedResponse[DeepsetPipeline]

Lists pipelines and returns the first page of results.

The returned object can be iterated over to fetch subsequent pages.

Parameters:

Name	Type	Description	Default
limit	int	The maximum number of pipelines to return per page.	10
after	str | None	The cursor to fetch the next page of results.	None

Returns:

Type	Description
PaginatedResponse[DeepsetPipeline]	A PaginatedResponse object containing the first page of pipelines.

get async

get(
    pipeline_name: str, include_yaml: bool = True
) -> DeepsetPipeline

Fetch a single pipeline by its name.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline to fetch.	required
include_yaml	bool	Whether to include YAML configuration in the response.	True

Returns:

Type	Description
DeepsetPipeline	DeepsetPipeline instance.

create async

create(
    pipeline_name: str, yaml_config: str
) -> NoContentResponse

Create a new pipeline with a name and YAML config.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the new pipeline.	required
yaml_config	str	YAML configuration for the pipeline.	required

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful creation.

update async

update(
    pipeline_name: str,
    updated_pipeline_name: str | None = None,
    yaml_config: str | None = None,
) -> NoContentResponse

Update name and/or YAML config of an existing pipeline.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Current name of the pipeline.	required
updated_pipeline_name	str | None	New name for the pipeline (optional).	None
yaml_config	str | None	New YAML configuration (optional).	None

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful update.

Raises:

Type	Description
ValueError	If neither updated_pipeline_name nor yaml_config is provided.

get_logs async

get_logs(
    pipeline_name: str,
    limit: int = 30,
    level: LogLevel | None = None,
    after: str | None = None,
) -> PaginatedResponse[PipelineLog]

Fetch logs for a specific pipeline and returns the first page of results.

The returned object can be iterated over to fetch subsequent pages.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline to fetch logs for.	required
limit	int	Maximum number of log entries to return per page.	30
level	LogLevel | None	Filter logs by level. If None, returns all levels.	None
after	str | None	The cursor to fetch the next page of results.	None

Returns:

Type	Description
PaginatedResponse[PipelineLog]	A PaginatedResponse object containing the first page of logs.

deploy async

deploy(pipeline_name: str) -> PipelineValidationResult

Deploy a pipeline to production.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline to deploy.	required

Returns:

Type	Description
PipelineValidationResult	PipelineValidationResult containing deployment status and any errors.

Raises:

Type	Description
UnexpectedAPIError	If the API returns an unexpected status code.

delete async

delete(pipeline_name: str) -> NoContentResponse

Delete a pipeline.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline to delete.	required

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful deletion.

Raises:

Type	Description
UnexpectedAPIError	If the API returns an unexpected status code.

search async

search(
    pipeline_name: str,
    query: str,
    debug: bool = False,
    view_prompts: bool = False,
    params: dict[str, Any] | None = None,
    filters: dict[str, Any] | None = None,
) -> DeepsetSearchResponse

Search using a pipeline.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline to use for search.	required
query	str	Search query.	required
debug	bool	Whether to include debug information.	False
view_prompts	bool	Whether to include prompts in the response.	False
params	dict[str, Any] | None	Additional parameters for pipeline components.	None
filters	dict[str, Any] | None	Search filters to apply.	None

Returns:

Type	Description
DeepsetSearchResponse	SearchResponse containing search results.

search_stream async

search_stream(
    pipeline_name: str,
    query: str,
    debug: bool = False,
    view_prompts: bool = False,
    params: dict[str, Any] | None = None,
    filters: dict[str, Any] | None = None,
) -> AsyncIterator[DeepsetStreamEvent]

Search using a pipeline with response streaming.

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline to use for search.	required
query	str	Search query.	required
debug	bool	Whether to include debug information.	False
view_prompts	bool	Whether to include prompts in the response.	False
params	dict[str, Any] | None	Additional parameters for pipeline components.	None
filters	dict[str, Any] | None	Search filters to apply.	None

Returns:

Type	Description
AsyncIterator[DeepsetStreamEvent]	AsyncIterator streaming the result.

deepset_mcp.api.indexes.IndexResource

Bases: IndexResourceProtocol

Resource for interacting with deepset indexes.

__init__

__init__(
    client: AsyncClientProtocol, workspace: str
) -> None

Initialize the index resource.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The async REST client.	required
workspace	str	The workspace to use.	required

list async

list(
    limit: int = 10, after: str | None = None
) -> PaginatedResponse[Index]

Lists indexes and returns the first page of results.

The returned object can be iterated over to fetch subsequent pages.

Parameters:

Name	Type	Description	Default
limit	int	The maximum number of indexes to return per page.	10
after	str | None	The cursor to fetch the next page of results.	None

Returns:

Type	Description
PaginatedResponse[Index]	A PaginatedResponse object containing the first page of indexes.

get async

get(index_name: str) -> Index

Get a specific index.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the index.	required

Returns:

Type	Description
Index	Index details.

create async

create(
    index_name: str,
    yaml_config: str,
    description: str | None = None,
) -> Index

Create a new index with the given name and configuration.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the index	required
yaml_config	str	YAML configuration for the index	required
description	str | None	Optional description for the index	None

Returns:

Type	Description
Index	Created index details

update async

update(
    index_name: str,
    updated_index_name: str | None = None,
    yaml_config: str | None = None,
) -> Index

Update name and/or configuration of an existing index.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the index to update	required
updated_index_name	str | None	Optional new name for the index	None
yaml_config	str | None	Optional new YAML configuration	None

Returns:

Type	Description
Index	Updated index details

delete async

delete(index_name: str) -> None

Delete an index.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the index to delete.	required

deploy async

deploy(index_name: str) -> PipelineValidationResult

Deploy an index.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the index to deploy.	required

Returns:

Type	Description
PipelineValidationResult	PipelineValidationResult containing deployment status and any errors.

Raises:

Type	Description
UnexpectedAPIError	If the API returns an unexpected status code.

validate async

validate(yaml_config: str) -> PipelineValidationResult

Validate an index's YAML configuration against the API.

Parameters:

Name	Type	Description	Default
yaml_config	str	The YAML configuration string to validate.	required

Returns:

Type	Description
PipelineValidationResult	PipelineValidationResult containing validation status and any errors.

Raises:

Type	Description
ValueError	If the YAML is not valid (422 error) or contains syntax errors.

deepset_mcp.api.pipeline_template.PipelineTemplateResource

Bases: PipelineTemplateResourceProtocol

Resource for interacting with pipeline templates in a workspace.

__init__

__init__(
    client: AsyncClientProtocol, workspace: str
) -> None

Initialize the pipeline template resource.

Parameters

client : AsyncClientProtocol Client to use for making API requests workspace : str Workspace to operate in

get_template async

get_template(template_name: str) -> PipelineTemplate

Fetch a single pipeline template by its name.

Parameters

template_name : str Name of the template to fetch

Returns

PipelineTemplate The requested pipeline template

list async

list(
    limit: int = 10,
    after: str | None = None,
    field: str = "created_at",
    order: str = "DESC",
    filter: str | None = None,
) -> PaginatedResponse[PipelineTemplate]

Lists pipeline templates and returns the first page of results.

The returned object can be iterated over to fetch subsequent pages.

Parameters:

Name	Type	Description	Default
limit	int	The maximum number of pipeline templates to return per page.	10
after	str | None	The cursor to fetch the next page of results.	None
field	str	Field to sort by (default: "created_at").	'created_at'
order	str	Sort order, either "ASC" or "DESC" (default: "DESC").	'DESC'
filter	str | None	OData filter expression for filtering templates.	None

Returns:

Type	Description
PaginatedResponse[PipelineTemplate]	A PaginatedResponse object containing the first page of pipeline templates.

deepset_mcp.api.haystack_service.HaystackServiceResource

Bases: HaystackServiceProtocol

Manages interactions with the deepset Haystack service API.

__init__

__init__(client: AsyncClientProtocol) -> None

Initializes a HaystackServiceResource instance.

get_component_schemas async

get_component_schemas() -> dict[str, Any]

Fetch the component schema from the API.

Returns: The component schema as a dictionary

get_component_input_output async

get_component_input_output(
    component_name: str,
) -> dict[str, Any]

Fetch the component input and output schema from the API.

Args: component_name: The name of the component to fetch the input/output schema for

Returns: The component input/output schema as a dictionary

run_component async

run_component(
    component_type: str,
    init_params: dict[str, Any] | None = None,
    input_data: dict[str, Any] | None = None,
    input_types: dict[str, str] | None = None,
    workspace: str | None = None,
) -> dict[str, Any]

Run a Haystack component with the given parameters.

Parameters:

Name	Type	Description	Default
component_type	str	The type of component to run (e.g., "haystack.components.builders.PromptBuilder")	required
init_params	dict[str, Any] | None	Initialization parameters for the component	None
input_data	dict[str, Any] | None	Input data for the component	None
input_types	dict[str, str] | None	Optional type information for inputs (inferred if not provided)	None
workspace	str | None	Optional workspace name to run the component in	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing the component's output sockets

Integration API resources and models.

deepset_mcp.api.integrations.IntegrationResource

Bases: IntegrationResourceProtocol

Manages interactions with the deepset integrations API.

__init__

__init__(client: AsyncClientProtocol) -> None

Initialize an IntegrationResource instance.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The async client protocol instance.	required

list async

list() -> list[Integration]

Retrieve all integrations.

Returns:

Type	Description
list[Integration]	list containing all available integrations.

get async

get(provider: IntegrationProvider) -> Integration

Retrieve a specific integration by provider.

Parameters:

Name	Type	Description	Default
provider	IntegrationProvider	The integration provider to retrieve.	required

Returns:

Type	Description
Integration	Integration instance for the specified provider.

deepset_mcp.api.custom_components.CustomComponentsResource

Bases: CustomComponentsProtocol

Resource for managing custom components in deepset.

__init__

__init__(client: AsyncClientProtocol) -> None

Initialize a CustomComponentsResource.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The API client to use for requests.	required

list_installations async

list_installations(
    limit: int = 20,
    after: str | None = None,
    field: str = "created_at",
    order: str = "DESC",
) -> PaginatedResponse[CustomComponentInstallation]

Lists custom component installations and returns the first page of results.

The returned object can be iterated over to fetch subsequent pages.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of installations to return per page.	20
after	str | None	The cursor to fetch the next page of results.	None
field	str	Field to sort by.	'created_at'
order	str	Sort order (ASC or DESC).	'DESC'

Returns:

Type	Description
PaginatedResponse[CustomComponentInstallation]	A PaginatedResponse object containing the first page of installations.

get_latest_installation_logs async

get_latest_installation_logs() -> str | None

Get the logs from the latest custom component installation.

Returns:

Type	Description
str | None	Latest installation logs.

deepset_mcp.api.secrets.SecretResource

Bases: SecretResourceProtocol

Resource for managing secrets in deepset.

__init__

__init__(client: AsyncClientProtocol) -> None

Initialize a SecretResource.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The API client to use for requests.	required

list async

list(
    limit: int = 10,
    field: str = "created_at",
    order: str = "DESC",
    after: str | None = None,
) -> PaginatedResponse[Secret]

List secrets with pagination.

The returned object can be iterated over to fetch subsequent pages.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of secrets to return.	10
field	str	Field to sort by.	'created_at'
order	str	Sort order (ASC or DESC).	'DESC'
after	str | None	The cursor to fetch the next page of results.	None

Returns:

Type	Description
PaginatedResponse[Secret]	A PaginatedResponse object containing the first page of secrets.

create async

create(name: str, secret: str) -> NoContentResponse

Create a new secret.

Parameters:

Name	Type	Description	Default
name	str	The name of the secret.	required
secret	str	The secret value.	required

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful creation.

get async

get(secret_id: str) -> Secret

Get a specific secret by ID.

Parameters:

Name	Type	Description	Default
secret_id	str	The ID of the secret to retrieve.	required

Returns:

Type	Description
Secret	Secret information.

delete async

delete(secret_id: str) -> NoContentResponse

Delete a secret by ID.

Parameters:

Name	Type	Description	Default
secret_id	str	The ID of the secret to delete.	required

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful deletion.

Workspace API module.

deepset_mcp.api.workspace.WorkspaceResource

Bases: WorkspaceResourceProtocol

Manages interactions with the deepset workspace API.

__init__

__init__(client: AsyncClientProtocol) -> None

Initialize a WorkspaceResource instance.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The async client protocol instance.	required

list async

list() -> list[Workspace]

List all workspaces.

Returns:

Type	Description
list[Workspace]	A list containing all workspaces.

get async

get(workspace_name: str) -> Workspace

Get a specific workspace by name.

Parameters:

Name	Type	Description	Default
workspace_name	str	Name of the workspace to fetch.	required

Returns:

Type	Description
Workspace	A Workspace instance.

create async

create(name: str) -> NoContentResponse

Create a new workspace.

Parameters:

Name	Type	Description	Default
name	str	Name of the new workspace.	required

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful creation.

delete async

delete(workspace_name: str) -> NoContentResponse

Delete a workspace.

Parameters:

Name	Type	Description	Default
workspace_name	str	Name of the workspace to delete.	required

Returns:

Type	Description
NoContentResponse	NoContentResponse indicating successful deletion.

deepset_mcp.api.user.UserResource

Bases: UserResourceProtocol

Resource for managing users in deepset.

__init__

__init__(client: AsyncClientProtocol) -> None

Initialize a UserResource.

Parameters:

Name	Type	Description	Default
client	AsyncClientProtocol	The API client to use for requests.	required

get async

get(user_id: str) -> DeepsetUser

Get user information by user ID.

Parameters:

Name	Type	Description	Default
user_id	str	The ID of the user to fetch.	required

Returns:

Type	Description
DeepsetUser	User information.

Models

deepset_mcp.api.shared_models.NoContentResponse

Bases: BaseModel

Response model for an empty response.

success class-attribute instance-attribute

success: bool = True

Indicates whether the operation was successful

message class-attribute instance-attribute

message: str = 'No content'

Human-readable message describing the response

deepset_mcp.api.shared_models.DeepsetUser

Bases: BaseModel

Model representing a user on the deepset platform.

id class-attribute instance-attribute

id: str = Field(alias='user_id')

Unique identifier for the user

given_name class-attribute instance-attribute

given_name: str | None = None

User's given (first) name

family_name class-attribute instance-attribute

family_name: str | None = None

User's family (last) name

email class-attribute instance-attribute

email: str | None = None

User's email address

deepset_mcp.api.shared_models.PaginatedResponse

Bases: BaseModel, Generic[T]

A response model for a single page of cursor-paginated results.

This model also acts as an async iterator to fetch subsequent pages.

data instance-attribute

data: list[T]

List of items for the current page

has_more instance-attribute

has_more: bool

Whether there are more items available beyond this page

total class-attribute instance-attribute

total: int | None = None

Total number of items across all pages, if known

next_cursor class-attribute instance-attribute

next_cursor: str | None = None

Cursor for fetching the next page of results

populate_cursors_from_data classmethod

populate_cursors_from_data(
    data: dict[str, Any],
) -> dict[str, Any]

Populate next_cursor from the last element of data.

create_with_cursor_field classmethod

create_with_cursor_field(
    data: dict[str, Any], cursor_field: str
) -> PaginatedResponse[T]

Factory method that allows specifying the cursor field.

items async

items() -> AsyncIterator[T]

Asynchronously iterates over each item across all pages, starting from this page.

deepset_mcp.api.pipeline.models.PipelineServiceLevel

Bases: StrEnum

Describes the service level of a pipeline.

deepset_mcp.api.pipeline.models.DeepsetPipeline

Bases: BaseModel

Model representing a pipeline on the deepset platform.

id class-attribute instance-attribute

id: str = Field(alias='pipeline_id')

Unique identifier for the pipeline

name instance-attribute

name: str

Human-readable name of the pipeline

status instance-attribute

status: str

Current operational status of the pipeline

service_level instance-attribute

service_level: PipelineServiceLevel

Service level indicating the deployment stage

created_at instance-attribute

created_at: datetime

Timestamp when the pipeline was created

last_updated_at class-attribute instance-attribute

last_updated_at: datetime | None = Field(
    None, alias="last_edited_at"
)

Timestamp when the pipeline was last modified

created_by instance-attribute

created_by: DeepsetUser

User who created the pipeline

last_updated_by class-attribute instance-attribute

last_updated_by: DeepsetUser | None = Field(
    None, alias="last_edited_by"
)

User who last modified the pipeline

yaml_config class-attribute instance-attribute

yaml_config: str | None = None

YAML configuration defining the pipeline structure

Config

Configuration for serialization and deserialization.

deepset_mcp.api.pipeline.models.ValidationError

Bases: BaseModel

Model representing a validation error from the pipeline validation API.

code instance-attribute

code: str

Error code identifying the type of validation error

message instance-attribute

message: str

Human-readable description of the validation error

deepset_mcp.api.pipeline.models.PipelineValidationResult

Bases: BaseModel

Result of validating a pipeline configuration.

valid instance-attribute

valid: bool

Whether the pipeline configuration is valid

errors class-attribute instance-attribute

errors: list[ValidationError] = []

List of validation errors if the pipeline is invalid

deepset_mcp.api.pipeline.models.TraceFrame

Bases: BaseModel

Model representing a single frame in a stack trace.

filename instance-attribute

filename: str

Name of the file where the trace frame occurred

line_number instance-attribute

line_number: int

Line number in the file where the trace frame occurred

name instance-attribute

name: str

Function or method name where the trace frame occurred

deepset_mcp.api.pipeline.models.ExceptionInfo

Bases: BaseModel

Model representing exception information.

type instance-attribute

type: str

Exception class name

value instance-attribute

value: str

Exception message or string representation

trace instance-attribute

trace: list[TraceFrame]

Stack trace frames leading to the exception

deepset_mcp.api.pipeline.models.PipelineLog

Bases: BaseModel

Model representing a single log entry from a pipeline.

log_id instance-attribute

log_id: str

Unique identifier for the log entry

message instance-attribute

message: str

Log message content

logged_at instance-attribute

logged_at: datetime

Timestamp when the log entry was created

level instance-attribute

level: str

Log level (e.g., INFO, WARNING, ERROR)

origin instance-attribute

origin: str

Source component or service that generated the log

exceptions class-attribute instance-attribute

exceptions: list[ExceptionInfo] | None = None

Exception information if the log contains error details

extra_fields class-attribute instance-attribute

extra_fields: dict[str, Any] = Field(default_factory=dict)

Additional metadata fields associated with the log entry

deepset_mcp.api.pipeline.models.OffsetRange

Bases: BaseModel

Model representing an offset range.

start instance-attribute

start: int

Starting position of the offset range

end instance-attribute

end: int

Ending position of the offset range

deepset_mcp.api.pipeline.models.DeepsetAnswer

Bases: BaseModel

Model representing a search answer.

answer instance-attribute

answer: str

The generated answer text

context class-attribute instance-attribute

context: str | None = None

Context text used to generate the answer

document_id class-attribute instance-attribute

document_id: str | None = None

Identifier of the source document

document_ids class-attribute instance-attribute

document_ids: list[str] | None = None

List of source document identifiers

file class-attribute instance-attribute

file: dict[str, Any] | None = None

File metadata associated with the answer

files class-attribute instance-attribute

files: list[dict[str, Any]] | None = None

List of file metadata associated with the answer

meta class-attribute instance-attribute

meta: dict[str, Any] | None = None

Additional metadata about the answer

offsets_in_context class-attribute instance-attribute

offsets_in_context: list[OffsetRange] | None = None

Character offset ranges within the context text

offsets_in_document class-attribute instance-attribute

offsets_in_document: list[OffsetRange] | None = None

Character offset ranges within the source document

prompt class-attribute instance-attribute

prompt: str | None = None

Prompt used to generate the answer

result_id class-attribute instance-attribute

result_id: UUID | None = None

Unique identifier for this result

score class-attribute instance-attribute

score: float | None = None

Confidence or relevance score for the answer

type class-attribute instance-attribute

type: str | None = None

Type classification of the answer

deepset_mcp.api.pipeline.models.DeepsetDocument

Bases: BaseModel

Model representing a search document.

content instance-attribute

content: str

Text content of the document

meta instance-attribute

meta: dict[str, Any]

Metadata dictionary containing document properties

embedding class-attribute instance-attribute

embedding: list[float] | None = None

Vector embedding representation of the document

file class-attribute instance-attribute

file: dict[str, Any] | None = None

File metadata if the document originated from a file

id class-attribute instance-attribute

id: str | None = None

Unique identifier for the document

result_id class-attribute instance-attribute

result_id: UUID | None = None

Unique identifier for this search result

score class-attribute instance-attribute

score: float | None = None

Relevance or similarity score for the document

deepset_mcp.api.pipeline.models.DeepsetSearchResponse

Bases: BaseModel

Model representing a single search result.

debug class-attribute instance-attribute

debug: dict[str, Any] | None = Field(
    default=None, alias="_debug"
)

Debug information for the search operation

answers class-attribute instance-attribute

answers: list[DeepsetAnswer] = Field(default_factory=list)

List of generated answers from the search

documents class-attribute instance-attribute

documents: list[DeepsetDocument] = Field(
    default_factory=list
)

List of retrieved documents from the search

prompts class-attribute instance-attribute

prompts: dict[str, str] | None = None

Prompts used during the search operation

query class-attribute instance-attribute

query: str | None = None

Original search query text

query_id class-attribute instance-attribute

query_id: UUID | None = None

Unique identifier for the search query

normalize_response classmethod

normalize_response(data: dict[str, Any]) -> dict[str, Any]

Normalize the response from the search and search-stream endpoints.

The search endpoint returns a list of results, but we only ever use the first result. We are not sending batch queries, so there will never be more than one result. We use this validator to transform the data so that we can use the same response model for search and search-stream endpoints.

deepset_mcp.api.pipeline.models.StreamDelta

Bases: BaseModel

Model representing a streaming delta.

text instance-attribute

text: str

Incremental text content for streaming responses

meta class-attribute instance-attribute

meta: dict[str, Any] | None = None

Metadata associated with the streaming delta

deepset_mcp.api.pipeline.models.DeepsetStreamEvent

Bases: BaseModel

Model representing a stream event.

query_id class-attribute instance-attribute

query_id: str | UUID | None = None

Unique identifier for the associated query

type instance-attribute

type: str

Event type: 'delta', 'result', or 'error'

delta class-attribute instance-attribute

delta: StreamDelta | None = None

Streaming text delta if type is 'delta'

result class-attribute instance-attribute

result: DeepsetSearchResponse | None = None

Complete search result if type is 'result'

error class-attribute instance-attribute

error: str | None = None

Error message if type is 'error'

deepset_mcp.api.pipeline.models.LogLevel

Bases: StrEnum

Log level filter options for pipeline logs.

deepset_mcp.api.indexes.models.IndexStatus

Bases: BaseModel

Status information about documents in an index.

pending_file_count instance-attribute

pending_file_count: int

Number of files waiting to be processed

failed_file_count instance-attribute

failed_file_count: int

Number of files that failed during indexing

indexed_no_documents_file_count instance-attribute

indexed_no_documents_file_count: int

Number of files indexed but containing no documents

indexed_file_count instance-attribute

indexed_file_count: int

Number of successfully indexed files

total_file_count instance-attribute

total_file_count: int

Total number of files in the index

deepset_mcp.api.indexes.models.Index

Bases: BaseModel

A deepset index.

pipeline_index_id instance-attribute

pipeline_index_id: str

Unique identifier for the pipeline index

name instance-attribute

name: str

Human-readable name of the index

description class-attribute instance-attribute

description: str | None = None

Optional description of the index purpose and contents

yaml_config class-attribute instance-attribute

yaml_config: str = Field(alias='config_yaml')

YAML configuration defining the index structure and settings

workspace_id instance-attribute

workspace_id: str

ID of the workspace containing this index

settings instance-attribute

settings: dict[str, Any]

Index configuration settings and parameters

desired_status instance-attribute

desired_status: str

Target operational status for the index

deployed_at class-attribute instance-attribute

deployed_at: datetime | None = None

Timestamp when the index was deployed

last_edited_at class-attribute instance-attribute

last_edited_at: datetime | None = None

Timestamp when the index was last modified

max_index_replica_count instance-attribute

max_index_replica_count: int

Maximum number of replicas allowed for this index

created_at instance-attribute

created_at: datetime

Timestamp when the index was created

updated_at class-attribute instance-attribute

updated_at: datetime | None = None

Timestamp when the index was last updated

created_by instance-attribute

created_by: DeepsetUser

User who created the index

last_edited_by class-attribute instance-attribute

last_edited_by: DeepsetUser | None = None

User who last modified the index

status instance-attribute

status: IndexStatus

Current status information about documents in the index

deepset_mcp.api.pipeline_template.models.PipelineType

Bases: StrEnum

Enum representing the type of a pipeline template.

deepset_mcp.api.pipeline_template.models.PipelineTemplateTag

Bases: BaseModel

Model representing a tag on a pipeline template.

name instance-attribute

name: str

Human-readable name of the tag

tag_id instance-attribute

tag_id: UUID

Unique identifier for the tag

deepset_mcp.api.pipeline_template.models.PipelineTemplate

Bases: BaseModel

Model representing a pipeline template.

author instance-attribute

author: str

Name of the template author or creator

best_for instance-attribute

best_for: list[str]

List of use cases this template is best suited for

description instance-attribute

description: str

Detailed description of the pipeline template

template_name class-attribute instance-attribute

template_name: str = Field(alias='pipeline_name')

Internal name identifier for the template

display_name class-attribute instance-attribute

display_name: str = Field(alias='name')

User-friendly display name for the template

pipeline_template_id class-attribute instance-attribute

pipeline_template_id: UUID = Field(
    alias="pipeline_template_id"
)

Unique identifier for the pipeline template

potential_applications class-attribute instance-attribute

potential_applications: list[str] = Field(
    alias="potential_applications"
)

List of potential applications and scenarios for this template

yaml_config class-attribute instance-attribute

yaml_config: str | None = None

YAML configuration defining the pipeline structure

tags instance-attribute

tags: list[PipelineTemplateTag]

List of tags associated with the template for categorization

pipeline_type instance-attribute

pipeline_type: PipelineType

Type of pipeline (query or indexing)

populate_yaml_config classmethod

populate_yaml_config(values: Any) -> Any

Populate yaml_config from query_yaml or indexing_yaml based on pipeline_type.

Models for the integrations API.

deepset_mcp.api.integrations.models.IntegrationProvider

Bases: StrEnum

Supported integration providers.

deepset_mcp.api.integrations.models.Integration

Bases: BaseModel

Model representing an integration.

invalid instance-attribute

invalid: bool

Whether the integration configuration is invalid or misconfigured

model_registry_token_id instance-attribute

model_registry_token_id: UUID

Unique identifier for the model registry token

provider instance-attribute

provider: IntegrationProvider

The integration provider type (e.g., OpenAI, Azure, etc.)

provider_domain instance-attribute

provider_domain: str

Domain or endpoint URL for the integration provider

deepset_mcp.api.custom_components.models.CustomComponentInstallation

Bases: BaseModel

Model representing a custom component installation.

custom_component_id instance-attribute

custom_component_id: str

Unique identifier for the custom component

status instance-attribute

status: str

Current installation status of the component

version instance-attribute

version: str

Version number of the installed component

created_by_user_id instance-attribute

created_by_user_id: str

ID of the user who initiated the installation

logs instance-attribute

logs: list[dict[str, Any]]

Installation log entries with timestamps and messages

organization_id instance-attribute

organization_id: str

ID of the organization where the component is installed

user_info class-attribute instance-attribute

user_info: DeepsetUser | None = None

Detailed information about the user who created the installation

deepset_mcp.api.secrets.models.Secret

Bases: BaseModel

Model representing a secret in deepset.

name instance-attribute

name: str

Human-readable name of the secret

secret_id instance-attribute

secret_id: str

Unique identifier for the secret

Models for workspace API responses.

deepset_mcp.api.workspace.models.Workspace

Bases: BaseModel

Model representing a workspace on the deepset platform.

name instance-attribute

name: str

Human-readable name of the workspace

workspace_id instance-attribute

workspace_id: UUID

Unique identifier for the workspace

languages instance-attribute

languages: dict[str, Any]

Supported languages and their configuration settings

default_idle_timeout_in_seconds instance-attribute

default_idle_timeout_in_seconds: int

Default timeout in seconds before workspace becomes idle

Exceptions

deepset_mcp.api.exceptions.DeepsetAPIError

Bases: Exception

Base exception for all deepset API errors.

__init__

__init__(
    status_code: int | None = None,
    message: Any | None = None,
    detail: Any | None = None,
) -> None

Initialize the exception.

deepset_mcp.api.exceptions.ResourceNotFoundError

Bases: DeepsetAPIError

Exception raised when a resource is not found (HTTP 404).

__init__

__init__(
    message: Any = "Resource not found",
    detail: Any | None = None,
) -> None

Initialize the exception.

deepset_mcp.api.exceptions.BadRequestError

Bases: DeepsetAPIError

Exception raised for invalid requests (HTTP 400).

__init__

__init__(
    message: Any = "Bad request", detail: Any | None = None
) -> None

Initialize the exception.

deepset_mcp.api.exceptions.RequestTimeoutError

Bases: Exception

Exception raised when a request times out.

__init__

__init__(
    method: str,
    url: str,
    timeout: float | None | str,
    duration: float | None = None,
    detail: str | None = None,
)

Initialize the timeout exception with request context.

deepset_mcp.api.exceptions.UnexpectedAPIError

Bases: DeepsetAPIError

Catch-all exception for unexpected API errors.

__init__

__init__(
    status_code: int | None = None,
    message: Any = "Unexpected API error",
    detail: Any | None = None,
)

Initialize the exception.

Transport

deepset_mcp.api.transport.StreamReaderProtocol

Bases: Protocol

Protocol for reading from a stream.

aread async

aread() -> bytes

Read the entire response body.

aiter_lines

aiter_lines() -> AsyncIterator[str]

Iterate over response lines.

deepset_mcp.api.transport.TransportResponse dataclass

Bases: Generic[T]

Response envelope for regular HTTP transport.

success property

success: bool

Check if the response was successful (status code < 400).

deepset_mcp.api.transport.StreamingResponse dataclass

Response envelope for streaming HTTP transport.

success property

success: bool

Check if the response was successful (status code < 400).

iter_lines async

iter_lines() -> AsyncIterator[str]

Iterate over response lines.

For error responses (status >= 400), reads the entire body and yields it. For success responses, yields line by line.

read_body async

read_body() -> str

Read the entire response body. Useful for error handling.

deepset_mcp.api.transport.TransportProtocol

Bases: Protocol

Protocol for HTTP transport with separate streaming support.

request async

request(
    method: str,
    url: str,
    *,
    response_type: type[T],
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[T]

request(
    method: str,
    url: str,
    *,
    response_type: None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[Any]

request(
    method: str,
    url: str,
    *,
    response_type: type[T] | None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[Any]

Send a regular HTTP request and return the response.

stream

stream(
    method: str, url: str, **kwargs: Any
) -> AbstractAsyncContextManager[StreamingResponse]

Open a streaming HTTP connection.

Must be used as an async context manager to ensure proper cleanup.

close async

close() -> None

Clean up any resources (e.g., close connections).

deepset_mcp.api.transport.AsyncTransport

Asynchronous HTTP transport using httpx.AsyncClient.

__init__

__init__(
    base_url: str,
    api_key: str,
    config: dict[str, Any] | None = None,
)

Initialize an instance of AsyncTransport.

Parameters:

Name	Type	Description	Default
base_url	str	Base URL for the API	required
api_key	str	Bearer token for authentication	required
config	dict[str, Any] | None	Configuration for httpx.AsyncClient, e.g., {'timeout': 10.0}	None

request async

request(
    method: str,
    url: str,
    *,
    response_type: type[T],
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[T]

request(
    method: str,
    url: str,
    *,
    response_type: None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[Any]

request(
    method: str,
    url: str,
    *,
    response_type: type[T] | None = None,
    timeout: float | None | Literal["config"] = "config",
    **kwargs: Any,
) -> TransportResponse[Any]

Send a regular HTTP request and return the response.

Parameters:

Name	Type	Description	Default
method	str	HTTP method	required
url	str	URL endpoint	required
response_type	type[T] | None	Expected response type for type checking	None
timeout	float | None | Literal['config']	Request timeout in seconds. If "config", uses transport config timeout. If None, disables timeout. If float, uses specific timeout.	'config'
kwargs	Any	Additional arguments to pass to httpx	{}

Returns:

Type	Description
TransportResponse[Any]	The response with parsed JSON if available

stream

stream(
    method: str, url: str, **kwargs: Any
) -> AbstractAsyncContextManager[StreamingResponse]

Open a streaming HTTP connection.

:yields: Response object with streaming capabilities

.. code-block:: python

async with transport.stream("POST", "/api/stream", json=data) as response:
    if response.success:
        async for line in response.iter_lines():
            process_line(line)
    else:
        error = await response.read_body()
        handle_error(error)

Parameters:

Name	Type	Description	Default
method	str	HTTP method	required
url	str	URL endpoint	required
kwargs	Any	Additional arguments to pass to httpx.stream()	{}

close async

close() -> None

Clean up any resources (e.g., close connections).

deepset_mcp.api.transport.raise_for_status

raise_for_status(response: TransportResponse[Any]) -> None

Raises the appropriate exception based on the response status code.