Skip to content

DaprAgentSpecExporter

Class for exporting Dapr Agents configurations to Open Agent Spec (OAS) format.

Class Reference

DaprAgentSpecExporter 

DaprAgentSpecExporter()

Exporter for converting Dapr Agents components to OAS specifications.

This class provides methods to export Dapr Agents and workflows to Open Agent Spec (OAS) format in JSON or YAML.

Example 
exporter = DaprAgentSpecExporter()

# Export agent configuration to JSON
json_spec = exporter.to_json(agent_config)

# Export workflow to YAML file
exporter.to_yaml_file(workflow_def, "workflow.yaml")

# Export Dapr Agent instance
oas_spec = exporter.from_dapr_agent(dapr_agent)

Initialize the exporter.

Source code in src/dapr_agents_oas_adapter/exporter.py 
46
47
48
49
50
51
52
def __init__(self) -> None:
    """Initialize the exporter."""
    self._serializer = AgentSpecSerializer()
    self._agent_converter = AgentConverter()
    self._flow_converter = FlowConverter()
    self._tool_converter = ToolConverter()
    self._logger = get_logger()

Functions

to_dict 

to_dict(component: Any) -> dict[str, Any]

Export a Dapr component to OAS dictionary format.

Parameters:

Name Type Description Default
component Any

The Dapr component to export (or object)

 required

Returns:

Type Description
dict[str, Any]

Dictionary containing the OAS specification

Raises:

Type Description
ConversionError

If the component cannot be exported
Source code in src/dapr_agents_oas_adapter/exporter.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def to_dict(self, component: Any) -> dict[str, Any]:
    """Export a Dapr component to OAS dictionary format.

    Args:
        component: The Dapr component to export (or object)

    Returns:
        Dictionary containing the OAS specification

    Raises:
        ConversionError: If the component cannot be exported
    """
    component_type = type(component).__name__
    component_name = getattr(component, "name", None)

    self._logger.info(
        "export_to_dict started",
        extra={"component_type": component_type, "component_name": component_name},
    )
    if isinstance(component, DaprAgentConfig):
        result = self._agent_converter.to_dict(component)
        self._logger.debug("agent_exported", extra={"agent_name": component.name})
    elif isinstance(component, WorkflowDefinition):
        result = self._flow_converter.to_dict(component)
        self._logger.debug(
            "workflow_exported",
            extra={
                "workflow_name": component.name,
                "task_count": len(component.tasks),
            },
        )
    else:
        raise ConversionError(
            f"Unsupported component type: {component_type}",
            component,
            suggestion="Only DaprAgentConfig and WorkflowDefinition are supported",
        )

    result["agentspec_version"] = self.AGENTSPEC_VERSION
    return result

to_yaml 

to_yaml(
    component: DaprAgentConfig | WorkflowDefinition,
) -> str

Export a Dapr component to OAS YAML format.

Parameters:

Name Type Description Default
component DaprAgentConfig | WorkflowDefinition

The Dapr component to export

 required

Returns:

Type Description
str

YAML string containing the OAS specification

Raises:

Type Description
ConversionError

If the component cannot be exported
Source code in src/dapr_agents_oas_adapter/exporter.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def to_yaml(self, component: DaprAgentConfig | WorkflowDefinition) -> str:
    """Export a Dapr component to OAS YAML format.

    Args:
        component: The Dapr component to export

    Returns:
        YAML string containing the OAS specification

    Raises:
        ConversionError: If the component cannot be exported
    """
    oas_component = self.to_component(component)
    return self._serializer.to_yaml(oas_component)

to_json 

to_json(
    component: DaprAgentConfig | WorkflowDefinition,
    indent: int = 2,
) -> str

Export a Dapr component to OAS JSON format.

Parameters:

Name Type Description Default
component DaprAgentConfig | WorkflowDefinition

The Dapr component to export

 required
indent int

JSON indentation level

 2

Returns:

Type Description
str

JSON string containing the OAS specification

Raises:

Type Description
ConversionError

If the component cannot be exported
Source code in src/dapr_agents_oas_adapter/exporter.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def to_json(
    self,
    component: DaprAgentConfig | WorkflowDefinition,
    indent: int = 2,
) -> str:
    """Export a Dapr component to OAS JSON format.

    Args:
        component: The Dapr component to export
        indent: JSON indentation level

    Returns:
        JSON string containing the OAS specification

    Raises:
        ConversionError: If the component cannot be exported
    """
    oas_component = self.to_component(component)
    # Use serializer to get base JSON, then re-format with indent
    base_json = self._serializer.to_json(oas_component)
    if indent is not None and indent != 0:
        # Reparse and format with specified indentation
        return json.dumps(json.loads(base_json), indent=indent, ensure_ascii=False)
    return base_json

Usage Examples

Export to Dictionary

from dapr_agents_oas_adapter import DaprAgentSpecExporter
from dapr_agents_oas_adapter.types import DaprAgentConfig

exporter = DaprAgentSpecExporter()

config = DaprAgentConfig(
    name="my_agent",
    role="Assistant",
    goal="Help users"
)

oas_dict = exporter.to_dict(config)
# {"component_type": "Agent", "name": "my_agent", ...}

Export to YAML

yaml_str = exporter.to_yaml(config)
print(yaml_str)

Export to JSON

json_str = exporter.to_json(config)
print(json_str)

Export Workflow

from dapr_agents_oas_adapter.types import (
    WorkflowDefinition,
    WorkflowTaskDefinition,
    WorkflowEdgeDefinition
)

workflow = WorkflowDefinition(
    name="my_workflow",
    tasks=[
        WorkflowTaskDefinition(name="start", task_type="start"),
        WorkflowTaskDefinition(name="end", task_type="end")
    ],
    edges=[
        WorkflowEdgeDefinition(from_node="start", to_node="end")
    ],
    start_node="start",
    end_nodes=["end"]
)

oas = exporter.to_dict(workflow)