API Reference

This section provides comprehensive documentation for Pulsar’s Python API.

Core Classes

Pulsar Engine

class engine.executor.PulsarEngine(workflow, config=None)[source]

Bases: object

Main workflow execution engine for Pulsar.

Parameters:
__init__(workflow, config=None)[source]
Parameters:
async execute(user_input='')[source]

Execute the workflow with given input.

Parameters:

user_input (str)

Return type:

ExecutionResult

async execute_with_initial_state(initial_state)[source]

Execute the workflow with given initial state.

Parameters:

initial_state (Dict[str, Any])

Return type:

ExecutionResult

async execute_step(step)[source]

Execute a single step using appropriate handler.

Return type:

StepResult

get_current_state()[source]

Get current execution state (synchronous snapshot).

Return type:

Dict[str, Any]

async get_current_state_async()[source]

Get current execution state asynchronously.

Return type:

Dict[str, Any]

Advanced expression evaluator for Pulsar conditional execution.

Supports template variables, string operations, membership tests, and type checking.

exception engine.expression_evaluator.ExpressionError[source]

Bases: Exception

Custom exception for expression evaluation errors.

class engine.expression_evaluator.TokenType(value)[source]

Bases: Enum

Token types for expression parsing.

VARIABLE = 'variable'
STRING = 'string'
NUMBER = 'number'
BOOLEAN = 'boolean'
OPERATOR = 'operator'
FUNCTION = 'function'
PARENTHESIS = 'parenthesis'
EOF = 'eof'
class engine.expression_evaluator.Token(type, value, position)[source]

Bases: object

Token representation.

Parameters:
type: TokenType
value: Any
position: int
__init__(type, value, position)
Parameters:
Return type:

None

class engine.expression_evaluator.ExpressionParser(expression)[source]

Bases: object

Parser for expression language.

Parameters:

expression (str)

OPERATORS = {'!=': (3, <function ExpressionParser.<lambda>>), '%': (6, <function ExpressionParser.<lambda>>), '&&': (2, <function ExpressionParser.<lambda>>), '*': (6, <function ExpressionParser.<lambda>>), '+': (5, <function ExpressionParser.<lambda>>), '-': (5, <function ExpressionParser.<lambda>>), '/': (6, <function ExpressionParser.<lambda>>), '<': (4, <function ExpressionParser.<lambda>>), '<=': (4, <function ExpressionParser.<lambda>>), '==': (3, <function ExpressionParser.<lambda>>), '>': (4, <function ExpressionParser.<lambda>>), '>=': (4, <function ExpressionParser.<lambda>>), '||': (1, <function ExpressionParser.<lambda>>)}
FUNCTIONS: Dict[str, Callable[[...], Any]] = {'contains': <function ExpressionParser.<lambda>>, 'endsWith': <function ExpressionParser.<lambda>>, 'isArray': <function ExpressionParser.<lambda>>, 'isNumber': <function ExpressionParser.<lambda>>, 'isObject': <function ExpressionParser.<lambda>>, 'isString': <function ExpressionParser.<lambda>>, 'length': <function ExpressionParser.<lambda>>, 'lower': <function ExpressionParser.<lambda>>, 'startsWith': <function ExpressionParser.<lambda>>, 'trim': <function ExpressionParser.<lambda>>, 'upper': <function ExpressionParser.<lambda>>}
__init__(expression)[source]
Parameters:

expression (str)

parse()[source]

Parse the expression and return the AST root.

Return type:

ExpressionNode

class engine.expression_evaluator.ExpressionNode[source]

Bases: object

Base class for expression AST nodes.

class engine.expression_evaluator.LiteralNode(value)[source]

Bases: ExpressionNode

Literal value node.

Parameters:

value (Any)

value: Any
__init__(value)
Parameters:

value (Any)

Return type:

None

class engine.expression_evaluator.VariableNode(name)[source]

Bases: ExpressionNode

Variable reference node.

Parameters:

name (str)

name: str
__init__(name)
Parameters:

name (str)

Return type:

None

class engine.expression_evaluator.BinaryOpNode(left, operator, right)[source]

Bases: ExpressionNode

Binary operation node.

Parameters:
left: ExpressionNode
operator: str
right: ExpressionNode
__init__(left, operator, right)
Parameters:
Return type:

None

class engine.expression_evaluator.UnaryOpNode(operator, operand)[source]

Bases: ExpressionNode

Unary operation node.

Parameters:
operator: str
operand: ExpressionNode
__init__(operator, operand)
Parameters:
Return type:

None

class engine.expression_evaluator.FunctionCallNode(function_name, arguments)[source]

Bases: ExpressionNode

Function call node.

Parameters:
function_name: str
arguments: List[ExpressionNode]
__init__(function_name, arguments)
Parameters:
Return type:

None

class engine.expression_evaluator.ArrayLiteralNode(elements)[source]

Bases: ExpressionNode

Array literal node.

Parameters:

elements (List[ExpressionNode])

elements: List[ExpressionNode]
__init__(elements)
Parameters:

elements (List[ExpressionNode])

Return type:

None

class engine.expression_evaluator.ArrayAccessNode(array, index)[source]

Bases: ExpressionNode

Array access node (e.g., arr[0]).

Parameters:
array: ExpressionNode
index: ExpressionNode
__init__(array, index)
Parameters:
Return type:

None

class engine.expression_evaluator.ExpressionEvaluator(state)[source]

Bases: object

Evaluator for parsed expressions.

Parameters:

state (Dict[str, Any])

__init__(state)[source]
Parameters:

state (Dict[str, Any])

evaluate(node)[source]

Evaluate an expression node.

Parameters:

node (ExpressionNode)

Return type:

Any

engine.expression_evaluator.evaluate_expression(expression, state)[source]

Evaluate an expression with the given state.

Parameters:

  • expression (str) – Expression string to evaluate

  • state (Dict[str, Any]) – State dictionary for variable resolution

Returns:

Result of the expression evaluation

Raises:

ExpressionError – If expression is invalid or evaluation fails

Return type:

Any

engine.expression_evaluator.validate_expression(expression)[source]

Validate that an expression can be parsed without errors.

Parameters:

expression (str) – Expression string to validate

Returns:

True if expression is valid, False otherwise

Return type:

bool

class engine.results.StepResult(*, step_name, success, output=None, error=None, execution_time, started_at, completed_at, retries=0, metadata={})[source]

Bases: BaseModel

Result of executing a single step.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:
step_name: str
success: bool
output: Any | None
error: str | None
execution_time: float
started_at: datetime
completed_at: datetime
retries: int
metadata: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class engine.results.ExecutionResult(*, workflow_name, success, final_state, step_results, total_execution_time, started_at, completed_at, error=None, execution_history=[])[source]

Bases: BaseModel

Result of executing a complete workflow.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:
workflow_name: str
success: bool
final_state: Dict[str, Any]
step_results: List[StepResult]
total_execution_time: float
started_at: datetime
completed_at: datetime
error: str | None
execution_history: List[Dict[str, Any]]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Agent System

Base Agent

class agents.base.AgentResult(*, output, usage={}, model, metadata={}, cost=None)[source]

Bases: BaseModel

Result from agent execution.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:
output: Any
usage: Dict[str, int]
model: str
metadata: Dict[str, Any]
cost: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.base.BaseAgent[source]

Bases: ABC

Abstract base class for all AI agent providers.

abstract async execute(prompt, model, **parameters)[source]

Execute the agent with given prompt and parameters.

Parameters:
Return type:

AgentResult

abstract estimate_cost(usage, model)[source]

Estimate cost based on token usage.

Parameters:
Return type:

float

class agents.base.AgentConfig(provider, api_key=None, base_url=None, timeout=60, max_retries=3)[source]

Bases: object

Configuration for agent providers.

Parameters:

  • provider (str)

  • api_key (str | None)

  • base_url (str | None)

  • timeout (int)

  • max_retries (int)

provider: str
api_key: str | None = None
base_url: str | None = None
timeout: int = 60
max_retries: int = 3
__init__(provider, api_key=None, base_url=None, timeout=60, max_retries=3)
Parameters:

  • provider (str)

  • api_key (str | None)

  • base_url (str | None)

  • timeout (int)

  • max_retries (int)

Return type:

None

Agent Factory

class agents.factory.AgentFactory(config)[source]

Bases: object

Factory for creating agent instances based on provider.

Parameters:

config (PulsarConfig)

__init__(config)[source]
Parameters:

config (PulsarConfig)

get_agent(provider, agent_name=None)[source]

Get or create an agent instance for the given provider.

Parameters:

  • provider (str)

  • agent_name (str | None)

Return type:

BaseAgent

list_supported_providers()[source]

List all supported providers.

Return type:

list[str]

async execute_with_agent(provider, prompt, model, **parameters)[source]

Convenience method to execute with a specific provider.

Parameters:
Return type:

Any

Provider Agents

OpenAI Agent

class agents.openai_agent.OpenAIAgent(config)[source]

Bases: BaseAgent

OpenAI agent implementation supporting GPT models.

Parameters:

config (AgentConfig)

__init__(config)[source]
Parameters:

config (AgentConfig)

async execute(prompt, model, **parameters)[source]

Execute OpenAI API call with retry logic.

Parameters:
Return type:

AgentResult

estimate_cost(usage, model)[source]

Estimate cost in USD based on token usage.

Parameters:
Return type:

float

Anthropic Agent

class agents.anthropic_agent.AnthropicAgent(config)[source]

Bases: BaseAgent

Anthropic agent implementation supporting Claude models.

Parameters:

config (AgentConfig)

__init__(config)[source]
Parameters:

config (AgentConfig)

async execute(prompt, model, **parameters)[source]

Execute Anthropic API call with retry logic.

Parameters:
Return type:

AgentResult

estimate_cost(usage, model)[source]

Estimate cost in USD based on token usage.

Parameters:
Return type:

float

Local Agent (Ollama)

class agents.local_agent.LocalAgent(config)[source]

Bases: BaseAgent

Local agent implementation for Ollama and similar local models.

Parameters:

config (AgentConfig)

__init__(config)[source]
Parameters:

config (AgentConfig)

async execute(prompt, model, **parameters)[source]

Execute local model API call with retry logic.

Parameters:
Return type:

AgentResult

estimate_cost(usage, model)[source]

Local models have no cost.

Parameters:
Return type:

float

Agent Configuration

class agents.config.ProviderConfig(*, api_key=None, base_url=None, timeout=60, max_retries=3)[source]

Bases: BaseModel

Configuration for a specific provider.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

  • api_key (str | None)

  • base_url (str | None)

  • timeout (int)

  • max_retries (int)

api_key: str | None
base_url: str | None
timeout: int
max_retries: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.config.PulsarConfig(*, openai=<factory>, anthropic=<factory>, local=<factory>)[source]

Bases: BaseModel

Main configuration for Pulsar agents.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:
openai: ProviderConfig
anthropic: ProviderConfig
local: ProviderConfig
classmethod from_env()[source]

Load configuration from environment variables.

Return type:

PulsarConfig

classmethod from_dict(config_dict)[source]

Load configuration from dictionary.

Parameters:

config_dict (Dict)

Return type:

PulsarConfig

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

CLI Interface

Main CLI

Pulsar Compose CLI - Docker-like interface for AI workflows

CLI Configuration

class cli.config.CLIConfig(*, default_provider='openai', history_dir='~/.pulsar/history', workflow_dir='~/.pulsar/workflows', log_level='INFO', enable_progress=True, max_history=100, plugins=<factory>)[source]

Bases: BaseModel

Configuration for Pulsar CLI.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:
default_provider: str
history_dir: str
workflow_dir: str
log_level: str
enable_progress: bool
max_history: int
plugins: Dict[str, Any]
classmethod load()[source]

Load configuration from file or create default.

Return type:

CLIConfig

save()[source]

Save configuration to file.

Return type:

None

property history_path: Path

Get expanded history directory path.

property workflow_path: Path

Get expanded workflow directory path.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

History Management

class cli.history.ExecutionHistory[source]

Bases: object

Manages execution history persistence.

__init__()[source]
save_execution(workflow_name, result)[source]

Save execution result and return run ID.

Parameters:
Return type:

str

get_execution(run_id)[source]

Get execution data by run ID.

Parameters:

run_id (str)

Return type:

Dict[str, Any] | None

list_executions(limit=10)[source]

List recent executions.

Parameters:

limit (int)

Return type:

List[Dict[str, Any]]

get_execution_result(run_id)[source]

Get ExecutionResult object for run ID.

Parameters:

run_id (str)

Return type:

ExecutionResult | None

Progress Tracking

class cli.progress.ProgressDisplay[source]

Bases: object

Handles progress visualization for workflow execution.

__init__()[source]
start_execution(workflow_name, total_steps)[source]

Start progress display for workflow execution.

Parameters:

  • workflow_name (str)

  • total_steps (int)

update_step(step_name, success)[source]

Update progress for completed step.

Parameters:
finish_execution(result)[source]

Finish progress display and show summary.

Parameters:

result (ExecutionResult)

show_error(message)[source]

Display error message.

Parameters:

message (str)

show_success(message)[source]

Display success message.

Parameters:

message (str)

show_warning(message)[source]

Display warning message.

Parameters:

message (str)

show_execution_result(result)[source]

Display detailed execution results.

Parameters:

result (ExecutionResult)

Plugin System

class cli.plugins.PluginManager[source]

Bases: object

Manages CLI plugins.

__init__()[source]
load_plugins()[source]

Load plugins from configuration.

get_commands()[source]

Get all plugin commands.

Return type:

List[Command]

Input Providers

Base Provider

Console Provider

File Provider

Web Provider

Test Provider

Step Handlers

Base Handler

Agent Handler

Condition Handler

Interaction Handler

Models

class models.state.StateManager(initial_state=None)[source]

Bases: object

State manager for Pulsar workflow execution with advanced features.

Parameters:

initial_state (Optional[Dict[str, Any]])

__init__(initial_state=None)[source]
Parameters:

initial_state (Dict[str, Any] | None)

async set(key, value)[source]

Set a value in the state using dot notation for nested access.

Parameters:
Return type:

None

async get(key, default=None)[source]

Get a value from the state using dot notation for nested access.

Parameters:

  • key (str)

  • default (Any | None)

Return type:

Any

async render_template(template)[source]

Render a template string with state variables.

Parameters:

template (str)

Return type:

str

async update_from_agent_output(step_name, output)[source]

Update state with agent output and record in execution history.

Parameters:

  • step_name (str)

  • output (Any)

Return type:

None

async get_execution_history()[source]

Get the execution history of all steps.

Return type:

List[Dict[str, Any]]

async get_state_snapshot()[source]

Get a deep copy of the current state.

Return type:

Dict[str, Any]

class models.template.TemplateRenderer[source]

Bases: object

Template rendering system using Jinja2 for {{variables}}.

render(template_str, context)[source]

Render template string with given context variables.

Parameters:
Return type:

str

render_with_fallback(template_str, context, fallback='')[source]

Render template if present, otherwise return fallback.

Parameters:
Return type:

str

Exceptions

class engine.executor.PulsarEngine(workflow, config=None)[source]

Bases: object

Main workflow execution engine for Pulsar.

Parameters:
__init__(workflow, config=None)[source]
Parameters:
async execute(user_input='')[source]

Execute the workflow with given input.

Parameters:

user_input (str)

Return type:

ExecutionResult

async execute_with_initial_state(initial_state)[source]

Execute the workflow with given initial state.

Parameters:

initial_state (Dict[str, Any])

Return type:

ExecutionResult

async execute_step(step)[source]

Execute a single step using appropriate handler.

Return type:

StepResult

get_current_state()[source]

Get current execution state (synchronous snapshot).

Return type:

Dict[str, Any]

async get_current_state_async()[source]

Get current execution state asynchronously.

Return type:

Dict[str, Any]

class agents.base.AgentResult(*, output, usage={}, model, metadata={}, cost=None)[source]

Bases: BaseModel

Result from agent execution.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:
output: Any
usage: Dict[str, int]
model: str
metadata: Dict[str, Any]
cost: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.base.AgentConfig(provider, api_key=None, base_url=None, timeout=60, max_retries=3)[source]

Bases: object

Configuration for agent providers.

Parameters:

  • provider (str)

  • api_key (str | None)

  • base_url (str | None)

  • timeout (int)

  • max_retries (int)

provider: str
api_key: str | None = None
base_url: str | None = None
timeout: int = 60
max_retries: int = 3
__init__(provider, api_key=None, base_url=None, timeout=60, max_retries=3)
Parameters:

  • provider (str)

  • api_key (str | None)

  • base_url (str | None)

  • timeout (int)

  • max_retries (int)

Return type:

None

Pulsar Compose CLI - Docker-like interface for AI workflows

Utility Functions

Expression Evaluation

Advanced expression evaluator for Pulsar conditional execution.

Supports template variables, string operations, membership tests, and type checking.

exception engine.expression_evaluator.ExpressionError[source]

Bases: Exception

Custom exception for expression evaluation errors.

class engine.expression_evaluator.TokenType(value)[source]

Bases: Enum

Token types for expression parsing.

VARIABLE = 'variable'
STRING = 'string'
NUMBER = 'number'
BOOLEAN = 'boolean'
OPERATOR = 'operator'
FUNCTION = 'function'
PARENTHESIS = 'parenthesis'
EOF = 'eof'
class engine.expression_evaluator.Token(type, value, position)[source]

Bases: object

Token representation.

Parameters:
type: TokenType
value: Any
position: int
__init__(type, value, position)
Parameters:
Return type:

None

class engine.expression_evaluator.ExpressionParser(expression)[source]

Bases: object

Parser for expression language.

Parameters:

expression (str)

OPERATORS = {'!=': (3, <function ExpressionParser.<lambda>>), '%': (6, <function ExpressionParser.<lambda>>), '&&': (2, <function ExpressionParser.<lambda>>), '*': (6, <function ExpressionParser.<lambda>>), '+': (5, <function ExpressionParser.<lambda>>), '-': (5, <function ExpressionParser.<lambda>>), '/': (6, <function ExpressionParser.<lambda>>), '<': (4, <function ExpressionParser.<lambda>>), '<=': (4, <function ExpressionParser.<lambda>>), '==': (3, <function ExpressionParser.<lambda>>), '>': (4, <function ExpressionParser.<lambda>>), '>=': (4, <function ExpressionParser.<lambda>>), '||': (1, <function ExpressionParser.<lambda>>)}
FUNCTIONS: Dict[str, Callable[[...], Any]] = {'contains': <function ExpressionParser.<lambda>>, 'endsWith': <function ExpressionParser.<lambda>>, 'isArray': <function ExpressionParser.<lambda>>, 'isNumber': <function ExpressionParser.<lambda>>, 'isObject': <function ExpressionParser.<lambda>>, 'isString': <function ExpressionParser.<lambda>>, 'length': <function ExpressionParser.<lambda>>, 'lower': <function ExpressionParser.<lambda>>, 'startsWith': <function ExpressionParser.<lambda>>, 'trim': <function ExpressionParser.<lambda>>, 'upper': <function ExpressionParser.<lambda>>}
__init__(expression)[source]
Parameters:

expression (str)

parse()[source]

Parse the expression and return the AST root.

Return type:

ExpressionNode

class engine.expression_evaluator.ExpressionNode[source]

Bases: object

Base class for expression AST nodes.

class engine.expression_evaluator.LiteralNode(value)[source]

Bases: ExpressionNode

Literal value node.

Parameters:

value (Any)

value: Any
__init__(value)
Parameters:

value (Any)

Return type:

None

class engine.expression_evaluator.VariableNode(name)[source]

Bases: ExpressionNode

Variable reference node.

Parameters:

name (str)

name: str
__init__(name)
Parameters:

name (str)

Return type:

None

class engine.expression_evaluator.BinaryOpNode(left, operator, right)[source]

Bases: ExpressionNode

Binary operation node.

Parameters:
left: ExpressionNode
operator: str
right: ExpressionNode
__init__(left, operator, right)
Parameters:
Return type:

None

class engine.expression_evaluator.UnaryOpNode(operator, operand)[source]

Bases: ExpressionNode

Unary operation node.

Parameters:
operator: str
operand: ExpressionNode
__init__(operator, operand)
Parameters:
Return type:

None

class engine.expression_evaluator.FunctionCallNode(function_name, arguments)[source]

Bases: ExpressionNode

Function call node.

Parameters:
function_name: str
arguments: List[ExpressionNode]
__init__(function_name, arguments)
Parameters:
Return type:

None

class engine.expression_evaluator.ArrayLiteralNode(elements)[source]

Bases: ExpressionNode

Array literal node.

Parameters:

elements (List[ExpressionNode])

elements: List[ExpressionNode]
__init__(elements)
Parameters:

elements (List[ExpressionNode])

Return type:

None

class engine.expression_evaluator.ArrayAccessNode(array, index)[source]

Bases: ExpressionNode

Array access node (e.g., arr[0]).

Parameters:
array: ExpressionNode
index: ExpressionNode
__init__(array, index)
Parameters:
Return type:

None

engine.expression_evaluator.evaluate_expression(expression, state)[source]

Evaluate an expression with the given state.

Parameters:

  • expression (str) – Expression string to evaluate

  • state (Dict[str, Any]) – State dictionary for variable resolution

Returns:

Result of the expression evaluation

Raises:

ExpressionError – If expression is invalid or evaluation fails

Return type:

Any

engine.expression_evaluator.validate_expression(expression)[source]

Validate that an expression can be parsed without errors.

Parameters:

expression (str) – Expression string to validate

Returns:

True if expression is valid, False otherwise

Return type:

bool

Configuration Helpers

Plugin Loading

Type Definitions

from typing import Dict, List, Optional, Union, Any
from pathlib import Path

# Workflow types
WorkflowConfig = Dict[str, Any]
StepConfig = Dict[str, Any]
AgentConfig = Dict[str, Any]

# Execution types
ExecutionResult = Dict[str, Any]
StepResult = Dict[str, Any]

# Provider types
InputData = Union[str, Dict[str, Any], List[Any]]

# Template types
TemplateVars = Dict[str, Any]
RenderedTemplate = str

Constants

# Version information
__version__ = "0.1.0"

# Supported providers
SUPPORTED_PROVIDERS = ["openai", "anthropic", "ollama"]

# Default timeouts (seconds)
DEFAULT_AGENT_TIMEOUT = 30
DEFAULT_STEP_TIMEOUT = 60

# Retry configuration
DEFAULT_MAX_RETRIES = 3
DEFAULT_BACKOFF_FACTOR = 2.0

# Model defaults
DEFAULT_OPENAI_MODEL = "gpt-3.5-turbo"
DEFAULT_ANTHROPIC_MODEL = "claude-3-haiku-20240307"
DEFAULT_OLLAMA_MODEL = "llama2"

# File extensions
WORKFLOW_EXTENSIONS = [".yml", ".yaml"]
CONFIG_EXTENSIONS = [".toml", ".json", ".yaml", ".yml"]

Environment Variables

Core Configuration

PULSAR_CONFIG_FILE

Path to configuration file. Default: ~/.pulsar/config.toml

PULSAR_WORKFLOW_DIR

Directory containing workflow files. Default: ./workflows

PULSAR_LOG_LEVEL

Logging level (DEBUG, INFO, WARNING, ERROR). Default: INFO

PULSAR_LOG_FILE

Path to log file. Default: ~/.pulsar/pulsar.log

Provider Configuration

OpenAI

OPENAI_API_KEY

OpenAI API key (required for OpenAI agents)

OPENAI_BASE_URL

OpenAI API base URL. Default: https://api.openai.com/v1

OPENAI_TIMEOUT

OpenAI API timeout in seconds. Default: 30

Anthropic

ANTHROPIC_API_KEY

Anthropic API key (required for Anthropic agents)

ANTHROPIC_BASE_URL

Anthropic API base URL. Default: https://api.anthropic.com

ANTHROPIC_TIMEOUT

Anthropic API timeout in seconds. Default: 30

Ollama

OLLAMA_BASE_URL

Ollama API base URL. Default: http://localhost:11434

OLLAMA_TIMEOUT

Ollama API timeout in seconds. Default: 60

Execution Configuration

PULSAR_MAX_CONCURRENT_STEPS

Maximum concurrent step execution. Default: 5

PULSAR_DEFAULT_TIMEOUT

Default step timeout in seconds. Default: 300

PULSAR_MAX_RETRIES

Default maximum retries for failed steps. Default: 3

PULSAR_BACKOFF_FACTOR

Retry backoff factor. Default: 2.0

CLI Configuration

PULSAR_HISTORY_SIZE

Maximum command history size. Default: 1000

PULSAR_PROGRESS_STYLE

Progress bar style (auto, plain, rich). Default: auto

PULSAR_COLORS

Enable/disable colored output. Default: true