entity.plugins package

Public plugin interfaces and helpers.

class entity.plugins.Plugin(resources, config=None)[source]

Bases: ABC

Base class for all plugins.

Parameters:

  • resources (dict[str, Any])

  • config (Dict[str, Any] | None)

class ConfigModel[source]

Bases: BaseModel

Default empty configuration.

class Config[source]

Bases: object

extra = 'forbid'
model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

__init__(resources, config=None)[source]

Instantiate the plugin and run all startup validations.

Parameters:
dependencies: list[str] = []
async execute(context)[source]

Run the plugin.

Parameters:

context (Any)

Return type:

Any

should_execute(context)[source]

Determine if plugin should run for this context.

Returns:

True if the plugin should execute, False to skip.

Parameters:

context (PluginContext)

Return type:

bool

skip_conditions: List[Callable[['PluginContext'], bool]] = []
supported_stages: list[str] = []
validate_config()[source]

Validate config using ConfigModel and return the model.

Return type:

ValidationResult

validate_workflow(workflow)[source]

Validate that cls can run in stage before workflow execution.

Parameters:

workflow (Workflow)

Return type:

ValidationResult

class entity.plugins.PromptPlugin(resources, config=None)[source]

Bases: Plugin

LLM-driven reasoning or validation plugin.

Parameters:

  • resources (dict[str, Any])

  • config (Dict[str, Any] | None)

dependencies: list[str] = ['llm']
supported_stages: list[str] = ['think', 'review']
class entity.plugins.ToolPlugin(resources, config=None)[source]

Bases: Plugin

Plugin type for executing external actions.

Parameters:

  • resources (dict[str, Any])

  • config (Dict[str, Any] | None)

dependencies: list[str] = []
supported_stages: list[str] = ['do']
class entity.plugins.InputAdapterPlugin(resources, config=None)[source]

Bases: Plugin

Convert external input into workflow messages.

Parameters:

  • resources (dict[str, Any])

  • config (Dict[str, Any] | None)

supported_stages: list[str] = ['input']
class entity.plugins.OutputAdapterPlugin(resources, config=None)[source]

Bases: Plugin

Convert workflow responses into external representations.

Parameters:

  • resources (dict[str, Any])

  • config (Dict[str, Any] | None)

supported_stages: list[str] = ['output']

Subpackages

Submodules