Pipelines Config

The Airia Python SDK provides methods to retrieve pipeline configuration information. This allows you to access detailed information about pipelines, their versions, steps, and associated metadata.

Get Pipelines Configuration List

The get_pipelines_config method allows you to retrieve a list of all pipeline configurations, optionally filtered by project ID.

from airia import AiriaClient

# Initialize client
client = AiriaClient()

# Get all pipeline configurations
pipelines = client.pipelines_config.get_pipelines_config()

print(f"Total pipelines: {pipelines.total_count}")

# Iterate through pipeline configurations
for pipeline in pipelines.items:
    print(f"Pipeline: {pipeline.name}")
    print(f"  ID: {pipeline.id}")
    print(f"  Execution Name: {pipeline.execution_name}")
    print(f"  Description: {pipeline.description}")
    print(f"  Industry: {pipeline.industry}")

    # Access execution statistics
    if pipeline.execution_stats:
        print(f"  Success Count: {pipeline.execution_stats.success_count}")
        print(f"  Failure Count: {pipeline.execution_stats.failure_count}")
    else:
        print("  No execution statistics available")

    # Access deployment information
    if pipeline.deployment:
        print(f"  Deployment: {pipeline.deployment.name}")
        print(f"  Deployment Type: {pipeline.deployment.deployment_type}")
        print(f"  Is Recommended: {pipeline.deployment.is_recommended}")

    # Access active version information
    print(f"  Active Version: {pipeline.active_version.version_number}")
    print(f"  Version Count: {len(pipeline.versions)}")
    print()

Filter by Project ID

You can filter pipelines by project ID to get only pipelines belonging to a specific project:

# Get pipelines for a specific project
project_pipelines = client.pipelines_config.get_pipelines_config(
    project_id="your-project-id"
)

print(f"Project has {project_pipelines.total_count} pipelines")

for pipeline in project_pipelines.items:
    print(f"- {pipeline.name} ({pipeline.execution_name})")

Get Pipeline Configuration

The get_pipeline_config method allows you to retrieve detailed configuration information for a specific pipeline.

from airia import AiriaClient

# Initialize client
client = AiriaClient()

# Get pipeline configuration
pipeline_config = client.pipelines_config.get_pipeline_config(pipeline_id="your-pipeline-id")

# Access pipeline information
print(f"Pipeline Name: {pipeline_config.name}")
print(f"Description: {pipeline_config.description}")
print(f"Industry: {pipeline_config.industry}")
print(f"Is Seeded: {pipeline_config.is_seeded}")

# Access execution statistics
if pipeline_config.execution_stats:
    print(f"Success Count: {pipeline_config.execution_stats.success_count}")
    print(f"Failure Count: {pipeline_config.execution_stats.failure_count}")

# Access deployment information
if pipeline_config.deployment:
    print(f"Deployment Name: {pipeline_config.deployment.name}")
    print(f"Deployment Description: {pipeline_config.deployment.description}")
    print(f"Is Recommended: {pipeline_config.deployment.is_recommended}")

Working with Pipeline Versions

# Access pipeline versions
for version in pipeline_config.versions:
    print(f"Version: {version.version_number}")
    print(f"Is Latest: {version.is_latest}")
    print(f"Is Draft: {version.is_draft_version}")
    print(f"Steps Count: {len(version.steps or [])}")

    # Access individual steps
    if version.steps:
        for step in version.steps:
            print(f"  Step: {step.step_title} (Type: {step.step_type})")

Export Pipeline Definition

The export_pipeline_definition method allows you to export the complete definition of a pipeline including all its components such as metadata, agent configuration, data sources, prompts, tools, models, memories, Python code blocks, routers, and deployment information.

from airia import AiriaClient

# Initialize client
client = AiriaClient()

# Export complete pipeline definition
pipeline_export = client.pipelines_config.export_pipeline_definition(
    pipeline_id="your-pipeline-id"
)

# Access metadata
print(f"Export Version: {pipeline_export.metadata.export_version}")
print(f"Tagline: {pipeline_export.metadata.tagline}")
print(f"Industry: {pipeline_export.metadata.industry}")
print(f"Tasks: {pipeline_export.metadata.tasks}")
print(f"Credential Export Option: {pipeline_export.metadata.credential_export_option}")
print(f"Data Source Export Option: {pipeline_export.metadata.data_source_export_option}")

# Access agent configuration
print(f"Agent Name: {pipeline_export.agent.name}")
print(f"Agent Description: {pipeline_export.agent.agent_description}")
print(f"Execution Name: {pipeline_export.agent.execution_name}")
print(f"Steps Count: {len(pipeline_export.agent.steps)}")

# Access data sources
if pipeline_export.data_sources:
    print(f"Data Sources Count: {len(pipeline_export.data_sources)}")
    for data_source in pipeline_export.data_sources:
        print(f"  - {data_source.name} (Type: {data_source.data_source_type})")
        print(f"    Database Type: {data_source.database_type}")
        print(f"    Embedding Provider: {data_source.embedding_provider}")
        print(f"    File Count: {data_source.file_count}")

# Access prompts
if pipeline_export.prompts:
    print(f"Prompts Count: {len(pipeline_export.prompts)}")
    for prompt in pipeline_export.prompts:
        print(f"  - {prompt.name}")
        print(f"    Messages: {len(prompt.prompt_message_list)}")

# Access tools
if pipeline_export.tools:
    print(f"Tools Count: {len(pipeline_export.tools)}")
    for tool in pipeline_export.tools:
        print(f"  - {tool.name} (Type: {tool.tool_type})")
        print(f"    Endpoint: {tool.api_endpoint}")
        print(f"    Method: {tool.method_type}")

# Access models
if pipeline_export.models:
    print(f"Models Count: {len(pipeline_export.models)}")
    for model in pipeline_export.models:
        print(f"  - {model.display_name}")
        print(f"    Provider: {model.provider}")
        print(f"    Tool Support: {model.has_tool_support}")

# Access deployment information
if pipeline_export.deployment:
    print(f"Deployment Name: {pipeline_export.deployment.name}")
    print(f"Deployment Description: {pipeline_export.deployment.deployment_description}")
    print(f"Is Recommended: {pipeline_export.deployment.is_recommended}")
    print(f"Tags: {pipeline_export.deployment.tags}")

Working with Pipeline Steps from Export

# Access individual steps from the exported pipeline
for step in pipeline_export.agent.steps:
    print(f"Step: {step.step_title} (Type: {step.step_type})")
    print(f"  Position: ({step.position.x}, {step.position.y})")
    print(f"  Handles: {len(step.handles)}")
    print(f"  Dependencies: {len(step.dependencies_object)}")

    # Access step configuration
    if step.temperature is not None:
        print(f"  Temperature: {step.temperature}")
    if step.model_id:
        print(f"  Model ID: {step.model_id}")
    if step.prompt_id:
        print(f"  Prompt ID: {step.prompt_id}")
    if step.tool_ids:
        print(f"  Tool IDs: {step.tool_ids}")
    if step.data_source_id:
        print(f"  Data Source ID: {step.data_source_id}")

Working with Python Code Blocks and Memories

# Access Python code blocks
if pipeline_export.python_code_blocks:
    print(f"Python Code Blocks Count: {len(pipeline_export.python_code_blocks)}")
    for code_block in pipeline_export.python_code_blocks:
        print(f"  Code Block ID: {code_block.id}")
        print(f"  Code Preview: {code_block.code[:100]}...")

# Access memories
if pipeline_export.memories:
    print(f"Memories Count: {len(pipeline_export.memories)}")
    for memory in pipeline_export.memories:
        print(f"  - {memory.name}")
        print(f"    User Specific: {memory.is_user_specific}")

# Access routers
if pipeline_export.routers:
    print(f"Routers Count: {len(pipeline_export.routers)}")
    for router in pipeline_export.routers:
        print(f"  Router ID: {router.id}")
        print(f"  Model ID: {router.model_id}")
        print(f"  Router Config Keys: {list(router.router_config.keys())}")

Delete Pipeline

The delete_pipeline method allows you to permanently delete a pipeline and all its configuration from the Airia platform.

from airia import AiriaClient

# Initialize client
client = AiriaClient()

# Delete a pipeline
client.pipelines_config.delete_pipeline(pipeline_id="your-pipeline-id")

print("Pipeline deleted successfully")

Warning

This operation is permanent and cannot be reversed. Make sure you have the correct pipeline ID before deleting a pipeline. Consider exporting the pipeline definition first as a backup.

Async Usage

All methods are also available in the async client:

from airia import AiriaAsyncClient

# Initialize async client
async_client = AiriaAsyncClient()

# Get pipelines configuration list (async)
pipelines = await async_client.pipelines_config.get_pipelines_config()

# Get pipelines for a specific project (async)
project_pipelines = await async_client.pipelines_config.get_pipelines_config(
    project_id="your-project-id"
)

# Get pipeline configuration (async)
pipeline_config = await async_client.pipelines_config.get_pipeline_config(
    pipeline_id="your-pipeline-id"
)

# Export pipeline definition (async)
pipeline_export = await async_client.pipelines_config.export_pipeline_definition(
    pipeline_id="your-pipeline-id"
)

# Delete a pipeline (async)
await async_client.pipelines_config.delete_pipeline(
    pipeline_id="your-pipeline-id"
)

# Process the results as shown in the sync examples above