DaprAgentSpecLoader

The main class for loading Open Agent Spec (OAS) specifications and converting them to Dapr Agents configurations.

Class Reference

DaprAgentSpecLoader

DaprAgentSpecLoader(
    tool_registry: ToolRegistry | None = None,
)

Loader for converting OAS specifications to Dapr Agents components.

This class provides methods to load OAS (Open Agent Spec) configurations from JSON or YAML format and convert them to Dapr Agents components that can be executed.

Example

loader = DaprAgentSpecLoader(
    tool_registry={
        "search_tool": search_function,
        "calculator": calc_function,
    }
)

# Load from JSON string
agent = loader.load_json(json_string)

# Load from YAML file
workflow = loader.load_yaml_file("workflow.yaml")

# Create executable Dapr agent
dapr_agent = loader.create_agent(agent)
await dapr_agent.start()

Initialize the loader.

Parameters:

Name	Type	Description	Default
tool_registry	ToolRegistry | None	Dictionary mapping tool names to their callable implementations. Required for tools defined in OAS specifications to be executable.	None

Source code in src/dapr_agents_oas_adapter/loader.py

def __init__(self, tool_registry: ToolRegistry | None = None) -> None:
    """Initialize the loader.

    Args:
        tool_registry: Dictionary mapping tool names to their callable
                      implementations. Required for tools defined in
                      OAS specifications to be executable.
    """
    self._tool_registry = tool_registry or {}
    self._deserializer = AgentSpecDeserializer()
    self._agent_converter = AgentConverter(self._tool_registry)
    self._flow_converter = FlowConverter(self._tool_registry)
    self._logger = get_logger()

Attributes

tool_registry property writable

tool_registry: ToolRegistry

Get the current tool registry.

Functions

load_yaml

load_yaml(
    yaml_content: str,
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from YAML string.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML string containing the OAS specification	required

Returns:

Type	Description
DaprAgentConfig | WorkflowDefinition	DaprAgentConfig for Agent components, WorkflowDefinition for Flows

Raises:

Type	Description
ConversionError	If the YAML cannot be parsed or converted

Source code in src/dapr_agents_oas_adapter/loader.py

def load_yaml(self, yaml_content: str) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from YAML string.

    Args:
        yaml_content: YAML string containing the OAS specification

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows

    Raises:
        ConversionError: If the YAML cannot be parsed or converted
    """
    try:
        component = self._deserializer.from_yaml(yaml_content)
        return self.load_component(component)
    except ConversionError:
        raise
    except Exception as e:
        self._logger.error("load_yaml failed", exc_info=True)
        raise ConversionError(
            "Failed to load YAML",
            suggestion="Ensure the YAML is valid and follows the OAS schema",
        ) from e

load_json

load_json(
    json_content: str,
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from JSON string.

Parameters:

Name	Type	Description	Default
json_content	str	JSON string containing the OAS specification	required

Returns:

Type	Description
DaprAgentConfig | WorkflowDefinition	DaprAgentConfig for Agent components, WorkflowDefinition for Flows

Raises:

Type	Description
ConversionError	If the JSON cannot be parsed or converted

Source code in src/dapr_agents_oas_adapter/loader.py

def load_json(self, json_content: str) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from JSON string.

    Args:
        json_content: JSON string containing the OAS specification

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows

    Raises:
        ConversionError: If the JSON cannot be parsed or converted
    """
    try:
        component = self._deserializer.from_json(json_content)
        return self.load_component(component)
    except ConversionError:
        raise
    except Exception as e:
        self._logger.error("load_json failed", exc_info=True)
        raise ConversionError(
            "Failed to load JSON",
            suggestion="Ensure the JSON is valid and follows the OAS schema",
        ) from e

load_dict

load_dict(
    spec_dict: dict[str, Any],
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from a dictionary.

Parameters:

Name	Type	Description	Default
spec_dict	dict[str, Any]	Dictionary containing the OAS specification	required

Returns:

Type	Description
DaprAgentConfig | WorkflowDefinition	DaprAgentConfig for Agent components, WorkflowDefinition for Flows

Raises:

Type	Description
ConversionError	If the component type is not supported

Source code in src/dapr_agents_oas_adapter/loader.py

def load_dict(self, spec_dict: dict[str, Any]) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from a dictionary.

    Args:
        spec_dict: Dictionary containing the OAS specification

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows

    Raises:
        ConversionError: If the component type is not supported
    """
    component_type = spec_dict.get("component_type", "")
    component_name = spec_dict.get("name") or spec_dict.get("id")

    self._logger.info(
        "load_dict started",
        extra={"component_type": component_type, "component_name": component_name},
    )
    if component_type == "Agent":
        agent_config = self._agent_converter.from_dict(spec_dict)
        self._logger.debug("agent_loaded_from_dict", extra={"agent_name": agent_config.name})
        return agent_config
    if component_type == "Flow":
        workflow_def = self._flow_converter.from_dict(spec_dict)
        self._logger.debug(
            "workflow_loaded_from_dict",
            extra={
                "workflow_name": workflow_def.name,
                "task_count": len(workflow_def.tasks),
            },
        )
        return workflow_def
    raise ConversionError(
        f"Unsupported component type: {component_type}",
        spec_dict,
        suggestion="Set 'component_type' to 'Agent' or 'Flow' in the specification",
    )

create_agent

create_agent(
    config: DaprAgentConfig,
    additional_tools: dict[str, Callable[..., Any]]
    | None = None,
) -> Any

Create an executable Dapr Agent from configuration.

Parameters:

Name	Type	Description	Default
config	DaprAgentConfig	The agent configuration	required
additional_tools	dict[str, Callable[..., Any]] | None	Additional tool implementations to include	None

Returns:

Type	Description
Any	A Dapr Agent instance (AssistantAgent or ReActAgent)

Raises:

Type	Description
ConversionError	If agent creation fails

Source code in src/dapr_agents_oas_adapter/loader.py

def create_agent(
    self,
    config: DaprAgentConfig,
    additional_tools: dict[str, Callable[..., Any]] | None = None,
) -> Any:
    """Create an executable Dapr Agent from configuration.

    Args:
        config: The agent configuration
        additional_tools: Additional tool implementations to include

    Returns:
        A Dapr Agent instance (AssistantAgent or ReActAgent)

    Raises:
        ConversionError: If agent creation fails
    """
    self._logger.info(
        "create_agent started",
        extra={
            "agent_name": config.name,
            "agent_type": config.agent_type,
            "tool_count": len(config.tools),
        },
    )
    tools = {**self._tool_registry}
    if additional_tools:
        tools.update(additional_tools)
        self._logger.debug(
            "additional_tools_merged",
            extra={"additional_tool_count": len(additional_tools)},
        )

    return self._agent_converter.create_dapr_agent(config, tools)

create_workflow

create_workflow(
    workflow_def: WorkflowDefinition,
    task_implementations: dict[str, Callable[..., Any]]
    | None = None,
) -> NamedCallable

Create an executable Dapr workflow from definition.

Parameters:

Name	Type	Description	Default
workflow_def	WorkflowDefinition	The workflow definition	required
task_implementations	dict[str, Callable[..., Any]] | None	Task implementations for workflow activities	None

Returns:

Type	Description
NamedCallable	A workflow function that can be registered with Dapr

Raises:

Type	Description
ConversionError	If workflow creation fails

Source code in src/dapr_agents_oas_adapter/loader.py

def create_workflow(
    self,
    workflow_def: WorkflowDefinition,
    task_implementations: dict[str, Callable[..., Any]] | None = None,
) -> NamedCallable:
    """Create an executable Dapr workflow from definition.

    Args:
        workflow_def: The workflow definition
        task_implementations: Task implementations for workflow activities

    Returns:
        A workflow function that can be registered with Dapr

    Raises:
        ConversionError: If workflow creation fails
    """
    self._logger.info(
        "create_workflow started",
        extra={
            "workflow_name": workflow_def.name,
            "task_count": len(workflow_def.tasks),
            "edge_count": len(workflow_def.edges),
        },
    )
    return self._flow_converter.create_dapr_workflow(workflow_def, task_implementations)

generate_workflow_code

generate_workflow_code(
    workflow_def: WorkflowDefinition,
) -> str

Generate Python code for a Dapr workflow.

Parameters:

Name	Type	Description	Default
workflow_def	WorkflowDefinition	The workflow definition	required

Returns:

Type	Description
str	Python code string that can be saved and executed

Source code in src/dapr_agents_oas_adapter/loader.py

def generate_workflow_code(self, workflow_def: WorkflowDefinition) -> str:
    """Generate Python code for a Dapr workflow.

    Args:
        workflow_def: The workflow definition

    Returns:
        Python code string that can be saved and executed
    """
    return self._flow_converter.generate_workflow_code(workflow_def)

register_tool

register_tool(
    name: str, implementation: Callable[..., Any]
) -> None

Register a tool implementation.

Parameters:

Name	Type	Description	Default
name	str	The tool name as defined in the OAS specification	required
implementation	Callable[..., Any]	The callable implementation	required

Source code in src/dapr_agents_oas_adapter/loader.py

def register_tool(self, name: str, implementation: Callable[..., Any]) -> None:
    """Register a tool implementation.

    Args:
        name: The tool name as defined in the OAS specification
        implementation: The callable implementation
    """
    self._tool_registry[name] = implementation
    self._agent_converter.tool_registry = self._tool_registry
    self._flow_converter.tool_registry = self._tool_registry

Usage Examples

Basic Loading

from dapr_agents_oas_adapter import DaprAgentSpecLoader

loader = DaprAgentSpecLoader()

# Load from YAML
config = loader.load_yaml("""
component_type: Agent
name: assistant
system_prompt: You are helpful.
""")

With Tool Registry

def my_tool(query: str) -> str:
    return f"Result: {query}"

loader = DaprAgentSpecLoader(
    tool_registry={"my_tool": my_tool}
)

Creating Agents

from dapr_agents_oas_adapter.types import DaprAgentConfig

config = loader.load_yaml(yaml_content)
if isinstance(config, DaprAgentConfig):
    agent = loader.create_agent(config)

Creating Workflows

from dapr_agents_oas_adapter.types import WorkflowDefinition

workflow = loader.load_yaml(workflow_yaml)
if isinstance(workflow, WorkflowDefinition):
    workflow_fn = loader.create_workflow(workflow)

Related

• StrictLoader - Loader with validation
• CachedLoader - Loader with caching
• AsyncDaprAgentSpecLoader - Async version