Eduard van Valkenburg
6e95517659
* Python: Split type checkers by target (pyright source, 5 checkers on tests/samples) Rework the typing setup along the lines of the 'too many type checkers' approach: - Pyright (strict) is now the sole source-code type checker; mypy is removed from source and its [tool.mypy] block becomes a relaxed profile used only for tests/samples. - Tests are checked by all five checkers (pyright relaxed, mypy, pyrefly, ty, zuban); samples by pyright, pyrefly, and ty. All run in a relaxed/ basic profile so authors aren't forced into over-annotation. - Add pyrightconfig.tests.json and bump sample pyright configs to basic. - Unify test/sample typing onto the same parallel fan-out used by source pyright via run_command_items in task_runner.py. - Make version-conditional imports symmetric: keep or drop the '# type: ignore' on both branches so results match across interpreter versions (local vs CI). - Update SKILL.md, DEV_SETUP.md, and CODING_STANDARD.md for the five gating checkers and pyright on source+tests+samples. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Fix merge regressions from main (typing + runtime) Merging main into the type-checker split branch surfaced regressions that the new five-checker test suite and unit tests caught: Runtime fixes: - anthropic: restore the dropped `cache_read_input_token_count` mapping in _parse_usage_from_anthropic (lost during merge conflict resolution). - gemini: _get_function_calling_mode test helper returned str(enum) ('FunctionCallingConfigMode.AUTO') instead of the enum value ('AUTO'). - openai: _response_id_from_token test helper was an infinite self-recursion; return token['response_id']. - orchestrations: reset output_events per approval iteration so the terminal output assertion counts only the final run. - core: drop a stale duplicate harness test whose message ('non-negative') contradicted the source ('positive'). - purview: import PolicyLocation/PolicyScope/ProtectionScopeActivities/ ExecutionMode used by the processor tests. Type-checker fixes (tests, relaxed profile): - core: pyright/mypy/pyrefly/ty/zuban green-ups across the harness, MCP, observability and types tests. - anthropic/openai: route provider-namespaced UsageDetails keys through a dict cast (extra_items TypedDict unsupported by mypy/ty). - purview: typed model constructors and cache-mock casts. - ag-ui: annotate WorkflowContext[Any, Any] so yield_output accepts test payloads, guard Optional forwarded_props, and ty-ignore intentional bad args. Source pyright (sole source checker) flagged unnecessary ignores newly introduced by merged code in core _tools.py and declarative _declarative_base.py. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Isolate per-package mypy cache in test-typing fan-out The parallel test-typing fan-out runs many mypy processes concurrently, all defaulting to a single shared ./.mypy_cache. Concurrent writes corrupt the cache and mypy aborts with INTERNAL ERROR (intermittently, depending on worker timing) -- which is why CI's Test Typing job failed on a shifting set of packages while a single-package run was fine. Give each mypy invocation an isolated cache dir keyed by its target paths so incremental caching still works per package without races. Other checkers (zuban/pyrefly/ty/pyright) maintain their own caches and are unaffected. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Make lab pyright-only on source (drop source mypy) Lab was the last package still running mypy on its source code, requiring mypy-only `# type: ignore` comments that pyright (the sole source checker everywhere else) flags as unnecessary. Align lab with the rest of the monorepo: - Remove the lab source mypy poe tasks (mypy-gaia/lightning/tau2) and the now-dead strict [tool.mypy] config block. - Drop the 'Run lab mypy' CI step; lab source is type-checked by pyright only. Lab tests remain covered by the workspace test-typing fan-out (mypy, pyrefly, ty, zuban, pyright over tests using the relaxed root config). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Fix test-typing regressions from latest main merge A fresh merge from main brought in new test code never run under the five-checker test-typing suite. Green up across the affected packages: - core: narrow Optional span.attributes with 'and' guards in span filters and assert+cast the json.loads(...attributes[...]) reads (test_observability); match the existing as_agent ignore on the protocol-typed fixture (test_clients). - openai: align new streaming tests with the established chat_options dict pattern (ChatOptions TypedDict isn't assignable to dict), route Optional .annotations[0] access through a small _first_annotation helper (mirrors the file's assert-not-None convention), and annotate a mapped ResponseStream. - foundry_hosting: annotate error: dict[str, Any] = body.get(...) or {} (zuban needs the annotation). - foundry: narrow ignores for the live AIProjectClient credential arg (pyrefly) and connections.get_default (zuban) SDK type gaps. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * updated pyright version * pyright fix * Python: Fix source typing for pyright 1.1.410 Pyright 1.1.410 tightened several checks. Apply the same source fixes as upstream PR #6275: - anthropic: import AsyncAnthropicBedrock from anthropic.lib.bedrock and AsyncAnthropicVertex from anthropic.lib.vertex (no longer re-exported from the anthropic top-level package -> reportPrivateImportUsage). - core _types.py: cast the transform-hook result to UpdateT (reportAssignmentType). - core _workflows/_events.py: annotate the @contextmanager helper as Generator[None] instead of Iterator[None] (reportDeprecated). - redis: build the combined filter expression with an explicit loop instead of reduce(and_, ...), which pyright could no longer fully type (drops the now unused functools.reduce / operator.and_ imports). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Accept plain-text body in Azure Functions workflow/run endpoint The workflow_orchestrator already accepts plain strings as well as JSON objects via context.get_input(), but the start_workflow_orchestration HTTP handler only accepted JSON and returned 400 for any non-JSON body. This made the functions integration tests that POST text/plain to /api/workflow/run (e.g. test_09_workflow_shared_state) fail consistently with 400 != 202. Fall back to the raw request body (decoded as UTF-8) when the body is not JSON, rejecting only a truly empty body. The JSON path is unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Workflows Getting Started Samples
Installation
Microsoft Agent Framework Workflows support ships with the core agent-framework or agent-framework-core package, so no extra installation step is required.
To install with visualization support:
pip install agent-framework[viz] --pre
To export visualization images you also need to install GraphViz.
Samples Overview
Foundational Concepts - Start Here
Begin with the _start-here folder in order. These three samples introduce the core ideas of executors, edges, agents in workflows, and streaming.
| Sample | File | Concepts |
|---|---|---|
| Executors and Edges | _start-here/step1_executors_and_edges.py | Minimal workflow with basic executors and edges |
| Agents in a Workflow | _start-here/step2_agents_in_a_workflow.py | Introduces adding Agents as nodes; calling agents inside a workflow |
| Streaming (Basics) | _start-here/step3_streaming.py | Extends workflows with event streaming |
Once comfortable with these, explore the rest of the samples below.
Samples Overview (by directory)
functional
Write workflows as plain Python async functions — no graph concepts, no executor classes, no edges. Use native control flow (if/else, loops, asyncio.gather) for branching and parallelism.
| Sample | File | Concepts |
|---|---|---|
| Basic Pipeline | functional/basic_pipeline.py | Sequential steps as plain async functions |
| Basic Streaming Pipeline | functional/basic_streaming_pipeline.py | Stream workflow events in real time with run(stream=True) |
| Parallel Pipeline | functional/parallel_pipeline.py | Fan-out/fan-in with asyncio.gather |
| Steps and Checkpointing | functional/steps_and_checkpointing.py | @step decorator for per-step checkpointing and observability |
| Human-in-the-Loop Review | functional/hitl_review.py | HITL with ctx.request_info() and replay |
| Agent Integration | functional/agent_integration.py | Calling agents inside workflow steps |
| Naive Group Chat | functional/naive_group_chat.py | Simple round-robin group chat as a plain loop |
agents
| Sample | File | Concepts |
|---|---|---|
| Azure Chat Agents (Streaming) | agents/azure_chat_agents_streaming.py | Add Azure Chat agents as edges and handle streaming events |
| Azure AI Agents (Streaming) | agents/azure_ai_agents_streaming.py | Add Azure AI agents as edges and handle streaming events |
| Azure AI Agents (Shared Thread) | agents/azure_ai_agents_with_shared_session.py | Share a common message session between multiple Azure AI agents in a workflow |
| Custom Agent Executors | agents/custom_agent_executors.py | Create executors to handle agent run methods |
| Workflow as Agent (Reflection Pattern) | agents/workflow_as_agent_reflection_pattern.py | Wrap a workflow so it can behave like an agent (reflection pattern) |
| Workflow as Agent + HITL | agents/workflow_as_agent_human_in_the_loop.py | Extend workflow-as-agent with human-in-the-loop capability |
| Workflow as Agent with Session | agents/workflow_as_agent_with_session.py | Use AgentSession to maintain conversation history across workflow-as-agent invocations |
| Workflow as Agent kwargs | agents/workflow_as_agent_kwargs.py | Pass custom context (data, user tokens) via kwargs through workflow.as_agent() to @tool tools |
checkpoint
| Sample | File | Concepts |
|---|---|---|
| Checkpoint & Resume | checkpoint/checkpoint_with_resume.py | Create checkpoints, inspect them, and resume execution |
| Checkpoint & HITL Resume | checkpoint/checkpoint_with_human_in_the_loop.py | Combine checkpointing with human approvals and resume pending HITL requests |
| Checkpointed Sub-Workflow | checkpoint/sub_workflow_checkpoint.py | Save and resume a sub-workflow that pauses for human approval |
| Handoff + Tool Approval Resume | orchestrations/handoff_with_tool_approval_checkpoint_resume.py | Handoff workflow that captures tool-call approvals in checkpoints and resumes with human decisions |
| Workflow as Agent Checkpoint | checkpoint/workflow_as_agent_checkpoint.py | Enable checkpointing when using workflow.as_agent() with checkpoint_storage parameter |
| Cosmos DB Checkpoint Storage | checkpoint/cosmos_workflow_checkpointing.py | Use CosmosCheckpointStorage for durable workflow checkpointing backed by Azure Cosmos DB NoSQL |
| Cosmos DB + Foundry Checkpoint | checkpoint/cosmos_workflow_checkpointing_foundry.py | Multi-agent workflow using FoundryChatClient with CosmosCheckpointStorage for durable pause/resume |
composition
| Sample | File | Concepts |
|---|---|---|
| Sub-Workflow (Basics) | composition/sub_workflow_basics.py | Wrap a workflow as an executor and orchestrate sub-workflows |
| Sub-Workflow: Request Interception | composition/sub_workflow_request_interception.py | Intercept and forward sub-workflow requests using @handler for SubWorkflowRequestMessage |
| Sub-Workflow: Parallel Requests | composition/sub_workflow_parallel_requests.py | Multiple specialized interceptors handling different request types from same sub-workflow |
| Sub-Workflow: kwargs Propagation | composition/sub_workflow_kwargs.py | Pass custom context (user tokens, config) from parent workflow through to sub-workflow agents |
control-flow
| Sample | File | Concepts |
|---|---|---|
| Sequential Executors | control-flow/sequential_executors.py | Sequential workflow with explicit executor setup |
| Sequential (Streaming) | control-flow/sequential_streaming.py | Stream events from a simple sequential run |
| Edge Condition | control-flow/edge_condition.py | Conditional routing based on agent classification |
| Switch-Case Edge Group | control-flow/switch_case_edge_group.py | Switch-case branching using classifier outputs |
| Multi-Selection Edge Group | control-flow/multi_selection_edge_group.py | Select one or many targets dynamically (subset fan-out) |
| Simple Loop | control-flow/simple_loop.py | Feedback loop where an agent judges ABOVE/BELOW/MATCHED |
| Workflow Cancellation | control-flow/workflow_cancellation.py | Cancel a running workflow using asyncio tasks |
| Workflow and Intermediate Outputs | control-flow/intermediate_vs_terminal_outputs.py | Select Workflow Output and Intermediate Output executors; hide unselected yields; map Intermediate Output events to text_reasoning content via as_agent |
human-in-the-loop
| Sample | File | Concepts |
|---|---|---|
| Human-In-The-Loop (Guessing Game) | human-in-the-loop/guessing_game_with_human_input.py | Interactive request/response prompts with a human via ctx.request_info() |
| Agents with Approval Requests in Workflows | human-in-the-loop/agents_with_approval_requests.py | Agents that create approval requests during workflow execution and wait for human approval to proceed |
| Agents with Declaration-Only Tools | human-in-the-loop/agents_with_declaration_only_tools.py | Workflow pauses when agent calls a client-side tool (func=None), caller supplies the result |
Builder-oriented request-info samples are maintained in the orchestration sample set (sequential, concurrent, and group-chat builder variants).
tool-approval
Builder-based tool approval samples are maintained in the orchestration sample set.
observability
| Sample | File | Concepts |
|---|---|---|
| Executor I/O Observation | observability/executor_io_observation.py | Observe executor input/output data via executor_invoked events (type='executor_invoked') and executor_completed events (type='executor_completed') without modifying executor code |
For additional observability samples in Agent Framework, see the observability concept samples. The workflow observability sample demonstrates integrating observability into workflows.
orchestration
Orchestration-focused samples (Sequential, Concurrent, Handoff, GroupChat, Magentic), including builder-based
workflow.as_agent(...) variants, are documented in the orchestrations directory.
output selection
Workflow Output selection controls which ctx.yield_output(...) calls are visible to callers as type='output'
events and through WorkflowRunResult.get_outputs(). The core rule is that output_from is an allow-list for
Workflow Output, not a routing rule for every other executor output. Unselected executor payloads are hidden unless
intermediate_output_from explicitly selects them as Intermediate Output.
Use output_from and intermediate_output_from as the canonical API:
| Selection | Workflow Output | Intermediate Output | Hidden payloads |
|---|---|---|---|
| Omit both selections | Every executor yield_output; emits a deprecation warning |
None | None |
output_from="all" |
Every executor yield_output; no warning |
None | None |
output_from=[answerer] |
Only answerer |
None | All other executor payloads |
output_from=[answerer], intermediate_output_from="all_other" |
Only answerer |
Every output-capable executor not selected by output_from |
None |
intermediate_output_from="all_other" |
None | Every output-capable executor | None |
output_from=[], intermediate_output_from="all_other" |
None | Every output-capable executor | None |
output_from=[answerer], intermediate_output_from=[planner, researcher] |
Only answerer |
planner and researcher |
Any other executor payloads |
Invalid selections fail at construction or build time:
| Invalid selection | Why it fails |
|---|---|
output_from="all_other" |
"all_other" is only valid for intermediate_output_from |
intermediate_output_from="all" |
"all" is only valid for output_from |
| The same executor in both selections | One payload cannot be both Workflow Output and Intermediate Output |
| Duplicate executor selections | Duplicates are treated as configuration errors |
| Unknown executor selections | Typos and missing participants are rejected |
output_from=[], intermediate_output_from=[] |
Both explicit selections are empty |
Compatibility aliases such as output_executors emit deprecation warnings where supported. New samples and
applications should use output_from and intermediate_output_from.
When a workflow is wrapped with workflow.as_agent(), Workflow Output becomes normal agent text content. Intermediate
Output becomes text_reasoning content, so AgentResponse.text remains focused on the caller-facing answer while
callers can still inspect progress or supporting work from the response messages.
parallelism
| Sample | File | Concepts |
|---|---|---|
| Concurrent (Fan-out/Fan-in) | parallelism/fan_out_fan_in_edges.py | Dispatch to multiple executors and aggregate results |
| Aggregate Results of Different Types | parallelism/aggregate_results_of_different_types.py | Handle results of different types from multiple concurrent executors |
| Map-Reduce with Visualization | parallelism/map_reduce_and_visualization.py | Fan-out/fan-in pattern with diagram export |
state-management
| Sample | File | Concepts |
|---|---|---|
| State with Agents | state-management/state_with_agents.py | Store in state once and later reuse across agents |
| Workflow Kwargs - Global Context | state-management/workflow_kwargs_global.py | Pass custom context (data, user tokens) via kwargs to @tool tools in all agents |
| Workflow Kwargs - Per Agent | state-management/workflow_kwargs_per_agent.py | Pass custom context (data, user tokens) via kwargs to @tool tools in individual agents |
visualization
| Sample | File | Concepts |
|---|---|---|
| Concurrent with Visualization | visualization/concurrent_with_visualization.py | Fan-out/fan-in workflow with diagram export |
declarative
YAML-based declarative workflows allow you to define multi-agent orchestration patterns without writing Python code. See the declarative workflows README for more details on YAML workflow syntax and available actions.
| Sample | File | Concepts |
|---|---|---|
| Agent to Function Tool | declarative/agent_to_function_tool/ | Chain agent output to InvokeFunctionTool actions |
| Conditional Workflow | declarative/conditional_workflow/ | Nested conditional branching based on user input |
| Customer Support | declarative/customer_support/ | Multi-agent customer support with routing |
| Deep Research | declarative/deep_research/ | Research workflow with planning, searching, and synthesis |
| Function Tools | declarative/function_tools/ | Invoking Python functions from declarative workflows |
| Human-in-Loop | declarative/human_in_loop/ | Interactive workflows that request user input |
| Invoke Function Tool | declarative/invoke_function_tool/ | Call registered Python functions with InvokeFunctionTool |
| Marketing | declarative/marketing/ | Marketing content generation workflow |
| Simple Workflow | declarative/simple_workflow/ | Basic workflow with variable setting, conditionals, and loops |
| Student Teacher | declarative/student_teacher/ | Student-teacher interaction pattern |
resources
- Sample text inputs used by certain workflows:
Notes
- Agent-based samples use provider SDKs (Azure/OpenAI, etc.). Ensure credentials are configured, or adapt agents accordingly.
Sequential orchestration uses a few small adapter nodes for plumbing:
- "input-conversation" normalizes input to
list[Message] - "to-conversation:" converts agent responses into the shared conversation
- "complete" publishes the Workflow Output event (
type='output') These may appear in event streams (executor_invoked/executor_completed). They're analogous to concurrent’s dispatcher and aggregator and can be ignored if you only care about agent activity.
Why FoundryChatClient?
Workflow and orchestration samples use FoundryChatClient because they create agents locally and do not need
server-managed agent resources. This lightweight, project-backed chat client is a good fit for orchestration
patterns such as Sequential, Concurrent, Handoff, GroupChat, and Magentic.
If you need persistent server-side agent resources, use the hosted-agent flows rather than these workflow samples.
Environment Variables
Workflow samples that use FoundryChatClient expect:
FOUNDRY_PROJECT_ENDPOINT(Azure AI Foundry Agent Service (V2) project endpoint)FOUNDRY_MODEL(model deployment name)
These values are passed directly into the client constructor via os.getenv() in sample code.