Skip to content

scanner

agent_cover.instrumentation.promptflow.scanner

PromptFlow Integration Module.

This module provides functionality to scan directories for PromptFlow definitions (specifically flow.dag.yaml files) and register associated Jinja2 templates into the AgentRegistry.

It supports dependency injection for file system operations to facilitate testing without accessing the actual file system.

Classes

Functions

scan_promptflow_definitions(root_path=None, registry=None, walker_func=None, file_reader=None)

Scans for PromptFlow YAML files.

This function scans the directory tree for flow.dag.yaml files and processes them to register Jinja2 templates found within the nodes. It is completely isolated from the real file system via walker_func and file_reader injections.

Parameters:

Name Type Description Default
root_path Optional[str]

The root directory to start scanning. Defaults to the current working directory.

 None
registry Optional[AgentRegistry]

The registry instance to register definitions into. Defaults to the global registry.

 None
walker_func Optional[Callable]

A callable that behaves like os.walk. Defaults to _default_walker.

 None
file_reader Optional[Callable]

A callable that reads a file path and returns content. Defaults to _default_file_reader.

 None
Source code in src/agent_cover/instrumentation/promptflow/scanner.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def scan_promptflow_definitions(
    root_path: Optional[str] = None,
    registry: Optional[AgentRegistry] = None,
    walker_func: Optional[Callable] = None,
    file_reader: Optional[Callable] = None,
):
    """Scans for PromptFlow YAML files.

    This function scans the directory tree for `flow.dag.yaml` files and processes
    them to register Jinja2 templates found within the nodes. It is completely
    isolated from the real file system via `walker_func` and `file_reader` injections.

    Args:
        root_path: The root directory to start scanning. Defaults to the current
            working directory.
        registry: The registry instance to register definitions into. Defaults to
            the global registry.
        walker_func: A callable that behaves like os.walk. Defaults to `_default_walker`.
        file_reader: A callable that reads a file path and returns content.
            Defaults to `_default_file_reader`.
    """
    if root_path is None:
        root_path = os.getcwd()
    if registry is None:
        registry = get_registry()
    if walker_func is None:
        walker_func = _default_walker
    if file_reader is None:
        file_reader = _default_file_reader

    logger.debug(f"Scanning for PromptFlow DAGs in {root_path}...")

    dag_files = []
    # Use the injected walker
    for root, dirs, files in walker_func(root_path):
        if "flow.dag.yaml" in files:
            dag_files.append(os.path.join(root, "flow.dag.yaml"))

    count = 0
    for dag_path in dag_files:
        try:
            _process_dag_file(dag_path, registry, file_reader)
            count += 1
        except Exception as e:
            logger.warning(f"Error parsing DAG {dag_path}: {e}", exc_info=True)

    if count > 0:
        logger.debug(f"Processed {count} PromptFlow DAGs.")