Skip to content

Pipelex API Guide

This guide covers everything you need to know about using the Pipelex API to execute pipelines with flexible input formats.

Table of Contents

  1. Quick Reference
  2. Execute Pipeline
  3. Input Format: PipelineInputs - Case 1: Direct Content Format - Case 2: Explicit Format (Concept and Content)
  4. Search Domains Explained
  5. PLX Content: Execute Pipelines Inline
  6. Pipe Builder: Generate Pipelines from Natural Language - Build Pipeline - Generate Runner Code
  7. Error Handling
  8. Best Practices
  9. Client Libraries

Base URL

https://api.pipelex.ai/api/v1

Authentication

Include your API key in the Authorization header:

Authorization: Bearer YOUR_API_KEY

Quick Reference

Endpoints at a Glance

POST /pipeline/execute        - Execute a pipeline
POST /pipe-builder/build      - Generate pipeline from natural language
POST /pipe-builder/generate-runner - Generate Python code for a pipeline

Input Formats at a Glance

// 1. Simple text (Case 1.1)
{ "text": "Hello, world!" }

// 2. List of text (Case 1.2)
{ "documents": ["text1", "text2", "text3"] }

// 3. Explicit format (Case 2.1)
{ "text": { "concept": "Text", "content": "Hello!" } }

// 4. Structured data (Case 2.5)
{ "invoice": { "concept": "Invoice", "content": { "number": "001", "amount": 100 } } }

// 5. Domain-prefixed concept (Case 2.5)
{ "data": { "concept": "accounting.Invoice", "content": { ... } } }

// 6. DictStuff instance (Python clients only)
DictStuff(concept="Invoice", content={...})

Key Points

✅ DO: Use direct strings for text: {"text": "Hello"}
❌ DON'T: Over-complicate text: {"text": {"concept": "Text", "content": "Hello"}}

✅ DO: Specify domain when ambiguous: "accounting.Invoice"
❌ DON'T: Use generic names when you have duplicates: "Invoice"

✅ DO: Provide either pipe_code (for library pipes) or plx_content (for inline bundles)
✅ DO: Combine both to specify which pipe to run from an inline bundle
❌ DON'T: Omit both pipe_code and plx_content - at least one is required

Execute Pipeline

Execute a Pipelex pipeline with flexible inputs.

Endpoint: POST /pipeline/execute

Request Body:

{
  "pipe_code": null,
  "plx_content": null,
  "inputs": {
    "input_name": "simple text or object"
  },
  "output_name": null,
  "output_multiplicity": null,
  "dynamic_output_concept_code": null,
}

Request Fields:

  • pipe_code (string, optional): Code of the pipe to execute. Can reference a pre-registered pipe in the library, or specify which pipe to run from the plx_content bundle
  • plx_content (string, optional): Inline pipeline bundle definition (.plx file content) - see PLX Content
  • inputs (PipelineInputs): Flexible input format - see Input Format: PipelineInputs below
  • output_name (string, optional): Name for the output slot
  • output_multiplicity (string, optional): Output multiplicity setting
  • dynamic_output_concept_code (string, optional): Override output concept

Note: Either pipe_code or plx_content (or both) must be provided: - If only pipe_code: Executes a pre-registered pipe from the library - If only plx_content: The bundle must have a main_pipe property - If both: Executes the specified pipe_code from the plx_content bundle

Response:

{
  "status": "success",
  "message": null,
  "error": null,
  "pipeline_run_id": "abc123...",
  "created_at": "2025-10-20T12:00:00Z",
  "pipeline_state": "COMPLETED",
  "finished_at": "2025-10-20T12:00:05Z",
  "pipe_output": {
    "working_memory": {
      "root": { ... },
      "aliases": { ... }
    },
    "pipeline_run_id": "abc123..."
  },
  "main_stuff_name": "result"
}

Input Format: PipelineInputs

Run your pipeline with flexible inputs that adapt to your needs. Pipelex supports multiple formats for providing inputs, making it easy to work with simple text, structured data, or complex objects.

What is PipelineInputs?

The inputs field uses PipelineInputs format - a smart, flexible way to provide data to your pipelines. Instead of forcing you into a rigid structure, PipelineInputs intelligently interprets your data based on how you provide it.

How Input Formatting Works

Case 1: Direct Content - Provide the value directly (simplest)

  • 1.1: String → "my text"
  • 1.2: List of strings → ["text1", "text2"]
  • 1.3: StructuredContent object → MyClass(arg1="value")
  • 1.4: List of StuffContent objects → [MyClass(...), MyClass(...)]
  • 1.5: ListContent of StuffContent objects → ListContent(items=[MyClass(...), MyClass(...)])

Note: Cases 1.3 and 1.5 are at the same level - both handle content types that inherit from StuffContent, but for different purposes (custom classes vs. list wrappers).

Case 2: Explicit Format - Use {"concept": "...", "content": "..."} for control (plain dict or DictStuff instance)

  • 2.1: String with concept → {"concept": "Text", "content": "my text"}
  • 2.2: List of strings with concept → {"concept": "Text", "content": ["text1", "text2"]}
  • 2.3: StructuredContent object with concept → {"concept": "Invoice", "content": InvoiceObject}
  • 2.4: List of StructuredContent objects with concept → {"concept": "Invoice", "content": [...]}
  • 2.5: Dictionary (structured data) → {"concept": "Invoice", "content": {"field": "value"}}
  • 2.6: List of dictionaries → {"concept": "Invoice", "content": [{...}, {...}]}

Pro Tip: For text inputs specifically, skip the verbose format. Just provide the string directly: "text": "Hello" instead of "text": {"concept": "Text", "content": "Hello"}

Case 1: Direct Content Format

When you provide content directly (without the concept key), Pipelex intelligently infers the type.

1.1: Simple String (Text)

The simplest case - just provide a string directly:

{
  "inputs": {
    "my_text": "my text"
  }
}

Result: Automatically becomes TextContent with concept native.Text

1.2: List of Strings (Text List)

Provide multiple text items as a list:

{
  "inputs": {
    "my_texts": ["my text1", "my text2", "my text3"]
  }
}

Result: Becomes a ListContent containing multiple TextContent items

Note: The concept must be compatible with native.Text or an error will be raised.

1.3: StructuredContent Object

Provide a structured object directly (for Python clients):

# Python client example
from my_project.domain.domain_struct import MyConcept, MySubClass

inputs = {
    "invoice_data": MyConcept(arg1="arg1", arg2=1, arg3=MySubClass(arg4="arg4"))
}

What is StructuredContent?

  • StructuredContent is the base class for user-defined data structures in Pipelex
  • You create your own classes by inheriting from StructuredContent
  • These classes are defined in your project's Python files
  • Learn more: Python StructuredContent Classes

Concept Resolution:

  • The system searches all available domains for a concept matching the class name
  • If multiple concepts with the same name exist in different domains → Error: Must specify domain
  • If no concept is found → Error

1.4: List of StuffContent Objects

Provide multiple content objects in a plain Python list:

# Python client example
inputs = {
    "invoice_list": [
        MyConcept(arg1="arg1", arg2=1, arg3=MySubClass(arg4="arg4")),
        MyConcept(arg1="arg1_2", arg2=2, arg3=MySubClass(arg4="arg4_2"))
    ]
}

What it accepts:

  • Lists of StructuredContent objects (user-defined classes)
  • Lists of native content objects (TextContent, ImageContent, etc.)

Requirements:

  • All items must be of the same type
  • Concept resolution follows the same rules as 1.3
  • Creates a new ListContent wrapper internally

1.5: ListContent of StuffContent Objects

Provide an existing ListContent wrapper object (Python clients):

# Python client example
from pipelex.core.stuffs.list_content import ListContent

inputs = {
    "invoice_list": ListContent(items=[
        MyConcept(arg1="arg1", arg2=1, arg3=MySubClass(arg4="arg4")),
        MyConcept(arg1="arg1_2", arg2=2, arg3=MySubClass(arg4="arg4_2"))
    ])
}

Key Difference from Case 1.4:

  • Case 1.4: Plain Python list [item1, item2] → Creates a new ListContent wrapper
  • Case 1.5: Already wrapped ListContent(items=[item1, item2]) → Uses the wrapper directly

Why Case 1.5 is Separate from Case 1.3:

  • StructuredContent and ListContent are sibling classes (both inherit from StuffContent)
  • Case 1.3 handles user-defined structured data classes
  • Case 1.5 handles list container wrappers
  • They're at the same inheritance level, not parent-child

Requirements:

  • All items within the ListContent must be StuffContent objects (this includes both StructuredContent and native content like TextContent, ImageContent)
  • All items must be of the same type
  • The ListContent cannot be empty
  • Concept is inferred from the first item's class name (not from "ListContent")

Use Case: This format is useful when you already have data wrapped in a ListContent object from a previous pipeline execution or when working with Pipelex's internal data structures.

Case 2: Explicit Format (Concept and Content)

Use the explicit format {"concept": "...", "content": "..."} when you need precise control over concept selection or when working with domain-specific concepts.

2.1: Explicit String Input

{
  "inputs": {
    "text": {
      "concept": "Text",
      "content": "my text"
    }
  }
}

Concept Options:

  • "Text" or "native.Text" for native text
  • Any custom concept that is strictly compatible with native.Text

2.2: Explicit List of Strings

{
  "inputs": {
    "documents": {
      "concept": "Text",
      "content": ["text1", "text2", "text3"]
    }
  }
}

Result: ListContent with multiple TextContent items

2.3: Structured Object with Concept

{
  "inputs": {
    "invoice_data": {
      "concept": "Invoice",
      "content": {
        "invoice_number": "INV-001",
        "amount": 1250.00,
        "date": "2025-10-20"
      }
    }
  }
}

Concept Resolution with Search Domains:

When you specify a concept name without a domain prefix:

  • ✅ If the concept exists in only one domain → Automatically found
  • ❌ If the concept exists in multiple domains → Error: "Multiple concepts found. Please specify domain as 'domain.Concept'"
  • ❌ If the concept doesn't exist → Error: "Concept not found"

Using Domain Prefix: 

{
  "concept": "accounting.Invoice"
}

This explicitly tells Pipelex to use the Invoice concept from the accounting domain.

2.4: List of Structured Objects

{
  "inputs": {
    "invoices": {
      "concept": "Invoice",
      "content": [
        {
          "invoice_number": "INV-001",
          "amount": 1250.00
        },
        {
          "invoice_number": "INV-002",
          "amount": 890.00
        }
      ]
    }
  }
}

Result: ListContent with multiple structured content items

2.5: Dictionary Content

Provide structured data as a dictionary:

{
  "inputs": {
    "person": {
      "concept": "PersonInfo",
      "content": {
        "arg1": "something",
        "arg2": 1,
        "arg3": {
          "arg4": "something else"
        }
      }
    }
  }
}

The system will: 1. Find the concept structure (with domain resolution as explained above) 2. Validate the dictionary against the concept's structure 3. Create the appropriate content object

2.6: List of Dictionaries

{
  "inputs": {
    "people": {
      "concept": "PersonInfo",
      "content": [
        {
          "arg1": "something",
          "arg2": 1,
          "arg3": {"arg4": "something else"}
        },
        {
          "arg1": "something else",
          "arg2": 2,
          "arg3": {"arg4": "something else else"}
        }
      ]
    }
  }
}

Using DictStuff Instances (Python Clients Only)

For Python clients, you can also pass DictStuff instances instead of plain dicts. DictStuff is a Pydantic model with the same structure as the explicit format.

from pipelex.client import PipelexClient
from pipelex.core.stuffs.stuff import DictStuff

client = PipelexClient(api_token="YOUR_API_KEY")

# Using DictStuff instance with dict content
response = await client.execute_pipeline(
    pipe_code="process_invoice",
    inputs={
        "invoice": DictStuff(
            concept="accounting.Invoice",
            content={
                "invoice_number": "INV-001",
                "amount": 1250.00,
                "date": "2025-10-20"
            }
        )
    }
)

# Using DictStuff instance with list of dicts content
response = await client.execute_pipeline(
    pipe_code="process_invoices",
    inputs={
        "invoices": DictStuff(
            concept="accounting.Invoice",
            content=[
                {"invoice_number": "INV-001", "amount": 1250.00},
                {"invoice_number": "INV-002", "amount": 890.00}
            ]
        )
    }
)

# Using DictStuff instance with list of strings (for Text concept)
response = await client.execute_pipeline(
    pipe_code="analyze_texts",
    inputs={
        "documents": DictStuff(
            concept="Text",
            content=["document 1", "document 2", "document 3"]
        )
    }
)

DictStuff Structure:

  • concept (str): The concept code (with optional domain prefix)
  • content (dict[str, Any] | list[Any]): The actual data content

Content Types:

  • Dictionary: Single structured object → Creates a single Stuff
  • List of dicts: Multiple structured objects → Creates ListContent with validated items
  • List of strings (for Text-compatible concepts): Creates ListContent of TextContent

Note: DictStuff instances are automatically converted to plain dicts and processed through the standard Case 2 logic.

Search Domains Explained

When you reference a concept by name (like "Invoice" or "PersonInfo"), Pipelex needs to find it in your loaded domains.

{
  "concept": "Invoice"
}

What happens: 1. Pipelex searches all available domains for a concept named "Invoice" 2. If found in exactly one domain → ✅ Uses that concept 3. If found in multiple domains → ❌ Error: "Ambiguous concept: Found 'Invoice' in domains: accounting, billing. Use 'domain.Invoice' format." 4. If not found → ❌ Error: "Concept 'Invoice' not found in any domain"

Explicit Domain Specification

To avoid ambiguity, specify the domain explicitly:

{
  "concept": "accounting.Invoice"
}

Format: "domain_name.ConceptName"

This tells Pipelex exactly which concept to use, bypassing the search.

Best Practices

  • Use simple names ("Invoice") when you have unique concept names across domains
  • Use domain-prefixed names ("accounting.Invoice") when:
  • You have concepts with the same name in different domains
  • You want to be explicit about which concept to use
  • You're building APIs that need to be unambiguous

Multiple Input Combinations

Combine different input types in a single request:

{
  "inputs": {
    "text": "Analyze this contract for risks.",
    "category": {
      "concept": "Category",
      "content": {"name": "legal", "priority": "high"}
    },
    "options": ["option1", "option2", "option3"],
    "invoice": {
      "concept": "accounting.Invoice",
      "content": {
        "invoice_number": "INV-001",
        "amount": 1250.00
      }
    }
  }
}

In this example:

  • text uses direct string format (Case 1.1)
  • category uses explicit format with structured content (Case 2.5)
  • options uses direct list format (Case 1.2)
  • invoice uses explicit format with domain prefix and structured content (Case 2.5)

PLX Content: Execute Pipelines Inline

The plx_content field allows you to provide an inline pipeline bundle definition (.plx file) for execution. The bundle defines the domain, concepts, and pipes.

When to Use PLX Content

Use plx_content when you want to:

  • Execute dynamically generated pipelines
  • Test pipelines without deploying them to the library
  • Run one-off pipelines that don't need to be stored
  • Prototype new pipelines quickly

Basic Usage

Provide the complete pipeline bundle definition in plx_content:

curl -X POST https://api.pipelex.ai/api/v1/pipeline/execute \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "plx_content": "domain = \"my_domain\"\nmain_pipe = \"hello\"\n\n[concept]\nGreeting = \"A greeting message\"\n\n[pipe.hello]\ntype = \"PipeLLM\"\ndescription = \"Generate a greeting\"\noutput = \"Greeting\"\nprompt = \"Generate a friendly greeting\"",
    "inputs": {
      "name": "Alice"
    }
  }'

PLX Content Structure

The plx_content must contain a valid .plx bundle as a string:

{
  "plx_content": "domain = \"example\"\nmain_pipe = \"my_pipe\"\n\n[concept]\nResult = \"A result\"\n\n[pipe.my_pipe]\ntype = \"PipeLLM\"\ndescription = \"Process input\"\ninputs = { text = \"Text\" }\noutput = \"Result\"\nprompt = \"Process this: @text\""
}

Key Points:

  • The plx_content must be a valid Pipelex pipeline bundle definition
  • Include newlines (\n) to properly format the pipeline
  • The pipeline will be validated before execution
  • All concepts and pipes must be properly defined within the bundle

Specifying Which Pipe to Run:

There are two ways to specify which pipe in the bundle to execute:

  1. Using main_pipe property (in the bundle): 

    domain = "my_domain"
main_pipe = "my_pipe"
    When main_pipe is defined, you don't need to provide pipe_code in the request.

  2. Using pipe_code parameter (in the request): 

    {
  "plx_content": "...",
  "pipe_code": "my_pipe"
}
    Use this when the bundle doesn't have a main_pipe property, or when you want to execute a different pipe than the main one.

Example: Dynamic Pipeline with main_pipe

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Define a pipeline bundle with main_pipe
plx_pipeline = """
domain = "sentiment_analysis"
main_pipe = "analyze"

[concept]
SentimentScore = "A sentiment score"

[concept.SentimentScore.structure]
score = { type = "number", description = "Sentiment score from -1 to 1", required = true }
label = { type = "text", description = "Sentiment label", required = true }

[pipe.analyze]
type = "PipeLLM"
description = "Analyze sentiment of text"
inputs = { text = "Text" }
output = "SentimentScore"
model = { model = "gpt-4o-mini", temperature = 0.3 }
prompt = \"\"\"
Analyze the sentiment of this text and provide a score from -1 (negative) to 1 (positive):

@text
\"\"\"
"""

# Execute the pipeline (no pipe_code needed because main_pipe is set)
response = await client.execute_pipeline(
    plx_content=plx_pipeline,
    inputs={
        "text": "This product exceeded all my expectations!"
    }
)

print(response.pipe_output)

Example: Specifying pipe_code

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Define a bundle with multiple pipes (no main_pipe)
plx_pipeline = """
domain = "sentiment_analysis"

[concept]
SentimentScore = "A sentiment score"

[concept.SentimentScore.structure]
score = { type = "number", description = "Sentiment score from -1 to 1", required = true }
label = { type = "text", description = "Sentiment label", required = true }

[pipe.analyze_positive]
type = "PipeLLM"
description = "Analyze positive sentiment"
inputs = { text = "Text" }
output = "SentimentScore"
model = { model = "gpt-4o-mini", temperature = 0.3 }
prompt = \"\"\"
Focus on positive aspects and provide a sentiment score:
@text
\"\"\"

[pipe.analyze_negative]
type = "PipeLLM"
description = "Analyze negative sentiment"
inputs = { text = "Text" }
output = "SentimentScore"
model = { model = "gpt-4o-mini", temperature = 0.3 }
prompt = \"\"\"
Focus on negative aspects and provide a sentiment score:
@text
\"\"\"
"""

# Execute a specific pipe from the bundle
response = await client.execute_pipeline(
    plx_content=plx_pipeline,
    pipe_code="analyze_positive",  # Specify which pipe to run
    inputs={
        "text": "This product exceeded all my expectations!"
    }
)

print(response.pipe_output)

Combining PLX Content with Inputs

You can use all the PipelineInputs input formats described above with plx_content:

{
  "plx_content": "domain = \"processing\"...",
  "inputs": {
    "text": "Simple direct input",
    "data": {
      "concept": "DataRecord",
      "content": {
        "id": 123,
        "values": [1, 2, 3]
      }
    }
  }
}

Validation

When you provide plx_content: 1. The pipeline bundle is parsed and validated 2. All concepts and structures are verified 3. If a main_pipe is specified in the bundle or pipe_code is provided in the request, that pipe is executed 4. If neither is provided, an error is returned 5. Errors are returned if the pipeline definition is invalid

Flexible Execution

You have three options for pipeline execution: - Library pipe only: Provide pipe_code alone to execute a pre-registered pipe - Inline bundle with main_pipe: Provide plx_content with a main_pipe property - Inline bundle with pipe_code: Provide both plx_content and pipe_code to specify which pipe to run

Pipe Builder: Generate Pipelines from Natural Language

The Pipe Builder API allows you to generate complete pipeline bundles from natural language descriptions.

Build Pipeline

Generate a pipeline bundle from a brief description.

Endpoint: POST /pipe-builder/build

Request Body:

{
  "brief": "Extract invoice data from PDFs and validate the amounts"
}

Request Fields:

  • brief (string, required): Brief description of the pipeline to build

Response:

{
  "plx_content": "domain = \"invoice_processor\"\nmain_pipe = \"process_invoice\"...",
  "pipelex_bundle_blueprint": {
    "domain": "invoice_processor",
    "main_pipe": "process_invoice",
    ...
  },
  "pipe_structures": {
    "process_invoice": {
      "inputs": {...},
      "output": {...}
    }
  },
  "success": true,
  "message": "Pipeline generated successfully"
}

Response Fields:

  • plx_content (string): Generated PLX content as a string (ready to use with /pipeline/execute)
  • pipelex_bundle_blueprint (object): Structured blueprint representation of the generated bundle
  • pipe_structures (object): Structure information for each pipe's inputs and outputs
  • success (boolean): Whether the operation was successful
  • message (string): Status message

Example:

curl -X POST https://api.pipelex.ai/api/v1/pipe-builder/build \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "brief": "Analyze customer sentiment from support tickets and categorize them by urgency"
  }'

Generate Runner Code

Generate Python runner code for a specific pipe from PLX content.

Endpoint: POST /pipe-builder/generate-runner

Request Body:

{
  "plx_content": "domain = \"my_domain\"...",
  "pipe_code": "my_pipe"
}

Request Fields:

  • plx_content (string, required): PLX bundle content to load
  • pipe_code (string, required): Code of the pipe to generate runner code for

Response:

{
  "python_code": "import asyncio\nfrom pipelex import Pipelex\n...",
  "pipe_code": "my_pipe",
  "success": true,
  "message": "Runner code generated successfully"
}

Response Fields:

  • python_code (string): Generated Python code for running the workflow
  • pipe_code (string): The pipe code that was used
  • success (boolean): Whether the operation was successful
  • message (string): Status message

What This Does:

  1. Parses and validates the PLX content
  2. Loads pipes from the bundle
  3. Validates and dry-runs all pipes
  4. Generates ready-to-use Python runner code
  5. Cleans up loaded pipes
  6. Returns executable Python code

Example:

curl -X POST https://api.pipelex.ai/api/v1/pipe-builder/generate-runner \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "plx_content": "domain = \"sentiment_analysis\"\nmain_pipe = \"analyze\"...",
    "pipe_code": "analyze"
  }'

Use Case:

This is useful for: - Getting starter code for your pipeline - Generating examples to understand how to run a pipeline - Creating standalone scripts from PLX definitions - Learning the Python API through generated examples

Complete Workflow Example

Here's a complete example showing how to build a pipeline from a description and then execute it:

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Step 1: Build a pipeline from natural language
build_response = await client.execute_pipeline(
    pipe_code="pipe_builder",
    inputs={
        "brief": "Extract key information from customer emails and classify by urgency"
    }
)

# Step 2: Get the generated PLX content
plx_content = build_response.pipe_output.working_memory.get_stuff_content("plx_content")

# Step 3: Execute the generated pipeline
execute_response = await client.execute_pipeline(
    plx_content=plx_content,
    inputs={
        "email": "Dear Support, I urgently need help with my account..."
    }
)

# Step 4: Use the results
result = execute_response.pipe_output.working_memory
print(f"Classification: {result}")

Error Handling

Error Response Format

{
  "status": "error",
  "message": "Pipeline execution failed",
  "error": "Invalid concept: UnknownConcept",
  "pipeline_run_id": "abc123",
  "pipeline_state": "FAILED"
}

HTTP Status Codes

  • 200 OK: Pipeline executed successfully
  • 400 Bad Request: Invalid input format
  • 401 Unauthorized: Missing or invalid API key
  • 404 Not Found: Pipeline not found
  • 500 Internal Server Error: Server error

Common Errors

Concept Not Found 

{
  "status": "error",
  "error": "Concept 'UnknownConcept' not found in the library"
}

Ambiguous Concept 

{
  "status": "error",
  "error": "Ambiguous concept: Found 'Invoice' in domains: accounting, billing. Use 'domain.Invoice' format."
}

Invalid Input Type 

{
  "status": "error",
  "error": "Trying to create a Stuff from a list of strings but the items are not of the same type"
}

Best Practices

Input Design

  1. Use Direct Format for Simple Inputs 

    // ✅ Good
{ "text": "Hello" }

// ❌ Unnecessary
{ "text": { "concept": "Text", "content": "Hello" } }

  2. Be Explicit with Custom Concepts 

    // ✅ Good - clear and unambiguous
{ "invoice": { "concept": "accounting.Invoice", "content": {...} } }

  3. Validate Before Sending

  • Ensure all required fields are present
  • Match field types to concept structure
  • Use domain prefixes when concepts might be ambiguous

Performance Tips

  1. Batch Processing: Use list inputs when processing multiple items
  2. Reuse Connections: Keep the client instance for multiple requests
  3. Handle Async Properly: Use proper async/await patterns in Python

Security

  1. Protect API Keys: Never commit API keys to version control
  2. Use Environment Variables: Store credentials securely 
    import os
api_token = os.getenv("PIPELEX_API_TOKEN")

Client Libraries

Python

Installation

pip install pipelex

Execution Options

The examples below show inline bundles with plx_content. You can also execute pre-registered pipes from the library using only pipe_code (without plx_content). See the PLX Content section for more details on execution options.

Executing Pre-Registered Library Pipes

If you have a pipe already registered in the library, you can execute it using only pipe_code:

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Execute a pre-registered library pipe (no plx_content needed)
response = await client.execute_pipeline(
    pipe_code="my_registered_pipe",
    inputs={
        "text": "Process this input"
    }
)

print(response.pipe_output.working_memory)

Basic Usage (Direct Text Input)

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Define your pipeline bundle
plx_bundle = """
domain = "sentiment"
main_pipe = "analyze_sentiment"

[concept]
Sentiment = "A sentiment analysis result"

[pipe.analyze_sentiment]
type = "PipeLLM"
description = "Analyze sentiment"
inputs = { text = "Text" }
output = "Sentiment"
model = { model = "gpt-4o-mini", temperature = 0.3 }
prompt = "Analyze the sentiment of: @text"
"""

# Simple text input - no need for concept/content structure
response = await client.execute_pipeline(
    plx_content=plx_bundle,
    inputs={
        "text": "This is amazing!"
    }
)

print(response.pipe_output.working_memory)

Structured Input

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Define pipeline bundle
plx_bundle = """
domain = "invoices"
main_pipe = "extract_invoice"

[concept]
InvoiceData = "Extracted invoice information"

[concept.InvoiceData.structure]
invoice_number = "Invoice number"
amount = { type = "number", description = "Invoice amount" }

[pipe.extract_invoice]
type = "PipeLLM"
description = "Extract invoice data"
inputs = { document = "PDF" }
output = "InvoiceData"
model = { model = "gpt-4o", temperature = 0.1 }
prompt = "Extract invoice data from: @document"
"""

# Explicit concept and structured content
response = await client.execute_pipeline(
    plx_content=plx_bundle,
    inputs={
        "document": {
            "concept": "PDF",
            "content": {
                "url": "https://example.com/invoice.pdf"
            }
        }
    }
)

# Access the main output
invoice = response.pipe_output.working_memory.root["invoice"]
print(f"Invoice Number: {invoice['content']['invoice_number']}")

Multiple Inputs with Mixed Formats

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Define pipeline bundle
plx_bundle = """
domain = "contracts"
main_pipe = "contract_analysis"

[concept]
AnalysisConfig = "Configuration for contract analysis"

[concept.AnalysisConfig.structure]
priority = "Analysis priority"
focus_areas = { type = "list", item_type = "text", description = "Areas to focus on" }

[concept]
AnalysisResult = "Result of contract analysis"

[pipe.contract_analysis]
type = "PipeLLM"
description = "Analyze contract"
inputs = { contract_text = "Text", analysis_config = "AnalysisConfig", reference_docs = "Text[]" }
output = "AnalysisResult"
model = { model = "gpt-4o", temperature = 0.3 }
prompt = \"\"\"
Analyze this contract with priority $analysis_config.priority:
@contract_text

Reference documents:
@reference_docs
\"\"\"
"""

response = await client.execute_pipeline(
    plx_content=plx_bundle,
    inputs={
        # Direct text input
        "contract_text": "This is the contract text...",

        # Structured input with explicit concept
        "analysis_config": {
            "concept": "AnalysisConfig",
            "content": {
                "priority": "high",
                "focus_areas": ["liability", "termination"]
            }
        },

        # Direct list input
        "reference_docs": ["doc1", "doc2", "doc3"]
    }
)

print(response.pipeline_state)
print(response.main_stuff_name)

Using StructuredContent Objects (Advanced)

from pipelex.client import PipelexClient
from my_project.invoices.invoice_struct import Invoice, InvoiceLineItem

client = PipelexClient(api_token="YOUR_API_KEY")

# Note: plx_content with pipeline bundle is required but omitted for brevity
# See earlier examples for complete bundle definitions

# Create StructuredContent objects directly
invoice_data = Invoice(
    invoice_number="INV-001",
    amount=1250.00,
    line_items=[
        InvoiceLineItem(description="Service A", amount=500.00),
        InvoiceLineItem(description="Service B", amount=750.00)
    ]
)

# Pass StructuredContent object directly (Case 1.3)
response = await client.execute_pipeline(
    plx_content=your_plx_bundle,  # Pipeline bundle required
    pipe_code="validate_invoice",
    inputs={
        "invoice": invoice_data
    }
)

Using Domain-Prefixed Concepts

from pipelex.client import PipelexClient

client = PipelexClient(api_token="YOUR_API_KEY")

# Note: plx_content with pipeline bundle is required but omitted for brevity
# See earlier examples for complete bundle definitions

response = await client.execute_pipeline(
    plx_content=your_plx_bundle,  # Pipeline bundle required
    pipe_code="multi_domain_processing",
    inputs={
        "accounting_invoice": {
            "concept": "accounting.Invoice",  # From accounting domain
            "content": {
                "invoice_number": "INV-001",
                "amount": 1250.00
            }
        },
        "billing_invoice": {
            "concept": "billing.Invoice",  # From billing domain
            "content": {
                "customer_id": "CUST-123",
                "period": "2025-10"
            }
        }
    }
)

Support

For questions or issues: