Validation

Classes and functions for validating OAS specifications and workflow structures.

OASSchemaValidator

Validates OAS input dictionaries before conversion.

OASSchemaValidator

Validates OAS (Open Agent Spec) dictionaries before conversion.

Performs validation to ensure the input data has the correct structure before attempting to convert it to Dapr formats.

Usage

validator = OASSchemaValidator() result = validator.validate_agent(agent_dict) if not result.is_valid: for error in result.errors: print(error)

Or with strict mode:

validator.validate_agent(agent_dict, raise_on_error=True)

Functions

validate_component

validate_component(
    data: dict[str, Any], *, raise_on_error: bool = False
) -> ValidationResult

Validate any OAS component based on its type.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The component dictionary to validate	required
raise_on_error	bool	If True, raise OASSchemaValidationError on errors	False

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Raises:

Type	Description
OASSchemaValidationError	If raise_on_error=True and validation fails

Source code in src/dapr_agents_oas_adapter/validation.py

def validate_component(
    self,
    data: dict[str, Any],
    *,
    raise_on_error: bool = False,
) -> ValidationResult:
    """Validate any OAS component based on its type.

    Args:
        data: The component dictionary to validate
        raise_on_error: If True, raise OASSchemaValidationError on errors

    Returns:
        ValidationResult with all issues found

    Raises:
        OASSchemaValidationError: If raise_on_error=True and validation fails
    """
    component_type = data.get("component_type", "")

    if component_type == "Agent":
        return self.validate_agent(data, raise_on_error=raise_on_error)
    if component_type == "Flow":
        return self.validate_flow(data, raise_on_error=raise_on_error)
    result = ValidationResult()
    result.add_error(
        f"Unknown or missing component_type: '{component_type}'",
        field="component_type",
        suggestion="Set component_type to 'Agent' or 'Flow'",
    )
    if raise_on_error:
        raise OASSchemaValidationError(result.errors)
    return result

validate_agent

validate_agent(
    data: dict[str, Any], *, raise_on_error: bool = False
) -> ValidationResult

Validate an Agent component dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The agent dictionary to validate	required
raise_on_error	bool	If True, raise OASSchemaValidationError on errors	False

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Raises:

Type	Description
OASSchemaValidationError	If raise_on_error=True and validation fails

Source code in src/dapr_agents_oas_adapter/validation.py

def validate_agent(
    self,
    data: dict[str, Any],
    *,
    raise_on_error: bool = False,
) -> ValidationResult:
    """Validate an Agent component dictionary.

    Args:
        data: The agent dictionary to validate
        raise_on_error: If True, raise OASSchemaValidationError on errors

    Returns:
        ValidationResult with all issues found

    Raises:
        OASSchemaValidationError: If raise_on_error=True and validation fails
    """
    result = ValidationResult()

    # Check required fields
    for field_name in self.AGENT_REQUIRED_FIELDS:
        if field_name not in data or not data[field_name]:
            result.add_error(
                f"Required field '{field_name}' is missing or empty",
                field=field_name,
                suggestion=f"Provide a value for '{field_name}'",
            )

    # Warn about unknown fields
    known_fields = self.AGENT_REQUIRED_FIELDS | self.AGENT_OPTIONAL_FIELDS
    for field_name in data:
        if field_name not in known_fields:
            result.add_warning(
                f"Unknown field '{field_name}' in Agent component",
                field=field_name,
                suggestion="This field may be ignored during conversion",
            )

    # Validate tools if present
    if "tools" in data:
        tools_result = self._validate_tools(data["tools"])
        result.merge(tools_result)

    # Validate llm_config if present
    if "llm_config" in data:
        llm_result = self._validate_llm_config(data["llm_config"])
        result.merge(llm_result)

    if raise_on_error and not result.is_valid:
        raise OASSchemaValidationError(result.errors)

    return result

validate_flow

validate_flow(
    data: dict[str, Any], *, raise_on_error: bool = False
) -> ValidationResult

Validate a Flow component dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The flow dictionary to validate	required
raise_on_error	bool	If True, raise OASSchemaValidationError on errors	False

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Raises:

Type	Description
OASSchemaValidationError	If raise_on_error=True and validation fails

Source code in src/dapr_agents_oas_adapter/validation.py

def validate_flow(
    self,
    data: dict[str, Any],
    *,
    raise_on_error: bool = False,
) -> ValidationResult:
    """Validate a Flow component dictionary.

    Args:
        data: The flow dictionary to validate
        raise_on_error: If True, raise OASSchemaValidationError on errors

    Returns:
        ValidationResult with all issues found

    Raises:
        OASSchemaValidationError: If raise_on_error=True and validation fails
    """
    result = ValidationResult()

    # Check required fields
    for field_name in self.FLOW_REQUIRED_FIELDS:
        if field_name not in data or not data[field_name]:
            result.add_error(
                f"Required field '{field_name}' is missing or empty",
                field=field_name,
                suggestion=f"Provide a value for '{field_name}'",
            )

    # Warn about unknown fields
    known_fields = self.FLOW_REQUIRED_FIELDS | self.FLOW_OPTIONAL_FIELDS
    for field_name in data:
        if field_name not in known_fields:
            result.add_warning(
                f"Unknown field '{field_name}' in Flow component",
                field=field_name,
                suggestion="This field may be ignored during conversion",
            )

    # Validate nodes if present
    if "nodes" in data:
        nodes_result = self._validate_nodes(data["nodes"])
        result.merge(nodes_result)

    # Validate edges if present
    if "edges" in data:
        edges_result = self._validate_oas_edges(data["edges"], data.get("nodes", []))
        result.merge(edges_result)

    if raise_on_error and not result.is_valid:
        raise OASSchemaValidationError(result.errors)

    return result

StrictLoader

Loader wrapper that validates OAS specs before conversion.

StrictLoader

StrictLoader(
    tool_registry: ToolRegistry | None = None,
    *,
    warn_on_unknown_fields: bool = True,
)

A validating wrapper around DaprAgentSpecLoader.

This loader validates OAS specifications against schema rules before attempting conversion, providing early error detection and better error messages.

Example

from dapr_agents_oas_adapter import StrictLoader

# Create a strict loader
loader = StrictLoader()

# This will validate the dict before loading
config = loader.load_dict(spec_dict)

# Invalid specs will raise OASSchemaValidationError
try:
    config = loader.load_dict(invalid_dict)
except OASSchemaValidationError as e:
    for issue in e.issues:
        print(issue)

Initialize the strict loader.

Parameters:

Name	Type	Description	Default
tool_registry	ToolRegistry | None	Dictionary mapping tool names to their callable implementations.	None
warn_on_unknown_fields	bool	If True, unknown fields trigger warnings	True

Source code in src/dapr_agents_oas_adapter/loader.py

def __init__(
    self,
    tool_registry: ToolRegistry | None = None,
    *,
    warn_on_unknown_fields: bool = True,
) -> None:
    """Initialize the strict loader.

    Args:
        tool_registry: Dictionary mapping tool names to their callable
                      implementations.
        warn_on_unknown_fields: If True, unknown fields trigger warnings
    """
    from dapr_agents_oas_adapter.validation import OASSchemaValidator

    self._loader = DaprAgentSpecLoader(tool_registry)
    self._validator = OASSchemaValidator()
    self._warn_on_unknown_fields = warn_on_unknown_fields
    self._logger = get_logger()

Attributes

loader property

loader: DaprAgentSpecLoader

Get the underlying loader.

tool_registry property writable

tool_registry: ToolRegistry

Get the current tool registry.

Functions

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._loader.register_tool(name, implementation)

load_dict

load_dict(
    spec_dict: dict[str, Any], *, validate: bool = True
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from a dictionary with validation.

Parameters:

Name	Type	Description	Default
spec_dict	dict[str, Any]	Dictionary containing the OAS specification	required
validate	bool	Whether to validate before loading (default: True)	True

Returns:

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

Raises:

Type	Description
OASSchemaValidationError	If validation fails
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],
    *,
    validate: bool = True,
) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from a dictionary with validation.

    Args:
        spec_dict: Dictionary containing the OAS specification
        validate: Whether to validate before loading (default: True)

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows

    Raises:
        OASSchemaValidationError: If validation fails
        ConversionError: If the component type is not supported
    """
    if validate:
        self._logger.info(
            "strict_validate_dict started",
            extra={"component_type": spec_dict.get("component_type")},
        )
        self._validator.validate_component(spec_dict, raise_on_error=True)

    return self._loader.load_dict(spec_dict)

validate_dict

validate_dict(
    spec_dict: dict[str, Any],
) -> ValidationResult

Validate a dictionary without loading it.

Parameters:

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

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Source code in src/dapr_agents_oas_adapter/loader.py

def validate_dict(self, spec_dict: dict[str, Any]) -> ValidationResult:
    """Validate a dictionary without loading it.

    Args:
        spec_dict: Dictionary containing the OAS specification

    Returns:
        ValidationResult with all issues found
    """
    return self._validator.validate_component(spec_dict)

load_json

load_json(
    json_content: str, *, validate: bool = True
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from JSON string with optional validation.

Note: Full validation is only available for load_dict. This method parses JSON then delegates to load_dict for validation.

Parameters:

Name	Type	Description	Default
json_content	str	JSON string containing the OAS specification	required
validate	bool	Whether to validate before loading (default: True)	True

Returns:

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

Raises:

Type	Description
OASSchemaValidationError	If validation fails
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,
    *,
    validate: bool = True,
) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from JSON string with optional validation.

    Note: Full validation is only available for load_dict. This method
    parses JSON then delegates to load_dict for validation.

    Args:
        json_content: JSON string containing the OAS specification
        validate: Whether to validate before loading (default: True)

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows

    Raises:
        OASSchemaValidationError: If validation fails
        ConversionError: If the JSON cannot be parsed or converted
    """
    import json

    try:
        spec_dict = json.loads(json_content)
    except json.JSONDecodeError as e:
        raise ConversionError(
            "Invalid JSON",
            suggestion="Ensure the JSON syntax is valid",
        ) from e

    if validate and isinstance(spec_dict, dict):
        return self.load_dict(spec_dict, validate=True)

    return self._loader.load_json(json_content)

load_yaml

load_yaml(
    yaml_content: str, *, validate: bool = True
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from YAML string with optional validation.

Note: Full validation is only available for load_dict. This method parses YAML then delegates to load_dict for validation.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML string containing the OAS specification	required
validate	bool	Whether to validate before loading (default: True)	True

Returns:

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

Raises:

Type	Description
OASSchemaValidationError	If validation fails
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,
    *,
    validate: bool = True,
) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from YAML string with optional validation.

    Note: Full validation is only available for load_dict. This method
    parses YAML then delegates to load_dict for validation.

    Args:
        yaml_content: YAML string containing the OAS specification
        validate: Whether to validate before loading (default: True)

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows

    Raises:
        OASSchemaValidationError: If validation fails
        ConversionError: If the YAML cannot be parsed or converted
    """
    try:
        import yaml

        spec_dict = yaml.safe_load(yaml_content)
    except Exception as e:
        raise ConversionError(
            "Invalid YAML",
            suggestion="Ensure the YAML syntax is valid",
        ) from e

    if validate and isinstance(spec_dict, dict):
        return self.load_dict(spec_dict, validate=True)

    return self._loader.load_yaml(yaml_content)

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
    """
    return self._loader.create_agent(config, additional_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
    """
    return self._loader.create_workflow(workflow_def, task_implementations)

WorkflowValidator

Validates workflow structure after conversion.

WorkflowValidator

Comprehensive validator for WorkflowDefinition objects.

Performs three levels of validation: 1. Structure validation: Required fields, proper types 2. Reference validation: Node references exist, no orphans 3. Edge validation: Connected graph, no cycles in control flow, branching consistency

Usage

validator = WorkflowValidator() result = validator.validate(workflow_def) if not result.is_valid: for error in result.errors: print(error)

Or raise on first error:

result.raise_if_invalid()

Functions

validate

validate(workflow: WorkflowDefinition) -> ValidationResult

Validate a workflow definition comprehensively.

Parameters:

Name	Type	Description	Default
workflow	WorkflowDefinition	The workflow definition to validate	required

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Source code in src/dapr_agents_oas_adapter/validation.py

def validate(self, workflow: WorkflowDefinition) -> ValidationResult:
    """Validate a workflow definition comprehensively.

    Args:
        workflow: The workflow definition to validate

    Returns:
        ValidationResult with all issues found
    """
    result = ValidationResult()

    # Run all validation phases
    result.merge(self._validate_structure(workflow))
    result.merge(self._validate_references(workflow))
    result.merge(self._validate_edges(workflow))

    return result

ValidationResult

Result object returned by validators.

ValidationResult dataclass

ValidationResult(issues: list[ValidationIssue] = list())

Result of a validation operation.

Attributes

is_valid property

is_valid: bool

Check if validation passed (no errors).

errors property

errors: list[ValidationIssue]

Get only error-level issues.

warnings property

warnings: list[ValidationIssue]

Get only warning-level issues.

Functions

add_error

add_error(
    message: str,
    field: str | None = None,
    suggestion: str | None = None,
    cause: Exception | None = None,
) -> None

Add an error-level issue.

Source code in src/dapr_agents_oas_adapter/validation.py

def add_error(
    self,
    message: str,
    field: str | None = None,
    suggestion: str | None = None,
    cause: Exception | None = None,
) -> None:
    """Add an error-level issue."""
    self.issues.append(
        ValidationIssue(
            message=message,
            severity=ValidationSeverity.ERROR,
            field=field,
            suggestion=suggestion,
            cause=cause,
        )
    )

add_warning

add_warning(
    message: str,
    field: str | None = None,
    suggestion: str | None = None,
) -> None

Add a warning-level issue.

Source code in src/dapr_agents_oas_adapter/validation.py

def add_warning(
    self,
    message: str,
    field: str | None = None,
    suggestion: str | None = None,
) -> None:
    """Add a warning-level issue."""
    self.issues.append(
        ValidationIssue(
            message=message,
            severity=ValidationSeverity.WARNING,
            field=field,
            suggestion=suggestion,
        )
    )

merge

merge(other: ValidationResult) -> None

Merge another validation result into this one.

Source code in src/dapr_agents_oas_adapter/validation.py

def merge(self, other: "ValidationResult") -> None:
    """Merge another validation result into this one."""
    self.issues.extend(other.issues)

raise_if_invalid

raise_if_invalid() -> None

Raise WorkflowValidationError if there are any errors.

If the first error has a cause, it is chained via raise ... from.

Source code in src/dapr_agents_oas_adapter/validation.py

def raise_if_invalid(self) -> None:
    """Raise WorkflowValidationError if there are any errors.

    If the first error has a ``cause``, it is chained via ``raise ... from``.
    """
    if not self.is_valid:
        error = WorkflowValidationError(self.errors)
        first_cause = self.errors[0].cause if self.errors else None
        raise error from first_cause

validate_oas_dict

Convenience function for OAS validation.

validate_oas_dict

validate_oas_dict(
    data: dict[str, Any], *, raise_on_error: bool = False
) -> ValidationResult

Convenience function to validate an OAS component dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The OAS component dictionary to validate	required
raise_on_error	bool	If True, raise OASSchemaValidationError on errors	False

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Raises:

Type	Description
OASSchemaValidationError	If raise_on_error=True and validation fails

Source code in src/dapr_agents_oas_adapter/validation.py

def validate_oas_dict(
    data: dict[str, Any],
    *,
    raise_on_error: bool = False,
) -> ValidationResult:
    """Convenience function to validate an OAS component dictionary.

    Args:
        data: The OAS component dictionary to validate
        raise_on_error: If True, raise OASSchemaValidationError on errors

    Returns:
        ValidationResult with all issues found

    Raises:
        OASSchemaValidationError: If raise_on_error=True and validation fails
    """
    validator = OASSchemaValidator()
    return validator.validate_component(data, raise_on_error=raise_on_error)

validate_workflow

Convenience function for workflow validation.

validate_workflow

validate_workflow(
    workflow: WorkflowDefinition,
    *,
    raise_on_error: bool = False,
) -> ValidationResult

Convenience function to validate a workflow.

Parameters:

Name	Type	Description	Default
workflow	WorkflowDefinition	The workflow definition to validate	required
raise_on_error	bool	If True, raise WorkflowValidationError on validation errors	False

Returns:

Type	Description
ValidationResult	ValidationResult with all issues found

Raises:

Type	Description
WorkflowValidationError	If raise_on_error=True and validation fails

Source code in src/dapr_agents_oas_adapter/validation.py

def validate_workflow(
    workflow: WorkflowDefinition, *, raise_on_error: bool = False
) -> ValidationResult:
    """Convenience function to validate a workflow.

    Args:
        workflow: The workflow definition to validate
        raise_on_error: If True, raise WorkflowValidationError on validation errors

    Returns:
        ValidationResult with all issues found

    Raises:
        WorkflowValidationError: If raise_on_error=True and validation fails
    """
    validator = WorkflowValidator()
    result = validator.validate(workflow)
    if raise_on_error:
        result.raise_if_invalid()
    return result

Exceptions

OASSchemaValidationError

OASSchemaValidationError

OASSchemaValidationError(
    issues: list[ValidationIssue] | None = None,
    message: str = "",
    field_path: str | None = None,
)

Bases: ValidationError

Exception raised when OAS schema validation fails.

Initialize with list of validation issues.

Parameters:

Name	Type	Description	Default
issues	list[ValidationIssue] | None	List of validation issues found during validation	None
message	str	Optional error message (used by parent constructor)	''
field_path	str | None	Optional field path (used by parent constructor)	None

Source code in src/dapr_agents_oas_adapter/validation.py

def __init__(
    self,
    issues: list[ValidationIssue] | None = None,
    message: str = "",
    field_path: str | None = None,
) -> None:
    """Initialize with list of validation issues.

    Args:
        issues: List of validation issues found during validation
        message: Optional error message (used by parent constructor)
        field_path: Optional field path (used by parent constructor)
    """
    self.issues: list[ValidationIssue] = issues or []
    if not message and self.issues:
        messages = [str(issue) for issue in self.issues]
        message = (
            f"OAS schema validation failed with {len(self.issues)} error(s):\n"
            + "\n".join(messages)
        )
    super().__init__(message or "OAS schema validation failed", field_path)

Functions

WorkflowValidationError

WorkflowValidationError

WorkflowValidationError(
    issues: list[ValidationIssue] | None = None,
    message: str = "",
    field_path: str | None = None,
)

Bases: ValidationError

Exception raised when workflow validation fails.

Initialize with list of validation issues.

Parameters:

Name	Type	Description	Default
issues	list[ValidationIssue] | None	List of validation issues found during validation	None
message	str	Optional error message (used by parent constructor)	''
field_path	str | None	Optional field path (used by parent constructor)	None

Source code in src/dapr_agents_oas_adapter/validation.py

def __init__(
    self,
    issues: list[ValidationIssue] | None = None,
    message: str = "",
    field_path: str | None = None,
) -> None:
    """Initialize with list of validation issues.

    Args:
        issues: List of validation issues found during validation
        message: Optional error message (used by parent constructor)
        field_path: Optional field path (used by parent constructor)
    """
    self.issues: list[ValidationIssue] = issues or []
    if not message and self.issues:
        messages = [str(issue) for issue in self.issues]
        message = f"Workflow validation failed with {len(self.issues)} error(s):\n" + "\n".join(
            messages
        )
    super().__init__(message or "Workflow validation failed", field_path)

Functions

Usage Examples

OAS Validation

from dapr_agents_oas_adapter import (
    StrictLoader,
    OASSchemaValidator,
    validate_oas_dict
)
from dapr_agents_oas_adapter.validation import OASSchemaValidationError

# Using StrictLoader
loader = StrictLoader()
try:
    config = loader.load_dict(spec)
except OASSchemaValidationError as e:
    print(f"Errors: {e.issues}")

# Using validator directly
validator = OASSchemaValidator()
result = validator.validate(spec)
if not result.is_valid:
    for error in result.errors:
        print(error)

# Using convenience function
result = validate_oas_dict(spec)

Workflow Validation

from dapr_agents_oas_adapter import (
    WorkflowValidator,
    validate_workflow
)
from dapr_agents_oas_adapter.validation import WorkflowValidationError

# Using validator
validator = WorkflowValidator()
result = validator.validate(workflow)

# Using convenience function
result = validate_workflow(workflow)

# Raise on error
try:
    validate_workflow(workflow, raise_on_error=True)
except WorkflowValidationError as e:
    print(f"Validation failed: {e}")