Agent Spec Adapters#

This page presents all APIs and classes related to Agent Spec and WayFlow.

Visit the Agent Spec API Documentation to learn more about the native Agent Spec Components.

Agent Spec - API Reference

Tip

Click the button above ↑ to visit the Agent Spec Documentation

class wayflowcore.agentspec.agentspecexporter.AgentSpecExporter(plugins=None)#

Helper class to convert WayFlow objects to Agent Spec configurations.

Parameters:: plugins (List[ComponentSerializationPlugin]) – List of plugins to override existing plugins. By default, uses the latest supported plugins.

to_component(runtime_component)#

Transform the given WayFlow component into the respective PyAgentSpec Component.

Parameters:: runtime_component (Component) – WayFlow Component to serialize to a corresponding PyAgentSpec Component.
Return type:: Component

to_json(runtime_component: Component) → str#

to_json(runtime_component: Component, agentspec_version: AgentSpecVersionEnum | None) → str

to_json(runtime_component: Component, *, disaggregated_components: Sequence[Component | Tuple[Component, str]], export_disaggregated_components: Literal[True]) → Tuple[str, str]

to_json(runtime_component: Component, *, disaggregated_components: Sequence[Component | Tuple[Component, str]] | None, export_disaggregated_components: Literal[False]) → str

to_json(runtime_component: Component, *, disaggregated_components: Sequence[Component | Tuple[Component, str]] | None, export_disaggregated_components: bool) → str | Tuple[str, str]

to_json(runtime_component: Component, agentspec_version: AgentSpecVersionEnum | None, disaggregated_components: Sequence[Component | Tuple[Component, str]] | None, export_disaggregated_components: bool) → str | Tuple[str, str]

Transform the given WayFlow component into the respective Agent Spec JSON representation.

Parameters:

runtime_component – WayFlow component to serialize to an Agent Spec configuration.
agentspec_version – The Agent Spec version of the component.
disaggregated_components –
Configuration specifying the components/fields to disaggregate upon serialization. Each item can be:
- A Component: to disaggregate the component using its id
- A tuple (Component, str): to disaggregate the component using a custom id.
Note

Components in disaggregated_components are disaggregated even if export_disaggregated_components is False.
export_disaggregated_components – Whether to export the disaggregated components or not. Defaults to False.

Returns:

If export_disaggregated_components is True
str – The JSON serialization of the root component.
str – The JSON serialization of the disaggregated components.
If export_disaggregated_components is False
str – The JSON serialization of the root component.

Examples

Basic serialization is done as follows.

>>> from wayflowcore.agent import Agent
>>> from wayflowcore.agentspec import AgentSpecExporter
>>> from wayflowcore.models import VllmModel
>>> from wayflowcore.tools import tool
>>>
>>> llm = VllmModel(
...     model_id="model-id",
...     host_port="VLLM_HOST_PORT",
... )
>>> @tool
... def say_hello_tool() -> str:
...     '''This tool returns "hello"'''
...     return "hello"
...
>>> agent = Agent(
...     name="Simple Agent",
...     llm=llm,
...     tools=[say_hello_tool]
... )
>>> config = AgentSpecExporter().to_json(agent)

To use component disaggregation, specify the component(s) to disaggregate in the disaggregated_components parameter, and ensure that export_disaggregated_components is set to True.

>>> main_config, disag_config = AgentSpecExporter().to_json(
...     agent,
...     disaggregated_components=[llm],
...     export_disaggregated_components=True
... )

Finally, you can specify custom ids for the disaggregated components.

>>> main_config, disag_config = AgentSpecExporter().to_json(
...     agent,
...     disaggregated_components=[(llm, "custom_llm_id")],
...     export_disaggregated_components=True
... )

to_yaml(runtime_component: Component) → str#

to_yaml(runtime_component: Component, agentspec_version: AgentSpecVersionEnum | None) → str

to_yaml(runtime_component: Component, *, disaggregated_components: Sequence[Component | Tuple[Component, str]], export_disaggregated_components: Literal[True]) → Tuple[str, str]

to_yaml(runtime_component: Component, *, disaggregated_components: Sequence[Component | Tuple[Component, str]] | None, export_disaggregated_components: Literal[False]) → str

to_yaml(runtime_component: Component, *, disaggregated_components: Sequence[Component | Tuple[Component, str]] | None, export_disaggregated_components: bool) → str | Tuple[str, str]

to_yaml(runtime_component: Component, agentspec_version: AgentSpecVersionEnum | None, disaggregated_components: Sequence[Component | Tuple[Component, str]] | None, export_disaggregated_components: bool) → str | Tuple[str, str]

Transform the given WayFlow component into the respective Agent Spec YAML representation.

Parameters:

runtime_component – WayFlow component to serialize to an Agent Spec configuration.
agentspec_version – The Agent Spec version of the component.
disaggregated_components –
Configuration specifying the components/fields to disaggregate upon serialization. Each item can be:
- A Component: to disaggregate the component using its id
- A tuple (Component, str): to disaggregate the component using a custom id.
Note

Components in disaggregated_components are disaggregated even if export_disaggregated_components is False.
export_disaggregated_components – Whether to export the disaggregated components or not. Defaults to False.

Returns:

If export_disaggregated_components is True
str – The YAML serialization of the root component.
str – The YAML serialization of the disaggregated components.
If export_disaggregated_components is False
str – The YAML serialization of the root component.

Examples

Basic serialization is done as follows.

>>> from wayflowcore.agent import Agent
>>> from wayflowcore.agentspec import AgentSpecExporter
>>> from wayflowcore.models import VllmModel
>>> from wayflowcore.tools import tool
>>>
>>> llm = VllmModel(
...     model_id="model-id",
...     host_port="VLLM_HOST_PORT",
... )
>>> @tool
... def say_hello_tool() -> str:
...     '''This tool returns "hello"'''
...     return "hello"
...
>>> agent = Agent(
...     name="Simple Agent",
...     llm=llm,
...     tools=[say_hello_tool]
... )
>>> config = AgentSpecExporter().to_yaml(agent)

To use component disaggregation, specify the component(s) to disaggregate in the disaggregated_components parameter, and ensure that export_disaggregated_components is set to True.

>>> main_config, disag_config = AgentSpecExporter().to_yaml(
...     agent,
...     disaggregated_components=[llm],
...     export_disaggregated_components=True
... )

Finally, you can specify custom ids for the disaggregated components.

>>> main_config, disag_config = AgentSpecExporter().to_yaml(
...     agent,
...     disaggregated_components=[(llm, "custom_llm_id")],
...     export_disaggregated_components=True
... )

class wayflowcore.agentspec.runtimeloader.AgentSpecLoader(tool_registry=None, plugins=None)#

Helper class to convert Agent Spec configurations to WayFlow objects.

Parameters:

tool_registry (Dict[str, ServerTool | Callable[[...], Any]] | None) – Optional dictionary to enable converting/loading assistant configurations involving the use of tools. Keys must be the tool names as specified in the serialized configuration, and the values are the ServerTool objects or callables that will be used to create ServerTools.
plugins (List[ComponentSerializationPlugin]) – List of plugins to override existing plugins. By default, uses the latest supported plugins.

load_component(agentspec_component)#

Transform the given PyAgentSpec Component into the respective WayFlow Component

Parameters:: agentspec_component (Component) – PyAgentSpec Component to be converted to a WayFlow Component.
Return type:: Component

load_json(serialized_assistant: str) → Component#

load_json(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None) → Component

load_json(serialized_assistant: str, *, import_only_referenced_components: Literal[False]) → Component

load_json(serialized_assistant: str, *, import_only_referenced_components: Literal[True]) → Dict[str, Component]

load_json(serialized_assistant: str, *, import_only_referenced_components: bool) → Component | Dict[str, Component]

load_json(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None, import_only_referenced_components: Literal[False]) → Component

load_json(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None, import_only_referenced_components: Literal[True]) → Dict[str, Component]

load_json(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None, import_only_referenced_components: bool) → Component | Dict[str, Component]

Transform the given Agent Spec JSON representation into the respective WayFlow Component

Parameters:

serialized_assistant – Serialized Agent Spec configuration to be converted to a WayFlow Component.
components_registry – A dictionary of loaded WayFlow components to use when deserializing the main component.
import_only_referenced_components – When True, loads the referenced/disaggregated components into a dictionary to be used as the components_registry when deserializing the main component. Otherwise, loads the main component. Defaults to False

Returns:

If import_only_referenced_components is False
Component – The deserialized component.
If import_only_referenced_components is False
Dict[str, Component] – A dictionary containing the loaded referenced components.

Examples

Basic deserialization is done as follows. First, serialize a component (here an Agent).

>>> from wayflowcore.agent import Agent
>>> from wayflowcore.agentspec import AgentSpecExporter
>>> from wayflowcore.models import VllmModel
>>> from wayflowcore.tools import tool
>>> llm = VllmModel(
...     model_id="model-id",
...     host_port="VLLM_HOST_PORT",
... )
>>> @tool
... def say_hello_tool() -> str:
...     '''This tool returns "hello"'''
...     return "hello"
...
>>> agent = Agent(
...     name="Simple Agent",
...     llm=llm,
...     tools=[say_hello_tool]
... )
>>> config = AgentSpecExporter().to_json(agent)

Then deserialize using the AgentSpecLoader.

>>> from wayflowcore.agentspec import AgentSpecLoader
>>> TOOL_REGISTRY = {"say_hello_tool": say_hello_tool}
>>> loader = AgentSpecLoader(tool_registry=TOOL_REGISTRY)
>>> deser_agent = loader.load_json(config)

When using disaggregated components, the deserialization must be done in several phases, as follows.

>>> main_config, disag_config = AgentSpecExporter().to_json(
...     agent,
...     disaggregated_components=[(llm, "custom_llm_id")],
...     export_disaggregated_components=True
... )
>>> TOOL_REGISTRY = {"say_hello_tool": say_hello_tool}
>>> loader = AgentSpecLoader(tool_registry=TOOL_REGISTRY)
>>> disag_components = loader.load_json(
...     disag_config, import_only_referenced_components=True
... )
>>> deser_agent = loader.load_json(
...     main_config,
...     components_registry=disag_components
... )

load_yaml(serialized_assistant: str) → Component#

load_yaml(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None) → Component

load_yaml(serialized_assistant: str, *, import_only_referenced_components: Literal[False]) → Component

load_yaml(serialized_assistant: str, *, import_only_referenced_components: Literal[True]) → Dict[str, Component]

load_yaml(serialized_assistant: str, *, import_only_referenced_components: bool) → Component | Dict[str, Component]

load_yaml(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None, import_only_referenced_components: Literal[False]) → Component

load_yaml(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None, import_only_referenced_components: Literal[True]) → Dict[str, Component]

load_yaml(serialized_assistant: str, components_registry: Mapping[str, Component | Tuple[Component, str]] | None, import_only_referenced_components: bool) → Component | Dict[str, Component]

Transform the given Agent Spec YAML representation into the respective WayFlow Component

Parameters:

serialized_assistant – Serialized Agent Spec configuration to be converted to a WayFlow Component.
components_registry – A dictionary of loaded WayFlow components to use when deserializing the main component.
import_only_referenced_components – When True, loads the referenced/disaggregated components into a dictionary to be used as the components_registry when deserializing the main component. Otherwise, loads the main component. Defaults to False

Returns:

If import_only_referenced_components is False
Component – The deserialized component.
If import_only_referenced_components is False
Dict[str, Component] – A dictionary containing the loaded referenced components.

Examples

Basic deserialization is done as follows. First, serialize a component (here an Agent).

>>> from wayflowcore.agent import Agent
>>> from wayflowcore.agentspec import AgentSpecExporter
>>> from wayflowcore.models import VllmModel
>>> from wayflowcore.tools import tool
>>> llm = VllmModel(
...     model_id="model-id",
...     host_port="VLLM_HOST_PORT",
... )
>>> @tool
... def say_hello_tool() -> str:
...     '''This tool returns "hello"'''
...     return "hello"
...
>>> agent = Agent(
...     name="Simple Agent",
...     llm=llm,
...     tools=[say_hello_tool]
... )
>>> config = AgentSpecExporter().to_yaml(agent)

Then deserialize using the AgentSpecLoader.

>>> from wayflowcore.agentspec import AgentSpecLoader
>>> TOOL_REGISTRY = {"say_hello_tool": say_hello_tool}
>>> loader = AgentSpecLoader(tool_registry=TOOL_REGISTRY)
>>> deser_agent = loader.load_yaml(config)

When using disaggregated components, the deserialization must be done in several phases, as follows.

>>> main_config, disag_config = AgentSpecExporter().to_yaml(
...     agent,
...     disaggregated_components=[(llm, "custom_llm_id")],
...     export_disaggregated_components=True
... )
>>> TOOL_REGISTRY = {"say_hello_tool": say_hello_tool}
>>> loader = AgentSpecLoader(tool_registry=TOOL_REGISTRY)
>>> disag_components = loader.load_yaml(
...     disag_config, import_only_referenced_components=True
... )
>>> deser_agent = loader.load_yaml(
...     main_config,
...     components_registry=disag_components
... )

Custom Components#

These are example of custom Agent Spec components that can be used in Agent Spec configurations and loaded/executed in WayFlow.

Note

Both extended and plugin components are introduced to allow assistant developers to export their WayFlow assistants to Agent Spec.

They may be added as native Agent Spec components with modified component name and fields.

Extended Components#

Extended components are Agent Spec components extended with additional fields.

class wayflowcore.agentspec.components.agent.ExtendedAgent(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, llm_config, system_prompt, tools=<factory>, toolboxes=<factory>, context_providers=None, can_finish_conversation=False, max_iterations=10, initial_message='Hi! How can I help you?', caller_input_mode=CallerInputMode.ALWAYS, agents=<factory>, flows=<factory>, agent_template=None)#

Agent that can handle a conversation with a user, interact with external tools and follow interaction flows. Compared to the basic Agent Spec Agent, this ExtendedAgent supports composition with subflows and subagents, custom prompt templates, context providers, and some more customizations on the agent’s 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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
llm_config (LlmConfig) –
system_prompt (str) –
tools (List[Tool]) –
toolboxes (List[PluginToolBox]) –
context_providers (List[PluginContextProvider] | None) –
can_finish_conversation (bool) –
max_iterations (int) –
initial_message (str | None) –
caller_input_mode (CallerInputMode) –
agents (List[Agent]) –
flows (List[Flow]) –
agent_template (PluginPromptTemplate | None) –

agent_template: PluginPromptTemplate | None#: Specific agent template for more advanced prompting techniques. It will be overloaded with the current agent tools, and can have placeholders: * custom_instruction placeholder for the system_prompt parameter

agents: List[Agent]#: Other agents that the agent can call (expert agents).

caller_input_mode: CallerInputMode#: Whether the agent is allowed to ask the user questions (CallerInputMode.ALWAYS) or not (CallerInputMode.NEVER). If set to NEVER, the agent won’t be able to yield.

can_finish_conversation: bool#: Whether the agent can decide to end the conversation or not.

context_providers: List[PluginContextProvider] | None#: Context providers for jinja variables in the system_prompt.

flows: List[Flow]#

initial_message: str | None#: Initial message the agent will post if no previous user message. Default to Agent.DEFAULT_INITIAL_MESSAGE. If None, the LLM will generate it but the agent requires a custom_instruction.

max_iterations: int#: Maximum number of calls to the agent executor before yielding back to the user.

toolboxes: List[PluginToolBox]#: List of toolboxes that the agent can use to fulfil user requests

class wayflowcore.agentspec.components.flow.ExtendedFlow(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, start_node, nodes, control_flow_connections, data_flow_connections=None, context_providers=None, state=<factory>)#

Extension of the basic Agent Spec Flow that supports context providers

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
start_node (Node) –
nodes (List[Node]) –
control_flow_connections (List[ControlFlowEdge]) –
data_flow_connections (List[DataFlowEdge] | None) –
context_providers (List[PluginContextProvider] | None) –
state (List[Property]) –

context_providers: List[PluginContextProvider] | None#: List of providers that add context to specific steps.

state: List[Property]#: The list of properties that compose the state of the Flow

class wayflowcore.agentspec.components.nodes.ExtendedToolNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, tool, input_mapping=<factory>, output_mapping=<factory>, raise_exceptions)#

Extension of the Agent Spec ToolNode. Supports silencing exceptions raised by tools.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
tool (Tool) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
raise_exceptions (bool) –

raise_exceptions: bool#: Whether to raise or not exceptions raised by the tool. If False, it will put the error message as the result of the tool if the tool output type is string.

class wayflowcore.agentspec.components.nodes.ExtendedLlmNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, llm_config, prompt_template, input_mapping=<factory>, output_mapping=<factory>, prompt_template_object=None, send_message=False)#

Extended version of the Agent Spec LlmNode. Supports prompt templates and streaming.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
llm_config (LlmConfig) –
prompt_template (str) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
prompt_template_object (PluginPromptTemplate | None) –
send_message (bool) –

OUTPUT: ClassVar[str] = 'output'#: Output key for the output generated by the LLM, matching the Reference Runtime default value.

classmethod check_either_prompt_str_or_object_is_used(data, handler)#

Wrap validation_func and accumulate errors.

Parameters:

data (Dict[str, Any]) –
handler (Any) –

Return type:

BaseModelSelf

prompt_template_object: PluginPromptTemplate | None#: Prompt template object. Either use prompt_template or prompt_template_object.

send_message: bool#: Determines whether to send the generated content to the current message list or not. By default, the content is only exposed as an output.

class wayflowcore.agentspec.components.nodes.ExtendedMapNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, flow, unpack_input=<factory>, parallel_execution=False)#

Extension of the Agent Spec MapNode. Supports parallel execution and input extraction.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
flow (Flow) –
unpack_input (Dict[str, str]) –
parallel_execution (bool) –

ITERATED_INPUT: ClassVar[str] = 'iterated_input'#: Input key for the iterable to use the MapStep on.

flow: Flow#: Flow that is being executed with each iteration of the input.

parallel_execution: bool#: Executes the mapping operation in parallel. Cannot be set to true if the internal flow can yield. This feature is in beta, be aware that flows might have side effects on one another. Each thread will use a different IO dict, but they will all share the same message list.

unpack_input: Dict[str, str]#: Mapping to specify how to unpack when each iter item is a dict and we need to map its element to the inside flow inputs.

Plugin Components#

Plugin components are new components that are not natively supported in Agent Spec.

Model Context Protocol (MCP)#

class wayflowcore.agentspec.components.mcp.PluginClientTransport(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, session_parameters=<factory>)#

Base class for different MCP client transport mechanisms.

A Transport is responsible for establishing and managing connections to an MCP server, and providing a ClientSession within an async context.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
session_parameters (SessionParameters) –

class wayflowcore.agentspec.components.mcp.PluginSSETransport(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, session_parameters=<factory>, url, headers=None, timeout=5, sse_read_timeout=300)#

Transport implementation that connects to an MCP server via Server-Sent Events.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
session_parameters (SessionParameters) –
url (str) –
headers (Dict[str, str] | None) –
timeout (float) –
sse_read_timeout (float) –

class wayflowcore.agentspec.components.mcp.PluginMCPToolSpec(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None)#

Specification of MCP tool

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –

class wayflowcore.agentspec.components.tools.PluginToolBox(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Class to expose a list of tools to agentic components.

ToolBox is dynamic which means that agentic components equipped with a toolbox can may see its tools to evolve throughout its 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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

class wayflowcore.agentspec.components.mcp.PluginMCPToolBox(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, client_transport, tool_filter=None)#

Class to dynamically expose a list of tools from a MCP Server.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
client_transport (ClientTransport) –
tool_filter (List[PluginMCPToolSpec | str] | None) –

client_transport: ClientTransport#: Transport to use for establishing and managing connections to the MCP server.

tool_filter: List[PluginMCPToolSpec | str] | None#

Optional filter to select specific tools.

If None, exposes all tools from the MCP server.

Specifying a tool name (str) indicates that a tool of the given name is expected from the MCP server.
Specifying a tool signature (Tool) validate the presence and signature of the specified tool in the MCP Server.
- The name of the MCP tool should match the name of the tool from the MCP Server.
- Specifying a non-empty description will override the remote tool description.
- Input descriptors can be provided with description of each input. The names and types should match the remote tool schema.

Agentic patterns#

class wayflowcore.agentspec.components.swarm.PluginSwarm(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, first_agent, relationships=<factory>, handoff=True)#

Defines a Swarm conversational component.

A Swarm is a multi-agent conversational component in which each agent determines the next agent to be executed, based on a list of pre-defined relationships.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
first_agent (Agent) –
relationships (List[List[Agent]]) –
handoff (bool) –

first_agent: Agent#: What is the first Agent to interact with the human user.

handoff: bool#

When False, agent can only talk to each other, the first_agent is fixed for the entire conversation;
When True, agents can handoff the conversation to each other, i.e. transferring the list of messages between an agent and the user to another agent in the Swarm. They can also talk to each other as when handoff=False

relationships: List[List[Agent]]#: Determine the list of allowed interactions in the Swarm. Each element in the list is a tuple (caller_agent, recipient_agent) specifying that the caller_agent can query the recipient_agent.

Messages#

class wayflowcore.agentspec.components.messagelist.PluginMessage(*, role, contents=<factory>, tool_requests=None, tool_result=None, display_only=False, sender=None, recipients=<factory>, time_created=<factory>, time_updated=<factory>)#

Messages are an exchange medium between the user, LLM agent, and controller logic. This helps determining who provided what information.

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:

role (Literal['user', 'assistant', 'system']) –
contents (List[PluginTextContent | PluginImageContent]) –
tool_requests (List[PluginToolRequest] | None) –
tool_result (PluginToolResult | None) –
display_only (bool) –
sender (str | None) –
recipients (List[str]) –
time_created (datetime) –
time_updated (datetime) –

contents: List[PluginTextContent | PluginImageContent]#: Message content. Is a list of chunks with potentially different types

classmethod deserialize_time_created(v)#

Parameters:: v (Any) –
Return type:: Any

classmethod deserialize_time_updated(v)#

Parameters:: v (Any) –
Return type:: Any

display_only: bool#: If True, the message is excluded from any context. Its only purpose is to be displayed in the chat UI (e.g debugging message)

recipients: List[str]#: Recipients of the message in str format.

role: Literal['user', 'assistant', 'system']#: Role of the sender of the message. Can be user, system or assistant

sender: str | None#: Sender of the message in str format.

serialize_time_created(value)#

Parameters:: value (Any) –
Return type:: Any

serialize_time_updated(value)#

Parameters:: value (Any) –
Return type:: Any

time_created: datetime#: Creation timestamp of the message.

time_updated: datetime#: Update timestamp of the message.

tool_requests: List[PluginToolRequest] | None#: A list of ToolRequest objects representing the tools invoked as part of this message. Each request includes the tool’s name, arguments, and a unique identifier.

tool_result: PluginToolResult | None#: A ToolResult object representing the outcome of a tool invocation. It includes the returned content and a reference to the related tool request ID.

class wayflowcore.agentspec.components.messagelist.PluginTextContent(*, type='text', content='')#

Represents the content of a text message.

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:

type (Literal['text']) –
content (str) –

content: str#: The textual content of the message.

type: Literal['text']#

validate_text_content_type()#

Return type:: Self

class wayflowcore.agentspec.components.messagelist.PluginImageContent(*, type='image', base64_content)#

Represents the content of an image message, storing image data as a base64-encoded string.

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:

type (Literal['image']) –
base64_content (str) –

base64_content: str#: A base64-encoded string representing the image data.

type: Literal['image']#

class wayflowcore.agentspec.components.outputparser.PluginRegexPattern(*, pattern, match='first', flags=None)#

Represents a regex pattern and matching options for output parsing.

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:

pattern (str) –
match (Literal['first', 'last']) –
flags (RegexFlag | int | None) –

flags: RegexFlag | int | None#: Potential regex flags to use (re.DOTALL for multiline matching for example)

static from_str(pattern)#

Parameters:: pattern (str | PluginRegexPattern) –
Return type:: PluginRegexPattern

match: Literal['first', 'last']#: Whether to take the first match or the last match

pattern: str#: Regex pattern to match

class wayflowcore.agentspec.components.outputparser.PluginRegexOutputParser(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, regex_pattern, strict=True)#

Parses some text with Regex, potentially several regex to fill a dict

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
regex_pattern (Dict[str, str | PluginRegexPattern] | PluginRegexPattern | str) –
strict (bool) –

regex_pattern: Dict[str, str | PluginRegexPattern] | PluginRegexPattern | str#: Regex pattern to use

strict: bool#: Whether to return empty string if no match is found or return the raw text

class wayflowcore.agentspec.components.outputparser.PluginJsonOutputParser(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, properties=None)#

Parses output as JSON, repairing and serializing as needed.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
properties (Dict[str, str] | None) –

properties: Dict[str, str] | None#: Dictionary of property names and jq queries to manipulate the loaded JSON

class wayflowcore.agentspec.components.outputparser.PluginToolOutputParser(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, tools=None)#

Base parser for tool requests

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
tools (List[Tool] | None) –

tools: List[Tool] | None#

class wayflowcore.agentspec.components.outputparser.PluginJsonToolOutputParser(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, tools=None)#

Parses tool requests from JSON-formatted strings.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
tools (List[Tool] | None) –

class wayflowcore.agentspec.components.outputparser.PluginPythonToolOutputParser(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, tools=None)#

Parses tool requests from Python function call syntax.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
tools (List[Tool] | None) –

class wayflowcore.agentspec.components.outputparser.PluginReactToolOutputParser(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, tools=None)#

Parses ReAct-style tool requests.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
tools (List[Tool] | None) –

class wayflowcore.agentspec.components.template.PluginPromptTemplate(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, messages, output_parser=None, inputs=None, pre_rendering_transforms=None, post_rendering_transforms=None, tools=None, native_tool_calling=True, response_format=None, native_structured_generation=True, generation_config=None)#

Represents a flexible and extensible template for constructing prompts to be sent to large language models (LLMs).

The PromptTemplate class enables the definition of prompt messages with variable placeholders, supports both native and custom tool calling, and allows for structured output generation. It manages input descriptors, message transforms (pre- and post chat_history rendering), and partial formatting for efficiency. The class also integrates with output parsers, tools and llm generation configurations.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
messages (List[PluginMessage]) –
output_parser (List[PluginOutputParser] | PluginOutputParser | None) –
inputs (List[Property] | None) –
pre_rendering_transforms (List[PluginMessageTransform] | None) –
post_rendering_transforms (List[PluginMessageTransform] | None) –
tools (List[Tool] | None) –
native_tool_calling (bool) –
response_format (Property | None) –
native_structured_generation (bool) –
generation_config (LlmGenerationConfig | None) –

CHAT_HISTORY_PLACEHOLDER: ClassVar[PluginMessage] = PluginMessage(role='user', contents=[], tool_requests=None, tool_result=None, display_only=False, sender=None, recipients=[])#: Message placeholder in case the chat history is formatted as a chat.

CHAT_HISTORY_PLACEHOLDER_NAME: ClassVar[str] = '__CHAT_HISTORY__'#: Reserved name of the placeholder for the chat history, if rendered in one message.

RESPONSE_FORMAT_PLACEHOLDER_NAME: ClassVar[str] = '__RESPONSE_FORMAT__'#: Reserved name of the placeholder for the expected output format. Only used if non-native structured generation, to be able to specify the JSON format anywhere in the prompt.

TOOL_PLACEHOLDER_NAME: ClassVar[str] = '__TOOLS__'#: Reserved name of the placeholder for tools.

generation_config: LlmGenerationConfig | None#: Parameters to configure the generation.

inputs: List[Property] | None#: Input descriptors that will be picked up by PromptExecutionStep or AgentExecutionStep. Resolved by default from the variables present in the messages.

messages: List[PluginMessage]#: List of messages for the prompt.

native_structured_generation: bool#: Whether to use native structured generation or not. All llm providers might not support it.

native_tool_calling: bool#: Whether to use the native tool calling of the model or not. All llm providers might not support it.

output_parser: List[PluginOutputParser] | PluginOutputParser | None#: Post-processing applied on the raw output of the LLM.

post_rendering_transforms: List[PluginMessageTransform] | None#: Message transform applied on the rendered list of messages.

pre_rendering_transforms: List[PluginMessageTransform] | None#: Message transform applied before rendering the list of messages into the template.

response_format: Property | None#: Specific format the llm answer should follow.

tools: List[Tool] | None#: Tools to use in the prompt.

class wayflowcore.agentspec.components.transforms.PluginCoalesceSystemMessagesTransform(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Transform that merges consecutive system messages at the start of a message list into a single system message. This is useful for reducing redundancy and ensuring that only one system message appears at the beginning of the conversation.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

class wayflowcore.agentspec.components.transforms.PluginRemoveEmptyNonUserMessageTransform(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Transform that removes messages which are empty and not from the user.

Any message with empty content and no tool requests, except for user messages, will be filtered out from the message list.

This is useful in case the template contains optional messages, which will be discarded if their content is empty (with a string template such as “{% if __PLAN__ %}{{ __PLAN__ }}{% endif %}”).

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

class wayflowcore.agentspec.components.transforms.PluginAppendTrailingSystemMessageToUserMessageTransform(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Transform that appends the content of a trailing system message to the previous user message.

If the last message in the list is a system message and the one before it is a user message, this transform merges the system message content into the user message, reducing message clutter.

This is useful if the underlying LLM does not support system messages at the end.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

class wayflowcore.agentspec.components.transforms.PluginLlamaMergeToolRequestAndCallsTransform(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Llama-specific message transform

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

class wayflowcore.agentspec.components.transforms.PluginReactMergeToolRequestAndCallsTransform(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Simple message processor that joins tool requests and calls into a python-like message

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

Nodes#

class wayflowcore.agentspec.components.nodes.PluginCatchExceptionNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, flow, except_on=None, catch_all_exceptions=False)#

Executes a Flow inside a step and catches specific potential exceptions. If no exception is caught, it will transition to the branches of its subflow. If an exception is caught, it will transition to some specific exception branch has configured in this 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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
flow (Flow) –
except_on (Dict[str, str] | None) –
catch_all_exceptions (bool) –

DEFAULT_EXCEPTION_BRANCH: ClassVar[str] = 'default_exception_branch'#: Name of the branch where the step will transition if catch_all_exceptions is True and an exception was caught.

EXCEPTION_NAME_OUTPUT_NAME: ClassVar[str] = 'exception_name'#: Variable containing the name of the caught exception.

EXCEPTION_PAYLOAD_OUTPUT_NAME: ClassVar[str] = 'exception_payload_name'#: Variable containing the exception payload. Does not contain any higher-level stacktrace information than the wayflowcore stacktraces.

catch_all_exceptions: bool#: Whether to catch any exception and redirect to the default exception branch.

except_on: Dict[str, str] | None#: Names of exceptions to catch and their associated branches.

flow: Flow#: The flow to run and catch exceptions from.

class wayflowcore.agentspec.components.nodes.PluginExtractNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, output_values, llm_config=None, retry=False)#

Node to extract information from a raw json text.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
output_values (Dict[str, str]) –
llm_config (LlmConfig | None) –
retry (bool) –

TEXT: ClassVar[str] = 'text'#: Input key for the raw json text to be parsed.

llm_config: LlmConfig | None#: LLM to use to rephrase the message. Only required if retry=True.

output_values: Dict[str, str]#: The keys are output names of this step. The values are the jq formulas to extract them from the json detected

retry: bool#: Whether to reprompt a LLM to fix the error or not

class wayflowcore.agentspec.components.nodes.PluginInputMessageNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, message=None, input_mapping=<factory>, output_mapping=<factory>, message_template, rephrase=False, llm_config=None)#

Node to get an input from the conversation with the user.

The input step prints a message to the user, asks for an answer and returns it as an output of the step. It places both messages in the messages list so that it is possible to visualize the conversation, but also returns the user input as an output.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
message (str | None) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
message_template (str | None) –
rephrase (bool) –
llm_config (LlmConfig | None) –

USER_PROVIDED_INPUT: ClassVar[str] = 'user_provided_input'#: Output key for the input text provided by the user.

llm_config: LlmConfig | None#: LLM to use to rephrase the message. Only required if rephrase=True.

message_template: str | None#: The message template to use to ask for more information to the user, in jinja format.

rephrase: bool#: Whether to rephrase the message. Requires llm to be set.

class wayflowcore.agentspec.components.nodes.PluginOutputMessageNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, message_template, input_mapping=<factory>, output_mapping=<factory>, message_type=MessageType.AGENT, rephrase=False, llm_config=None, expose_message_as_output=True)#

Node to output a message to the chat history.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
message_template (str) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
message_type (MessageType) –
rephrase (bool) –
llm_config (LlmConfig | None) –
expose_message_as_output (bool) –

OUTPUT: ClassVar[str] = 'output_message'#: Output key for the output message generated by the PluginOutputMessageNode.

expose_message_as_output: bool#: Whether the message generated by this step should appear among the output descriptors

llm_config: LlmConfig | None#: LLM to use to rephrase the message. Only required if rephrase=True.

message: str#: Content of the agent message to append. Allows placeholders, which can define inputs.

message_type: MessageType#: Message type of the message added to the message history.

rephrase: bool#: Whether to rephrase the message. Requires llm to be set.

class wayflowcore.agentspec.components.datastores.nodes.PluginDatastoreCreateNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, datastore, collection_name, ENTITY='entity', CREATED_ENTITY='created_entity')#

Node that can create a new entity in a PluginDatastore.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
datastore (PluginDatastore) –
collection_name (str) –
ENTITY (str) –
CREATED_ENTITY (str) –

CREATED_ENTITY: str#

Output key for the newly created entity.

Type:: str

ENTITY: str#

Input key for the entity to be created.

Type:: str

collection_name: str#: Collection in the datastore manipulated by this step. Can be parametrized using jinja variables, and the resulting input descriptors will be inferred by the step.

datastore: PluginDatastore#: PluginDatastore this step operates on

class wayflowcore.agentspec.components.datastores.nodes.PluginDatastoreDeleteNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, datastore, collection_name, where)#

Step that can delete entities in a PluginDatastore.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
datastore (PluginDatastore) –
collection_name (str) –
where (Dict[str, Any]) –

collection_name: str#: Collection in the datastore manipulated by this node. Can be parametrized using jinja variables, and the resulting input descriptors will be inferred by the node.

datastore: PluginDatastore#: PluginDatastore this node operates on

where: Dict[str, Any]#: Filtering to be applied when deleting entities. The dictionary is composed of property name and value pairs to filter by with exact matches. Only entities matching all conditions in the dictionary will be deleted. For example, {“name”: “Fido”, “breed”: “Golden Retriever”} will match all Golden Retriever dogs named Fido.

class wayflowcore.agentspec.components.datastores.nodes.PluginDatastoreListNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, datastore, collection_name, where, limit, unpack_single_entity_from_list, ENTITIES='entities')#

Step that can list entities in a PluginDatastore.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
datastore (PluginDatastore) –
collection_name (str) –
where (Dict[str, Any] | None) –
limit (int | None) –
unpack_single_entity_from_list (bool | None) –
ENTITIES (str) –

ENTITIES: str#

Output key for the entities listed by this step.

Type:: str

collection_name: str#: Collection in the datastore manipulated by this step. Can be parametrized using jinja variables, and the resulting input descriptors will be inferred by the step.

datastore: PluginDatastore#: PluginDatastore this step operates on

limit: int | None#: Maximum number of entities to list. By default retrieves all entities.

unpack_single_entity_from_list: bool | None#: When limit is set to 1, one may optionally decide to unpack the single entity in the list and only return a the dictionary representing the retrieved entity. This can be useful when, e.g., reading a single entity by its ID.

where: Dict[str, Any] | None#: Filtering to be applied when retrieving entities. The dictionary is composed of property name and value pairs to filter by with exact matches. Only entities matching all conditions in the dictionary will be retrieved. For example, {“name”: “Fido”, “breed”: “Golden Retriever”} will match all Golden Retriever dogs named Fido.

class wayflowcore.agentspec.components.datastores.nodes.PluginDatastoreQueryNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, datastore, query, RESULT='result')#

Step to execute a parameterized SQL query on a relational PluginDatastore (PluginOracleDatabaseDatastore), that supports SQL queries (the specific SQL dialect depends on the database backing the datastore).

This step enables safe, flexible querying of datastores using parameterized SQL. Queries must use bind variables (e.g., :customer_id). String templating within queries is forbidden for security reasons; any such usage raises an error.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
datastore (PluginRelationalDatastore) –
query (str) –
RESULT (str) –

RESULT: str#

Output key for the query result (list of dictionaries, one per row).

Type:: str

datastore: PluginRelationalDatastore#: The PluginDatastore to execute the query against

query: str#: SQL query string using bind variables (e.g., SELECT * FROM table WHERE id = :val). String templating/interpolation is forbidden and will raise an exception.

class wayflowcore.agentspec.components.datastores.nodes.PluginDatastoreUpdateNode(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>, input_mapping=<factory>, output_mapping=<factory>, datastore, collection_name, where, ENTITIES='entities', UPDATE='update')#

Step that can update entities in a PluginDatastore.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –
input_mapping (Dict[str, str]) –
output_mapping (Dict[str, str]) –
datastore (PluginDatastore) –
collection_name (str) –
where (Dict[str, Any]) –
ENTITIES (str) –
UPDATE (str) –

ENTITIES: str#

Output key for the entities listed by this step.

Type:: str

UPDATE: str#

Input key for the dictionary of the updates to be made.

Type:: str

collection_name: str#: Collection in the datastore manipulated by this step. Can be parametrized using jinja variables, and the resulting input descriptors will be inferred by the step.

datastore: PluginDatastore#: PluginDatastore this step operates on

where: Dict[str, Any]#: Filtering to be applied when updating entities. The dictionary is composed of property name and value pairs to filter by with exact matches. Only entities matching all conditions in the dictionary will be updated. For example, {“name”: “Fido”, “breed”: “Golden Retriever”} will match all Golden Retriever dogs with name Fido.

Context Providers#

class wayflowcore.agentspec.components.contextprovider.PluginContextProvider(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, inputs=None, outputs=None, branches=<factory>)#

Context providers are callable components that are used to provide dynamic contextual information to WayFlow assistants. They are useful to connect external datasources to an assistant.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
inputs (List[Property] | None) –
outputs (List[Property] | None) –
branches (List[str]) –

Datastores#

class wayflowcore.agentspec.components.datastores.datastore.PluginDatastore(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Store and perform basic manipulations on collections of entities of various types.

Provides an interface for listing, creating, deleting and updating collections. It also provides a way of describing the entities in this datastore.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

wayflowcore.agentspec.components.datastores.entity.PluginEntity#: alias of Property

class wayflowcore.agentspec.components.datastores.inmemory_datastore.PluginInMemoryDatastore(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, datastore_schema)#

In-memory datastore for testing and development purposes.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
datastore_schema (Dict[str, Property]) –

datastore_schema: Dict[str, Property]#: Mapping of collection names to entity definitions used by this datastore.

class wayflowcore.agentspec.components.datastores.oracle_datastore.PluginMTlsOracleDatabaseConnectionConfig(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, config_dir, dsn, user, password, wallet_location, wallet_password)#

Mutual-TLS Connection Configuration to Oracle Database.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
config_dir (str) –
dsn (str) –
user (str) –
password (str) –
wallet_location (str) –
wallet_password (str) –

config_dir: str#: TNS Admin directory

dsn: str#: Connection string for the database, or entry in the tnsnames.ora file

password: str#: Password for the provided user

user: str#: Connection string for the database

wallet_location: str#: Location where the Oracle Database wallet is stored.

wallet_password: str#: Password for the provided wallet.

class wayflowcore.agentspec.components.datastores.oracle_datastore.PluginOracleDatabaseConnectionConfig(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

Base class used for configuring connections to Oracle Database.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –

class wayflowcore.agentspec.components.datastores.oracle_datastore.PluginOracleDatabaseDatastore(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, datastore_schema, connection_config)#

In-memory datastore for testing and development purposes.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
datastore_schema (Dict[str, Property]) –
connection_config (PluginOracleDatabaseConnectionConfig) –

connection_config: PluginOracleDatabaseConnectionConfig#: Configuration of connection parameters

datastore_schema: Dict[str, Property]#: Mapping of collection names to entity definitions used by this datastore.

class wayflowcore.agentspec.components.datastores.oracle_datastore.PluginTlsOracleDatabaseConnectionConfig(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1, user, password, dsn, config_dir=None)#

TLS Connection Configuration to Oracle Database.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –
user (str) –
password (str) –
dsn (str) –
config_dir (str | None) –

config_dir: str | None#: Configuration directory for the database connection. Set this if you are using an alias from your tnsnames.ora files as a DSN. Make sure that the specified DSN is appropriate for TLS connections (as the tnsnames.ora file in a downloaded wallet will only include DSN entries for mTLS connections)

dsn: str#: Connection string for the database (e.g., created using oracledb.make_dsn)

password: str#: Password for the provided user

user: str#: User used to connect to the database

class wayflowcore.agentspec.components.datastores.relational_datastore.PluginRelationalDatastore(*, id=<factory>, name, description=None, metadata=<factory>, min_agentspec_version=AgentSpecVersionEnum.v25_4_1, max_agentspec_version=AgentSpecVersionEnum.v25_4_1)#

A relational data store that supports querying data using SQL-like queries.

This class extends the PluginDatastore class and adds support for querying data using SQL-like queries.

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:

id (str) –
name (str) –
description (str | None) –
metadata (Dict[str, Any] | None) –
min_agentspec_version (AgentSpecVersionEnum) –
max_agentspec_version (AgentSpecVersionEnum) –