Step Types

Multi-Agent Planning supports five step types for different operations. Each step type has specific configuration options and use cases.

Overview

Step Type	Description	Use Case
llm-call	Invoke an LLM for inference	Text generation, analysis, reasoning
connector-call	Query an MCP connector	Database queries, internal APIs
conditional	Branch based on conditions	Decision logic, routing
function-call	Execute a custom function	Data transformation, validation
api-call	Call an external HTTP API	Third-party integrations

llm-call

The llm-call step type invokes a language model for inference tasks.

Configuration

steps:
  - name: analyze-sentiment
    type: llm-call
    agent: sentiment-analyzer    # Agent name (uses agent's LLM config)
    input:
      text: "{{input.userMessage}}"
    output:
      sentiment: "{{result.sentiment}}"
      confidence: "{{result.confidence}}"

Fields

Field	Type	Required	Description
name	string	Yes	Unique step identifier
type	string	Yes	Must be llm-call
agent	string	Yes	Agent name to execute
input	object	Yes	Input data for the agent
output	object	No	Output mapping
timeout	string	No	Step timeout (overrides agent default)
retryPolicy	object	No	Retry configuration

Example: Text Analysis

apiVersion: axonflow.io/v1
kind: AgentConfig
metadata:
  name: text-analyzer
  domain: generic
spec:
  type: specialist
  description: Analyze text for key information
  capabilities:
    - text_analysis
    - entity_extraction
  llm:
    provider: openai
    model: gpt-4
    temperature: 0.3
  promptTemplate: |
    Analyze the following text and extract:
    1. Main topics
    2. Key entities (people, places, organizations)
    3. Sentiment (positive, negative, neutral)

    Text: {{input.text}}

    Return as JSON:
    {
      "topics": [...],
      "entities": [...],
      "sentiment": "..."
    }

Step definition:

steps:
  - name: analyze-text
    type: llm-call
    agent: text-analyzer
    input:
      text: "{{input.document}}"

Example: Multi-Turn Reasoning

steps:
  - name: initial-analysis
    type: llm-call
    agent: reasoning-agent
    input:
      question: "{{input.question}}"
      context: "{{input.context}}"

  - name: deep-analysis
    type: llm-call
    agent: reasoning-agent
    dependsOn: [initial-analysis]
    input:
      question: "{{input.question}}"
      previousAnalysis: "{{steps.initial-analysis.output}}"
      instruction: "Build on the initial analysis and provide deeper insights"

connector-call

The connector-call step type queries MCP connectors for data operations.

Configuration

steps:
  - name: fetch-user-data
    type: connector-call
    connector:
      name: postgresql           # Connector ID
      operation: query           # Operation type
      parameters:
        query: "SELECT * FROM users WHERE id = $1"
        args: ["{{input.userId}}"]
    output:
      user: "{{result.rows[0]}}"

Fields

Field	Type	Required	Description
name	string	Yes	Unique step identifier
type	string	Yes	Must be connector-call
connector	object	Yes	Connector configuration
connector.name	string	Yes	Connector ID
connector.operation	string	Yes	Operation to perform
connector.parameters	object	No	Operation parameters
output	object	No	Output mapping
timeout	string	No	Step timeout

Supported Connectors

Connector	Operations	Description
postgresql	query, execute	PostgreSQL database
mysql	query, execute	MySQL database
mongodb	find, aggregate, insert	MongoDB
redis	get, set, hget, hset	Redis cache
s3	get, put, list	AWS S3 storage
http	get, post, put, delete	HTTP API calls

Example: Database Query

steps:
  - name: get-customer-orders
    type: connector-call
    connector:
      name: postgresql
      operation: query
      parameters:
        query: |
          SELECT o.*, c.name as customer_name
          FROM orders o
          JOIN customers c ON o.customer_id = c.id
          WHERE c.id = $1
          ORDER BY o.created_at DESC
          LIMIT 10
        args: ["{{input.customerId}}"]
    output:
      orders: "{{result.rows}}"

Example: API Integration

steps:
  - name: search-flights
    type: connector-call
    connector:
      name: amadeus
      operation: flightOffers
      parameters:
        origin: "{{input.origin}}"
        destination: "{{input.destination}}"
        departureDate: "{{input.date}}"
        adults: 1
    output:
      flights: "{{result.data}}"

Example: Cache Lookup

steps:
  - name: check-cache
    type: connector-call
    connector:
      name: redis
      operation: get
      parameters:
        key: "user:{{input.userId}}:preferences"
    output:
      cachedPreferences: "{{result.value}}"

conditional

The conditional step type branches execution based on conditions.

Configuration

steps:
  - name: route-request
    type: conditional
    condition: "{{input.requestType}}"
    branches:
      - when: "support"
        goto: handle-support
      - when: "billing"
        goto: handle-billing
      - when: "technical"
        goto: handle-technical
      - default:
        goto: handle-general

Fields

Field	Type	Required	Description
name	string	Yes	Unique step identifier
type	string	Yes	Must be conditional
condition	string	Yes	Condition expression
branches	array	Yes	Branch definitions
branches[].when	string	Yes*	Value to match
branches[].if	string	Yes*	Boolean expression
branches[].goto	string	Yes	Target step name
branches[].default	boolean	No	Default branch flag

*Use either when (value match) or if (expression) per branch.

Example: Value Matching

steps:
  - name: route-by-intent
    type: conditional
    condition: "{{steps.classify-intent.output.intent}}"
    branches:
      - when: "book_flight"
        goto: flight-booking-flow
      - when: "book_hotel"
        goto: hotel-booking-flow
      - when: "cancel_booking"
        goto: cancellation-flow
      - default:
        goto: general-inquiry

Example: Boolean Expressions

steps:
  - name: check-eligibility
    type: conditional
    branches:
      - if: "{{input.age}} >= 18 && {{input.verified}} == true"
        goto: process-application
      - if: "{{input.age}} < 18"
        goto: require-guardian
      - default:
        goto: require-verification

Example: Nested Conditions

steps:
  - name: tier-routing
    type: conditional
    condition: "{{input.customerTier}}"
    branches:
      - when: "platinum"
        goto: platinum-service
      - when: "gold"
        goto: gold-service
      - default:
        goto: standard-service

  - name: platinum-service
    type: conditional
    branches:
      - if: "{{input.urgency}} == 'high'"
        goto: immediate-response
      - default:
        goto: priority-queue

function-call

The function-call step type executes custom functions for data transformation.

Configuration

steps:
  - name: transform-data
    type: function-call
    function:
      name: formatCurrency       # Function name
      module: utils              # Optional module
    input:
      amount: "{{steps.calculate.output.total}}"
      currency: "USD"
    output:
      formattedTotal: "{{result}}"

Fields

Field	Type	Required	Description
name	string	Yes	Unique step identifier
type	string	Yes	Must be function-call
function	object	Yes	Function configuration
function.name	string	Yes	Function name
function.module	string	No	Module containing function
input	object	Yes	Function arguments
output	object	No	Output mapping

Built-in Functions

Function	Description	Input	Output
formatDate	Format timestamp	{timestamp, format}	Formatted string
formatCurrency	Format currency	{amount, currency}	Formatted string
parseJson	Parse JSON string	{json}	Object
stringifyJson	Convert to JSON	{object}	JSON string
calculateSum	Sum array values	{values}	Number
filterArray	Filter array	{array, condition}	Filtered array
mergeObjects	Merge objects	{objects}	Merged object

Example: Data Transformation

steps:
  - name: fetch-transactions
    type: connector-call
    connector:
      name: postgresql
      operation: query
      parameters:
        query: "SELECT * FROM transactions WHERE user_id = $1"
        args: ["{{input.userId}}"]

  - name: calculate-total
    type: function-call
    dependsOn: [fetch-transactions]
    function:
      name: calculateSum
    input:
      values: "{{steps.fetch-transactions.output.rows}}"
      field: "amount"
    output:
      totalAmount: "{{result}}"

  - name: format-total
    type: function-call
    dependsOn: [calculate-total]
    function:
      name: formatCurrency
    input:
      amount: "{{steps.calculate-total.output.totalAmount}}"
      currency: "{{input.currency}}"
    output:
      displayTotal: "{{result}}"

Example: Data Validation

steps:
  - name: validate-input
    type: function-call
    function:
      name: validateSchema
      module: validation
    input:
      data: "{{input}}"
      schema:
        required: ["email", "name"]
        properties:
          email:
            type: string
            format: email
          name:
            type: string
            minLength: 2
    output:
      isValid: "{{result.valid}}"
      errors: "{{result.errors}}"

api-call

The api-call step type makes HTTP requests to external APIs.

Configuration

steps:
  - name: send-notification
    type: api-call
    request:
      method: POST
      url: "https://api.slack.com/api/chat.postMessage"
      headers:
        Authorization: "Bearer {{secrets.SLACK_TOKEN}}"
        Content-Type: "application/json"
      body:
        channel: "{{input.channel}}"
        text: "{{input.message}}"
    output:
      messageId: "{{response.body.ts}}"

Fields

Field	Type	Required	Description
name	string	Yes	Unique step identifier
type	string	Yes	Must be api-call
request	object	Yes	HTTP request configuration
request.method	string	Yes	HTTP method (GET, POST, PUT, DELETE)
request.url	string	Yes	Request URL
request.headers	object	No	Request headers
request.body	object	No	Request body
request.query	object	No	Query parameters
timeout	string	No	Request timeout
retryPolicy	object	No	Retry configuration
output	object	No	Output mapping

Example: REST API Call

steps:
  - name: create-ticket
    type: api-call
    request:
      method: POST
      url: "https://api.zendesk.com/api/v2/tickets"
      headers:
        Authorization: "Basic {{secrets.ZENDESK_AUTH}}"
        Content-Type: "application/json"
      body:
        ticket:
          subject: "{{input.subject}}"
          description: "{{input.description}}"
          priority: "{{input.priority}}"
          requester:
            email: "{{input.userEmail}}"
    output:
      ticketId: "{{response.body.ticket.id}}"
      ticketUrl: "{{response.body.ticket.url}}"

Example: Webhook Notification

steps:
  - name: notify-completion
    type: api-call
    request:
      method: POST
      url: "{{input.webhookUrl}}"
      headers:
        Content-Type: "application/json"
        X-Signature: "{{secrets.WEBHOOK_SECRET}}"
      body:
        event: "plan_completed"
        planId: "{{context.planId}}"
        status: "success"
        result: "{{steps.final-step.output}}"
        completedAt: "{{context.timestamp}}"

Example: API with Authentication

steps:
  - name: get-oauth-token
    type: api-call
    request:
      method: POST
      url: "https://oauth.example.com/token"
      headers:
        Content-Type: "application/x-www-form-urlencoded"
      body:
        grant_type: "client_credentials"
        client_id: "{{secrets.CLIENT_ID}}"
        client_secret: "{{secrets.CLIENT_SECRET}}"
    output:
      accessToken: "{{response.body.access_token}}"

  - name: call-protected-api
    type: api-call
    dependsOn: [get-oauth-token]
    request:
      method: GET
      url: "https://api.example.com/protected/resource"
      headers:
        Authorization: "Bearer {{steps.get-oauth-token.output.accessToken}}"
    output:
      data: "{{response.body}}"

Step Dependencies

Control execution order with dependsOn:

steps:
  - name: step-a
    type: llm-call
    agent: analyzer

  - name: step-b
    type: llm-call
    agent: summarizer
    dependsOn: [step-a]          # Runs after step-a

  - name: step-c
    type: connector-call
    connector: { ... }
    dependsOn: [step-a, step-b]  # Runs after both step-a and step-b

Parallel Execution

Group steps for parallel execution with parallelGroup:

steps:
  - name: search-flights
    type: connector-call
    parallelGroup: search        # These run in parallel
    connector: { ... }

  - name: search-hotels
    type: connector-call
    parallelGroup: search        # These run in parallel
    connector: { ... }

  - name: search-cars
    type: connector-call
    parallelGroup: search        # These run in parallel
    connector: { ... }

  - name: combine-results
    type: llm-call
    dependsOn: [search-flights, search-hotels, search-cars]
    agent: trip-planner

Error Handling

Configure error handling per step:

steps:
  - name: risky-operation
    type: api-call
    request: { ... }
    retryPolicy:
      maxRetries: 3
      initialDelay: 1s
      backoffMultiplier: 2.0
    onError:
      action: continue           # continue, fail, goto
      goto: fallback-step        # Target step if action is goto
      output:
        error: "{{error.message}}"

Error Actions

Action	Description
fail	Stop plan execution (default)
continue	Continue to next step
goto	Jump to specified step

Template Variables Reference

Variable	Description	Example
{{input.key}}	Input data	{{input.userId}}
{{steps.name.output}}	Step output	{{steps.analyze.output}}
{{steps.name.output.key}}	Specific output field	{{steps.analyze.output.score}}
{{context.planId}}	Plan ID	plan_abc123
{{context.timestamp}}	Current timestamp	2025-12-18T10:30:00Z
{{context.requestId}}	Request ID	req_xyz789
{{secrets.KEY}}	Secret value	{{secrets.API_KEY}}
{{result}}	Current step result	Used in output mapping
{{response.body}}	API response body	For api-call steps
{{error.message}}	Error message	In onError handlers

Custom Step Types

You can define and register custom step types to extend the planning engine with domain-specific logic.

Defining a Custom Step Type

Implement the StepExecutor interface in Go:

package steps

import (
    "context"
    "axonflow/platform/orchestrator/planning"
)

// CustomStepExecutor handles your custom step type.
type CustomStepExecutor struct {
    // Add any dependencies your step needs
}

func (e *CustomStepExecutor) Type() string {
    return "custom-transform"
}

func (e *CustomStepExecutor) Execute(ctx context.Context, step planning.Step, input map[string]interface{}) (*planning.StepResult, error) {
    // Your custom logic here
    transformed := transformData(input)
    return &planning.StepResult{
        Output: transformed,
        Status: planning.StepStatusCompleted,
    }, nil
}

Registering with the Planning Engine

Register your custom step type at startup:

engine := planning.NewEngine(config)
engine.RegisterStepType("custom-transform", &steps.CustomStepExecutor{})

Using in Workflow Definitions

Once registered, use the custom step type like any built-in type:

steps:
  - name: transform-payload
    type: custom-transform
    input:
      data: "{{steps.fetch-data.output.rows}}"
      format: "normalized"
    output:
      result: "{{result}}"

Best Practices for Custom Steps

1. Keep steps focused -- each custom step should do one thing well.
2. Return structured output -- use maps or JSON so downstream steps can access fields.
3. Handle errors gracefully -- return descriptive errors that help with debugging.
4. Add timeouts -- respect the step timeout to avoid blocking the entire plan.
5. Log execution -- emit structured logs for observability in the execution viewer.

Next Steps

• Planning Patterns - Common orchestration patterns
• Orchestration API Reference - Complete API documentation
• Agent Configuration - Agent YAML schema