"""Configuration component CRUD helpers for AWM."""
import json
from collections.abc import Callable, Generator
from typing import Any
import chariot_api._openapi.awm.models as api_models
from chariot import _apis, mcp_setting
from chariot.awm import _utils, models
ComponentType = models.ConfigurationType
# The generated OpenAPI client types config as []int, but the API accepts a JSON object
# or JSON-/YAML-encoded string (Go json.RawMessage). Dicts are sent as inline JSON objects;
# strings are sent as-is. Pass through model_construct at call sites.
ConfigPayload = dict[str, Any] | str
_API_TYPE_NAMES: dict[ComponentType, str] = {
models.COMPONENT_TYPE_WORKFLOW: "workflow",
models.COMPONENT_TYPE_AGENT: "agent",
models.COMPONENT_TYPE_TOOL_PROFILE: "tool_profile",
models.COMPONENT_TYPE_MCP_CONNECTION: "mcp_connection",
models.COMPONENT_TYPE_EVALUATOR: "evaluator",
models.COMPONENT_TYPE_SHARED: "shared",
}
_CREATE_REQUEST_MODELS: dict[ComponentType, type] = {
models.COMPONENT_TYPE_WORKFLOW: api_models.InputCreateConfigurationRequestInputWorkflowConfiguration,
models.COMPONENT_TYPE_AGENT: api_models.InputCreateConfigurationRequestInputAgentConfiguration,
models.COMPONENT_TYPE_TOOL_PROFILE: api_models.InputCreateConfigurationRequestInputToolProfileConfiguration,
models.COMPONENT_TYPE_MCP_CONNECTION: api_models.InputCreateConfigurationRequestInputMCPConnectionConfiguration,
models.COMPONENT_TYPE_EVALUATOR: api_models.InputCreateConfigurationRequestInputEvaluatorConfiguration,
models.COMPONENT_TYPE_SHARED: api_models.InputCreateConfigurationRequestInputSharedConfiguration,
}
_CREATE_VERSION_REQUEST_MODELS: dict[ComponentType, type] = {
models.COMPONENT_TYPE_WORKFLOW: api_models.InputCreateConfigurationVersionRequestInputWorkflowConfiguration,
models.COMPONENT_TYPE_AGENT: api_models.InputCreateConfigurationVersionRequestInputAgentConfiguration,
models.COMPONENT_TYPE_TOOL_PROFILE: api_models.InputCreateConfigurationVersionRequestInputToolProfileConfiguration,
models.COMPONENT_TYPE_MCP_CONNECTION: api_models.InputCreateConfigurationVersionRequestInputMCPConnectionConfiguration,
models.COMPONENT_TYPE_EVALUATOR: api_models.InputCreateConfigurationVersionRequestInputEvaluatorConfiguration,
models.COMPONENT_TYPE_SHARED: api_models.InputCreateConfigurationVersionRequestInputSharedConfiguration,
}
__all__ = [
"create_workflow_configuration",
"create_agent_configuration",
"create_tool_profile_configuration",
"create_mcp_connection_configuration",
"create_evaluator_configuration",
"create_shared_configuration",
"create_workflow_configuration_version",
"create_agent_configuration_version",
"create_tool_profile_configuration_version",
"create_mcp_connection_configuration_version",
"create_evaluator_configuration_version",
"create_shared_configuration_version",
"create_workflow_configuration_stage",
"create_agent_configuration_stage",
"create_tool_profile_configuration_stage",
"create_mcp_connection_configuration_stage",
"create_evaluator_configuration_stage",
"create_shared_configuration_stage",
"get_workflow_configuration_version",
"get_agent_configuration_version",
"get_tool_profile_configuration_version",
"get_mcp_connection_configuration_version",
"get_evaluator_configuration_version",
"get_shared_configuration_version",
"get_configuration_by_reference",
"list_workflow_configuration_components",
"list_agent_configuration_components",
"list_tool_profile_configuration_components",
"list_mcp_connection_configuration_components",
"list_evaluator_configuration_components",
"list_shared_configuration_components",
"list_workflow_configuration_versions",
"list_agent_configuration_versions",
"list_tool_profile_configuration_versions",
"list_mcp_connection_configuration_versions",
"list_evaluator_configuration_versions",
"list_shared_configuration_versions",
"delete_workflow_configuration_version",
"delete_agent_configuration_version",
"delete_tool_profile_configuration_version",
"delete_mcp_connection_configuration_version",
"delete_evaluator_configuration_version",
"delete_shared_configuration_version",
]
def _configuration_api():
return _apis.awm.configuration_api
def _api_method(
component_type: ComponentType,
action: str,
*,
raw: bool = False,
) -> Callable[..., Any]:
api_type = _API_TYPE_NAMES[component_type]
method_names = {
"create": f"create_{api_type}_configuration",
"create_version": f"create_{api_type}_configuration_version",
"create_stage": f"create_{api_type}_configuration_stage",
"get_version": f"get_{api_type}_configuration_version",
"delete_version": f"delete_{api_type}_configuration_version",
"list_components": f"list_{api_type}_configuration_components",
"list_versions": f"list_{api_type}_configuration_versions",
}
method_name = method_names[action]
if raw:
method_name = f"{method_name}_without_preload_content"
return getattr(_configuration_api(), method_name)
def _require_raw_payload(response: Any, operation: str) -> dict[str, Any]:
payload_bytes = getattr(response, "data", None)
if payload_bytes is None and hasattr(response, "read"):
payload_bytes = response.read()
if payload_bytes is None:
raise RuntimeError(f"Received malformed response (missing response body) from {operation}")
payload_text = (
payload_bytes.decode("utf-8") if isinstance(payload_bytes, bytes) else str(payload_bytes)
)
payload = json.loads(payload_text)
if not isinstance(payload, dict):
raise RuntimeError(f"Received malformed response (invalid JSON object) from {operation}")
return payload
def _require_version_payload(
payload: dict[str, Any], operation: str
) -> models.ConfigurationComponentVersion:
version_data = payload.get("data")
if version_data is None:
raise RuntimeError(f"Received malformed response (missing `data`) from {operation}")
if not isinstance(version_data, dict):
raise TypeError("Expected configuration version payload to be a dict")
return _to_version(version_data)
def _config_payload(config: ConfigPayload) -> ConfigPayload:
return config
def _api_payload(data: Any) -> dict[str, Any]:
"""Convert a generated OpenAPI client object to a plain dict at the SDK boundary."""
if hasattr(data, "model_dump"):
return data.model_dump()
if isinstance(data, dict):
return data
raise TypeError(f"Expected API response payload to be a dict, got {type(data)!r}")
def _required(raw: dict[str, Any], *keys: str) -> Any:
for key in keys:
value = raw.get(key)
if value is not None:
return value
raise RuntimeError(f"Received malformed configuration payload (missing {'/'.join(keys)})")
def _to_component(raw: dict[str, Any]) -> models.ConfigurationComponent:
return models.ConfigurationComponent(
id=_required(raw, "id"),
project_id=_required(raw, "project_id", "projectID"),
component_type=_required(raw, "component_type", "componentType"),
name=_required(raw, "name"),
key=_required(raw, "key"),
deleted_at=raw.get("deleted_at", raw.get("deletedAt")),
)
def _to_version_summary(raw: dict[str, Any]) -> models.ConfigurationComponentVersionSummary:
return models.ConfigurationComponentVersionSummary(
id=_required(raw, "id"),
component_id=_required(raw, "component_id", "componentID"),
component_type=_required(raw, "component_type", "componentType"),
project_id=_required(raw, "project_id", "projectID"),
name=_required(raw, "name"),
key=_required(raw, "key"),
description=raw.get("description") or "",
version=int(_required(raw, "version")),
created_at=_required(raw, "created_at", "createdAt"),
created_by=_required(raw, "created_by", "createdBy"),
version_changelog=raw.get("version_changelog", raw.get("versionChangelog")),
stages=raw.get("stages"),
deleted_at=raw.get("deleted_at", raw.get("deletedAt")),
deleted_by=raw.get("deleted_by", raw.get("deletedBy")),
)
def _to_version(raw: dict[str, Any]) -> models.ConfigurationComponentVersion:
return models.ConfigurationComponentVersion(
id=_required(raw, "id"),
component_id=_required(raw, "component_id", "componentID"),
component_type=_required(raw, "component_type", "componentType"),
project_id=_required(raw, "project_id", "projectID"),
name=_required(raw, "name"),
key=_required(raw, "key"),
description=raw.get("description") or "",
version=int(_required(raw, "version")),
created_at=_required(raw, "created_at", "createdAt"),
created_by=_required(raw, "created_by", "createdBy"),
config=raw.get("config"),
raw_config=raw.get("raw_config", raw.get("rawConfig")),
version_changelog=raw.get("version_changelog", raw.get("versionChangelog")),
stages=raw.get("stages"),
deleted_at=raw.get("deleted_at", raw.get("deletedAt")),
deleted_by=raw.get("deleted_by", raw.get("deletedBy")),
)
def _to_resolved(raw: dict[str, Any]) -> models.ResolvedConfiguration:
version_data = raw.get("data")
if version_data is None:
raise RuntimeError("Received malformed response (missing `data`) from configuration get")
if not isinstance(version_data, dict):
raise TypeError("Expected configuration version payload to be a dict")
return models.ResolvedConfiguration(
data=_to_version(version_data),
resolved=raw.get("resolved"),
)
def _configuration_endpoint(func: Callable[..., Any]) -> Callable[..., Any]:
return mcp_setting(ignore=True)(_apis.login_required(func))
def _create_configuration(
component_type: ComponentType,
project_id: str,
name: str,
key: str,
description: str,
config: ConfigPayload,
) -> models.ConfigurationComponentVersion:
request_model = _CREATE_REQUEST_MODELS[component_type]
operation = f"create_{component_type}_configuration"
response = _api_method(component_type, "create", raw=True)(
request_model.model_construct(
project_id=project_id,
name=name,
key=key,
description=description,
config=_config_payload(config),
)
)
return _require_version_payload(_require_raw_payload(response, operation), operation)
def _create_configuration_version(
component_type: ComponentType,
component_id: str,
config: ConfigPayload,
version_changelog: str,
description: str | None = None,
) -> models.ConfigurationComponentVersion:
request_model = _CREATE_VERSION_REQUEST_MODELS[component_type]
operation = f"create_{component_type}_configuration_version"
response = _api_method(component_type, "create_version", raw=True)(
component_id,
request_model.model_construct(
config=_config_payload(config),
version_changelog=version_changelog,
description=description,
),
)
return _require_version_payload(_require_raw_payload(response, operation), operation)
def _create_configuration_stage(
component_type: ComponentType,
version_id: str,
stage: models.ConfigurationStage,
) -> None:
_api_method(component_type, "create_stage")(
version_id,
api_models.InputCreateConfigurationStageRequest(stage=stage),
)
def _get_configuration_version(
component_type: ComponentType,
version_id: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
operation = f"get_{component_type}_configuration_version"
response = _api_method(component_type, "get_version", raw=True)(
version_id,
resolved=resolved,
)
return _to_resolved(_require_raw_payload(response, operation))
[docs]
@_configuration_endpoint
def get_configuration_by_reference(
project_id: str,
component_type: ComponentType,
key: str,
version_or_stage: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
"""Get a configuration component by project, type, key, and version or stage."""
operation = "get_configuration_by_reference"
response = _configuration_api().get_configuration_by_reference_without_preload_content(
project_id=project_id,
component_type=component_type,
key=key,
version_or_stage=version_or_stage,
resolved=resolved,
)
return _to_resolved(_require_raw_payload(response, operation))
def _list_configuration_components(
component_type: ComponentType,
name: str | None = None,
project_ids: list[str] | None = None,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_type: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponent, None, None]:
params = {
k: v
for k, v in {
"name": name,
"project_id": project_ids,
"include_deleted": include_deleted,
"limit": limit,
"after_key": after_key,
"sort_type": sort_type,
"sort_order": sort_order,
}.items()
if v is not None
}
def _paginate_fn(**params: Any) -> Any:
return _api_method(component_type, "list_components")(**params)
return _utils.paginate(
func=_paginate_fn,
params=params,
transform_item=lambda item: _to_component(_api_payload(item)),
)
def _list_configuration_versions(
component_type: ComponentType,
component_id: str,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponentVersionSummary, None, None]:
params = {
k: v
for k, v in {
"include_deleted": include_deleted,
"limit": limit,
"after_key": after_key,
"sort_order": sort_order,
}.items()
if v is not None
}
def _paginate_fn(**params: Any) -> Any:
return _api_method(component_type, "list_versions")(
component_id=component_id,
**params,
)
return _utils.paginate(
func=_paginate_fn,
params=params,
transform_item=lambda item: _to_version_summary(_api_payload(item)),
)
def _delete_configuration_version(component_type: ComponentType, version_id: str) -> None:
_api_method(component_type, "delete_version")(version_id)
[docs]
@_configuration_endpoint
def create_workflow_configuration(
project_id: str,
name: str,
key: str,
description: str,
config: ConfigPayload,
) -> models.ConfigurationComponentVersion:
"""Create a workflow configuration component."""
return _create_configuration(
models.COMPONENT_TYPE_WORKFLOW,
project_id,
name,
key,
description,
config,
)
[docs]
@_configuration_endpoint
def create_agent_configuration(
project_id: str,
name: str,
key: str,
description: str,
config: ConfigPayload,
) -> models.ConfigurationComponentVersion:
"""Create an agent configuration component."""
return _create_configuration(
models.COMPONENT_TYPE_AGENT,
project_id,
name,
key,
description,
config,
)
[docs]
@_configuration_endpoint
def create_mcp_connection_configuration(
project_id: str,
name: str,
key: str,
description: str,
config: ConfigPayload,
) -> models.ConfigurationComponentVersion:
"""Create an MCP connection configuration component."""
return _create_configuration(
models.COMPONENT_TYPE_MCP_CONNECTION,
project_id,
name,
key,
description,
config,
)
[docs]
@_configuration_endpoint
def create_evaluator_configuration(
project_id: str,
name: str,
key: str,
description: str,
config: ConfigPayload,
) -> models.ConfigurationComponentVersion:
"""Create an evaluator configuration component."""
return _create_configuration(
models.COMPONENT_TYPE_EVALUATOR,
project_id,
name,
key,
description,
config,
)
[docs]
@_configuration_endpoint
def create_shared_configuration(
project_id: str,
name: str,
key: str,
description: str,
config: ConfigPayload,
) -> models.ConfigurationComponentVersion:
"""Create a shared configuration component."""
return _create_configuration(
models.COMPONENT_TYPE_SHARED,
project_id,
name,
key,
description,
config,
)
[docs]
@_configuration_endpoint
def create_workflow_configuration_version(
component_id: str,
config: ConfigPayload,
version_changelog: str,
description: str | None = None,
) -> models.ConfigurationComponentVersion:
"""Create a new workflow configuration component version."""
return _create_configuration_version(
models.COMPONENT_TYPE_WORKFLOW,
component_id,
config,
version_changelog,
description,
)
[docs]
@_configuration_endpoint
def create_agent_configuration_version(
component_id: str,
config: ConfigPayload,
version_changelog: str,
description: str | None = None,
) -> models.ConfigurationComponentVersion:
"""Create a new agent configuration component version."""
return _create_configuration_version(
models.COMPONENT_TYPE_AGENT,
component_id,
config,
version_changelog,
description,
)
[docs]
@_configuration_endpoint
def create_mcp_connection_configuration_version(
component_id: str,
config: ConfigPayload,
version_changelog: str,
description: str | None = None,
) -> models.ConfigurationComponentVersion:
"""Create a new MCP connection configuration component version."""
return _create_configuration_version(
models.COMPONENT_TYPE_MCP_CONNECTION,
component_id,
config,
version_changelog,
description,
)
[docs]
@_configuration_endpoint
def create_evaluator_configuration_version(
component_id: str,
config: ConfigPayload,
version_changelog: str,
description: str | None = None,
) -> models.ConfigurationComponentVersion:
"""Create a new evaluator configuration component version."""
return _create_configuration_version(
models.COMPONENT_TYPE_EVALUATOR,
component_id,
config,
version_changelog,
description,
)
[docs]
@_configuration_endpoint
def create_shared_configuration_version(
component_id: str,
config: ConfigPayload,
version_changelog: str,
description: str | None = None,
) -> models.ConfigurationComponentVersion:
"""Create a new shared configuration component version."""
return _create_configuration_version(
models.COMPONENT_TYPE_SHARED,
component_id,
config,
version_changelog,
description,
)
[docs]
@_configuration_endpoint
def create_workflow_configuration_stage(
version_id: str,
stage: models.ConfigurationStage,
) -> None:
"""Create a stage for a workflow configuration component version."""
_create_configuration_stage(models.COMPONENT_TYPE_WORKFLOW, version_id, stage)
[docs]
@_configuration_endpoint
def create_agent_configuration_stage(
version_id: str,
stage: models.ConfigurationStage,
) -> None:
"""Create a stage for an agent configuration component version."""
_create_configuration_stage(models.COMPONENT_TYPE_AGENT, version_id, stage)
[docs]
@_configuration_endpoint
def create_mcp_connection_configuration_stage(
version_id: str,
stage: models.ConfigurationStage,
) -> None:
"""Create a stage for an MCP connection configuration component version."""
_create_configuration_stage(models.COMPONENT_TYPE_MCP_CONNECTION, version_id, stage)
[docs]
@_configuration_endpoint
def create_evaluator_configuration_stage(
version_id: str,
stage: models.ConfigurationStage,
) -> None:
"""Create a stage for an evaluator configuration component version."""
_create_configuration_stage(models.COMPONENT_TYPE_EVALUATOR, version_id, stage)
[docs]
@_configuration_endpoint
def create_shared_configuration_stage(
version_id: str,
stage: models.ConfigurationStage,
) -> None:
"""Create a stage for a shared configuration component version."""
_create_configuration_stage(models.COMPONENT_TYPE_SHARED, version_id, stage)
[docs]
@_configuration_endpoint
def get_workflow_configuration_version(
version_id: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
"""Get a workflow configuration component version."""
return _get_configuration_version(models.COMPONENT_TYPE_WORKFLOW, version_id, resolved)
[docs]
@_configuration_endpoint
def get_agent_configuration_version(
version_id: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
"""Get an agent configuration component version."""
return _get_configuration_version(models.COMPONENT_TYPE_AGENT, version_id, resolved)
[docs]
@_configuration_endpoint
def get_mcp_connection_configuration_version(
version_id: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
"""Get an MCP connection configuration component version."""
return _get_configuration_version(
models.COMPONENT_TYPE_MCP_CONNECTION,
version_id,
resolved,
)
[docs]
@_configuration_endpoint
def get_evaluator_configuration_version(
version_id: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
"""Get an evaluator configuration component version."""
return _get_configuration_version(models.COMPONENT_TYPE_EVALUATOR, version_id, resolved)
[docs]
@_configuration_endpoint
def get_shared_configuration_version(
version_id: str,
resolved: bool | None = None,
) -> models.ResolvedConfiguration:
"""Get a shared configuration component version."""
return _get_configuration_version(models.COMPONENT_TYPE_SHARED, version_id, resolved)
[docs]
@_configuration_endpoint
def list_workflow_configuration_components(
name: str | None = None,
project_ids: list[str] | None = None,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_type: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponent, None, None]:
"""List workflow configuration components."""
return _list_configuration_components(
models.COMPONENT_TYPE_WORKFLOW,
name,
project_ids,
include_deleted,
limit,
after_key,
sort_type,
sort_order,
)
[docs]
@_configuration_endpoint
def list_agent_configuration_components(
name: str | None = None,
project_ids: list[str] | None = None,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_type: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponent, None, None]:
"""List agent configuration components."""
return _list_configuration_components(
models.COMPONENT_TYPE_AGENT,
name,
project_ids,
include_deleted,
limit,
after_key,
sort_type,
sort_order,
)
[docs]
@_configuration_endpoint
def list_mcp_connection_configuration_components(
name: str | None = None,
project_ids: list[str] | None = None,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_type: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponent, None, None]:
"""List MCP connection configuration components."""
return _list_configuration_components(
models.COMPONENT_TYPE_MCP_CONNECTION,
name,
project_ids,
include_deleted,
limit,
after_key,
sort_type,
sort_order,
)
[docs]
@_configuration_endpoint
def list_evaluator_configuration_components(
name: str | None = None,
project_ids: list[str] | None = None,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_type: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponent, None, None]:
"""List evaluator configuration components."""
return _list_configuration_components(
models.COMPONENT_TYPE_EVALUATOR,
name,
project_ids,
include_deleted,
limit,
after_key,
sort_type,
sort_order,
)
[docs]
@_configuration_endpoint
def list_shared_configuration_components(
name: str | None = None,
project_ids: list[str] | None = None,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_type: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponent, None, None]:
"""List shared configuration components."""
return _list_configuration_components(
models.COMPONENT_TYPE_SHARED,
name,
project_ids,
include_deleted,
limit,
after_key,
sort_type,
sort_order,
)
[docs]
@_configuration_endpoint
def list_workflow_configuration_versions(
component_id: str,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponentVersionSummary, None, None]:
"""List workflow configuration component versions."""
return _list_configuration_versions(
models.COMPONENT_TYPE_WORKFLOW,
component_id,
include_deleted,
limit,
after_key,
sort_order,
)
[docs]
@_configuration_endpoint
def list_agent_configuration_versions(
component_id: str,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponentVersionSummary, None, None]:
"""List agent configuration component versions."""
return _list_configuration_versions(
models.COMPONENT_TYPE_AGENT,
component_id,
include_deleted,
limit,
after_key,
sort_order,
)
[docs]
@_configuration_endpoint
def list_mcp_connection_configuration_versions(
component_id: str,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponentVersionSummary, None, None]:
"""List MCP connection configuration component versions."""
return _list_configuration_versions(
models.COMPONENT_TYPE_MCP_CONNECTION,
component_id,
include_deleted,
limit,
after_key,
sort_order,
)
[docs]
@_configuration_endpoint
def list_evaluator_configuration_versions(
component_id: str,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponentVersionSummary, None, None]:
"""List evaluator configuration component versions."""
return _list_configuration_versions(
models.COMPONENT_TYPE_EVALUATOR,
component_id,
include_deleted,
limit,
after_key,
sort_order,
)
[docs]
@_configuration_endpoint
def list_shared_configuration_versions(
component_id: str,
include_deleted: bool | None = None,
limit: int | None = None,
after_key: str | None = None,
sort_order: str | None = None,
) -> Generator[models.ConfigurationComponentVersionSummary, None, None]:
"""List shared configuration component versions."""
return _list_configuration_versions(
models.COMPONENT_TYPE_SHARED,
component_id,
include_deleted,
limit,
after_key,
sort_order,
)
[docs]
@_configuration_endpoint
def delete_workflow_configuration_version(version_id: str) -> None:
"""Delete a workflow configuration component version."""
_delete_configuration_version(models.COMPONENT_TYPE_WORKFLOW, version_id)
[docs]
@_configuration_endpoint
def delete_agent_configuration_version(version_id: str) -> None:
"""Delete an agent configuration component version."""
_delete_configuration_version(models.COMPONENT_TYPE_AGENT, version_id)
[docs]
@_configuration_endpoint
def delete_mcp_connection_configuration_version(version_id: str) -> None:
"""Delete an MCP connection configuration component version."""
_delete_configuration_version(models.COMPONENT_TYPE_MCP_CONNECTION, version_id)
[docs]
@_configuration_endpoint
def delete_evaluator_configuration_version(version_id: str) -> None:
"""Delete an evaluator configuration component version."""
_delete_configuration_version(models.COMPONENT_TYPE_EVALUATOR, version_id)
[docs]
@_configuration_endpoint
def delete_shared_configuration_version(version_id: str) -> None:
"""Delete a shared configuration component version."""
_delete_configuration_version(models.COMPONENT_TYPE_SHARED, version_id)