Menu
Convert a Prompt Flow PRS pipeline submission to run a Microsoft Agent Framework workflow.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Review code changes and report issues by severity with actionable fixes.
Sharpen a fuzzy intention into one measurable objective string that drives the rest of the work.
Build a Model Context Protocol (MCP) server that lets an LLM call into external tools and resources.
Summarize PDF documents into concise bullet-point digests.
Bump a dependency version across a pnpm workspace and update lockfile.
Convert Prompt Flow flow.dag.yaml definitions into runnable Microsoft Agent Framework workflow code.
Based on SOC occupation classification
| name | maf-prs-job |
| description | Convert a Prompt Flow PRS pipeline submission to run a Microsoft Agent Framework workflow. |
| category | data-pipelines |
| tags | ["ai","api","backend","cli","deployment"] |
| source | null |
| license | MIT |
| author | Team |
| version | 0.1.0-draft |
| needs_review | false |
| slug | maf-prs-job |
| created | 2026-06-12 |
| updated | 2026-06-19 |
| inputs | [{"name":"request","type":"string","required":true,"description":"User request or task description"}] |
| output | {"format":"markdown","description":"Generated content based on the user request"} |
| quality | draft |
Use this skill when converting Prompt Flow PRS pipelines to MAF PRS pipelines.
User request or task description.
Generated content based on the user request.
Follow the guidelines in this skill when converting Prompt Flow PRS pipelines to MAF PRS pipelines.
Convert an existing Prompt Flow Parallel Run Step (PRS) pipeline submission (one that uses
load_component("flow.dag.yaml")) into a PRS pipeline that runs a Microsoft Agent Framework (MAF) workflow as the parallel component.What
load_component(flow.dag.yaml)did automatically — and which pieces this skill produces by hand — is documented in references/pf-vs-maf-prs.md §0.
Activate this skill when the user wants to:
load_component(flow.dag.yaml) with a hand-built parallel component
that wraps a MAF workflow.Also activate on these PF-user phrasings (people who learned from the run-flow-with-pipeline notebook will describe their need in PF terms):
load_component a MAF workflow?" / "load_component('workflow.py')
doesn't work — what's the right way?"flow_component = load_component('flow.dag.yaml')."flow_node in my existing AML pipeline."result_parser_node(pf_output_data=flow_node.outputs.flow_outputs, pf_debug_data=flow_node.outputs.debug_info) — keep that working with MAF."connections={...} / ${data.url} column mapping when the
step is a MAF workflow instead of a flow?"In all of these cases the user's mental model is the PF auto-converted flow
component with predefined data / flow_outputs / debug_info ports.
This skill produces the hand-built MAF equivalent and preserves those names
so downstream pipeline DSL, scheduler, and batch-endpoint code copy-paste
unchanged.
Do not use this skill to convert the flow.dag.yaml itself — that is the
job of promptflow-to-maf. This skill assumes
the MAF workflow already exists (or will be produced by promptflow-to-maf)
and only deals with the PRS / pipeline plumbing around it.
For an input project containing a MAF workflow (workflow.py exporting
create_workflow()) and an existing PF PRS submission script, add the
PRS plumbing into the MAF workflow folder itself (default — keeps
workflow.py and its deployment package together so customers manage one
folder per workflow):
<maf-workflow-folder>/
├── workflow.py ← existing MAF workflow (untouched)
├── requirements.txt ← existing (untouched)
├── src/ ← ADDED: PRS entry + plumbing
│ ├── entry.py ← thin PRS wrapper: init / run(mini_batch, context) / shutdown
│ ├── hooks.py ← THE ONLY USER-EDITED FILE: setup / build_workflow_input / serialize_output
│ └── maf_prs/ ← generic plumbing (mirrors promptflow-parallel's processor/executor split)
│ ├── __init__.py
│ ├── config.py ← argparse → MafPrsConfig
│ ├── executor.py ← per-row driver; calls into hooks.py
│ └── processor.py ← mini-batch dispatch, event-loop reuse, finalize
├── component.yaml ← ADDED: Azure ML parallel component (replaces flow.dag.yaml)
├── env/
│ └── conda.yml ← ADDED: runtime env (agent-framework + AML PRS deps)
├── submit_pipeline.py ← ADDED: MLClient + @pipeline DSL submission driver
└── data/sample.jsonl ← ADDED only when the source `Input(path=...)` is a local file the agent can read; reused verbatim for cloud paths
The original PF flow folder is never modified (it's a separate
folder). Existing files in the MAF folder (workflow.py,
requirements.txt, tests, …) are also left untouched — only new
files are added next to them.
The only file the user normally needs to edit after generation is
src/hooks.py — build_workflow_input(row), serialize_output(output),
and the optional setup(config) — and even those are auto-filled when
the source provides enough information (see
auto-derive-checks.md).
maf_prs/executor.py and the rest of the package are generic and can
be vendored unchanged across all converted workflows.
If the user explicitly asks to keep the MAF folder pristine (e.g. it is a
public doc sample), generate a sibling folder named
<maf-workflow-folder>-prs/ instead, and copy workflow.py into it
so code: ./ in component.yaml ships it to AML. Trade-off: duplicate
workflow.py to keep in sync. Default to the consolidated layout above
unless asked.
workflow.py, must export
create_workflow()). If create_workflow() is missing, route the user
to promptflow-to-maf first.# TODO stub that quotes the original PF source and the missing piece —
never invent endpoint URLs, data paths, or untyped handler inputs.run() on the same instance. The template
executor.execute(...) builds a fresh workflow per row from the cached
_create_workflow factory; do not "optimise" by caching an instance.processor.init() creates the loop;
process() reuses it via run_until_complete; finalize() closes it.
Do not call asyncio.run() per row — it leaks Azure SDK transports.entry.py exposes exactly three top-level
functions: init(), run(mini_batch, context), shutdown().
context.global_row_index_lower_bound is required to stamp a stable
line_number on each result; downstream PF eval tooling joins inputs
to outputs by it.connections= → component inputs + Managed Identity. Surface
endpoint URL / deployment / API version as component inputs:, pass
them via program_arguments. Prefer Managed Identity + Key Vault for
secrets; never hard-code keys in component.yaml.workflow.py
so AML's code: snapshot ships it.A single five-step loop. Each step combines the decision (what to ask / what to print to the user) with the action (what to write).
Use vscode_askQuestions for any of the following that are not obvious
from the workspace:
workflow.py with create_workflow()).Extract the PRS settings from the source script using the table in pf-vs-maf-prs.md §4 (compute, mini_batch_size, retry_settings, etc.) and show the populated table to the user before continuing.
Run the checks in auto-derive-checks.md (A–J) and print the verdict table showing which fields will be auto-filled vs. left as TODO. The same table doubles as the change log handed to the user in step 5.
Add assets/ files into the MAF workflow folder (default) or into a
new sibling <maf-workflow-folder>-prs/ (only if the user opted in).
Do not overwrite any pre-existing file in the MAF folder; if a file name
already exists (e.g. submit_pipeline.py), confirm with the user before
overwriting.
| File(s) | Action |
|---|---|
src/entry.py | Copy verbatim. Do not edit. |
src/maf_prs/{__init__,config,processor,executor}.py | Copy verbatim. Do not edit unless the workflow needs an extra component input (then add a flag in config.parse_args and surface it in component.yaml). |
src/hooks.py | Apply auto-derived bodies for build_workflow_input / serialize_output per the verdict table; insert TODO stubs (template in auto-derive-checks.md) where checks failed. If component inputs need to be turned into env vars / file paths before the workflow imports, fill the setup(config) body too. Add the matching from workflow import ... line at the top. |
component.yaml | Fill inputs: from check F; fill PRS settings from the audit table; set program_arguments to forward inputs + --output_dir ${{outputs.debug_info}}. Use code: ./ and entry_script: src/entry.py so workflow.py (sibling of src/) is shipped to AML. Set data input type: uri_file and always include the PF compatibility flag set in program_arguments (--amlbi_pf_enabled True --amlbi_pf_run_mode component --amlbi_file_format jsonl --amlbi_mini_batch_rows 1) — PRS rejects bare uri_file without these flags (gotcha #12). Wrap every optional: true input in $[[--flag ${{inputs.X}}]] in program_arguments — bare ${{inputs.X}} for an optional input fails registration with Optional input X must be placed in nested argument: $[[]] (gotcha #15). Do not add --pf_input_* flags. |
env/conda.yml | Add any extra pip packages imported by the workflow, always with a lower-bound version pin (package>=X.Y.Z) — bare entries trigger error: resolution-too-deep on the AML image build host and the job never reaches init() (gotcha #14). Keep the existing PRS runtime pins as-is. |
submit_pipeline.py | Fill data_input (check H) by preserving the source Input(path=..., type=..., mode=...) verbatim — same path, same type, same mode. Only rewrite the path to data/sample.jsonl if you also copied it locally per the rule below. Fill MODEL_ENDPOINT / MODEL_DEPLOYMENT (check G), and run-settings assignments. Do not pass ${data.col} arguments. |
data/sample.jsonl | Only copy from the source Input(path=...) when (a) it points at a local file the agent can read, and (b) the user did not explicitly ask to keep the original input. For remote / azureml:// / Input(...) already pointing at a workspace data asset, leave the source Input(...) unchanged in submit_pipeline.py and skip this file (do not invent a sample). Print a one-line note in the verdict table either way. |
workflow.py | Default (consolidated): already present in the target folder — do nothing. Sibling-folder mode only: copy from the source MAF folder. |
If no input-side TODO remains and a local jsonl file is available
(either copied as data/sample.jsonl or already pointed at by the source
Input(path=...)), run the local dry-run from the target folder:
cd <maf-workflow-folder> # or <maf-workflow-folder>-prs in sibling mode
python -c "
import pandas as pd
from types import SimpleNamespace
import sys; sys.path.insert(0, 'src')
from entry import init, run, shutdown
init()
ctx = SimpleNamespace(minibatch_index=0, global_row_index_lower_bound=0)
print(run(pd.read_json('<path-to-local-jsonl>', lines=True), ctx))
shutdown()
"
If the source Input(path=...) is a remote URI (datastore / data asset)
and no local sample exists, skip the dry-run and tell the user
the project will be exercised on the first AML submission instead.
If TODOs remain, skip the dry-run and tell the user which file to edit first. If the dry-run fails, consult references/gotchas.md, fix, retry.
Hand off python submit_pipeline.py with: the verdict table from step 3,
the exact command to run, and a one-line description of what to look for
in the streamed log (one JSONL row per input row in
outputs.flow_outputs/parallel_run_step.jsonl).
maf_prs/executor.py::_setup_tracing already wires it up when
APPLICATIONINSIGHTS_CONNECTION_STRING is set).Do not use this skill for tasks outside its scope.
See the "Workflow" and "Outputs" sections above for practical examples of the conversion process.