Agents in Workflows

This tutorial demonstrates how to integrate AI agents into workflows using Agent Framework. You'll learn to create workflows that leverage the power of specialized AI agents for content creation, review, and other collaborative tasks.

What You'll Build

You'll create a workflow that:

Uses Azure Foundry Agent Service to create intelligent agents
Implements a French translation agent that translates input to French
Implements a Spanish translation agent that translates French to Spanish
Implements an English translation agent that translates Spanish back to English
Connects agents in a sequential workflow pipeline
Streams real-time updates as agents process requests
Demonstrates proper resource cleanup for Azure Foundry agents

Concepts Covered

Prerequisites

.NET 8.0 SDK or later
An Azure Foundry project endpoint and model configured
Azure CLI installed and authenticated (for Azure credential authentication)
A new console application

Step 1: Install NuGet packages

First, install the required packages for your .NET project:

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

Step 2: Set Up Azure Foundry Client

Configure the Azure Foundry client with environment variables and authentication:

using Azure.AI.Projects;
using Azure.AI.Projects.Agents;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure AI Project client
        var endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
            ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-4o-mini";
        var aiProjectClient = new AIProjectClient(new Uri(endpoint), new AzureCliCredential());

Step 3: Create Agent Factory Method

Implement a helper method to create Azure Foundry agents with specific instructions:

    /// <summary>
    /// Creates a translation agent for the specified target language.
    /// </summary>
    /// <param name="targetLanguage">The target language for translation</param>
    /// <param name="aiProjectClient">The AIProjectClient to create the agent</param>
    /// <param name="model">The model to use for the agent</param>
    /// <returns>A FoundryAgent configured for the specified language</returns>
    private static async Task<FoundryAgent> GetTranslationAgentAsync(
        string targetLanguage,
        AIProjectClient aiProjectClient,
        string model)
    {
        string agentName = $"{targetLanguage}Translator";
        var version = await aiProjectClient.AgentAdministrationClient.CreateAgentVersionAsync(
            agentName,
            new ProjectsAgentVersionCreationOptions(
                new DeclarativeAgentDefinition(model)
                {
                    Instructions = $"You are a translation assistant that translates the provided text to {targetLanguage}."
                }));

        return aiProjectClient.AsAIAgent(version);
    }
}

Step 4: Create Specialized Azure Foundry Agents

Create three translation agents using the helper method:

        // Create agents
        AIAgent frenchAgent = await GetTranslationAgentAsync("French", aiProjectClient, deploymentName);
        AIAgent spanishAgent = await GetTranslationAgentAsync("Spanish", aiProjectClient, deploymentName);
        AIAgent englishAgent = await GetTranslationAgentAsync("English", aiProjectClient, deploymentName);

Step 5: Build the Workflow

Connect the agents in a sequential workflow using the WorkflowBuilder:

        // Build the workflow by adding executors and connecting them
        var workflow = new WorkflowBuilder(frenchAgent)
            .AddEdge(frenchAgent, spanishAgent)
            .AddEdge(spanishAgent, englishAgent)
            .Build();

Step 6: Execute with Streaming

Run the workflow with streaming to observe real-time updates from all agents:

        // Execute the workflow
        await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!"));

        // Must send the turn token to trigger the agents.
        // The agents are wrapped as executors. When they receive messages,
        // they will cache the messages and only start processing when they receive a TurnToken.
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
        {
            if (evt is AgentResponseUpdateEvent executorComplete)
            {
                Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            }
        }

Step 7: Resource Cleanup

Properly clean up the Azure Foundry agents after use:

        // Cleanup the agents created for the sample.
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(frenchAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(spanishAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(englishAgent.Name);
    }

How It Works

Azure Foundry Client Setup: Uses AIProjectClient with Azure CLI credentials for authentication
Agent Creation: Creates versioned agents on Azure Foundry with specific instructions for translation
Sequential Processing: French agent translates input first, then Spanish agent, then English agent
Turn Token Pattern: Agents cache messages and only process when they receive a TurnToken
Streaming Updates: AgentResponseUpdateEvent provides real-time token updates as agents generate responses
Resource Management: Proper cleanup of Azure Foundry agents using the Administration API

Key Concepts

Azure Foundry Agent Service: Cloud-based AI agents with advanced reasoning capabilities
AIProjectClient: Client for creating and managing agents on Azure Foundry
WorkflowEvent: Output events (type="output") contain agent output data (AgentResponseUpdate for streaming, AgentResponse for non-streaming)
TurnToken: Signal that triggers agent processing after message caching
Sequential Workflow: Agents connected in a pipeline where output flows from one to the next

Complete Implementation

For the complete working implementation of this Azure Foundry agents workflow, see the FoundryAgent Program.cs sample in the Agent Framework repository.

What You'll Build

You'll create a workflow that:

Uses FoundryChatClient to create intelligent agents
Implements a Writer agent that creates content based on prompts
Implements a Reviewer agent that provides feedback on the content
Connects agents in a sequential workflow pipeline
Streams real-time updates as agents process requests

Concepts Covered

Prerequisites

Python 3.10 or later
Agent Framework installed: pip install agent-framework
Azure OpenAI Responses configured with proper environment variables
Azure CLI authentication: az login

Step 1: Import Required Dependencies

Start by importing the necessary components for workflows and Azure OpenAI Responses agents:

import asyncio
import os

from agent_framework import AgentResponseUpdate, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

Step 2: Create Azure OpenAI Responses Client

Create one shared client that you can use to construct multiple agents:

async def main() -> None:
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        credential=AzureCliCredential(),
    )

Step 3: Create Specialized Agents

Create two specialized agents for content creation and review:

    # Create a Writer agent that generates content
    writer_agent = client.as_agent(
        name="Writer",
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
    )

    # Create a Reviewer agent that provides feedback
    reviewer_agent = client.as_agent(
        name="Reviewer",
        instructions=(
            "You are an excellent content reviewer. "
            "Provide actionable feedback to the writer about the provided content. "
            "Provide the feedback in the most concise manner possible."
        ),
    )

Step 4: Build the Workflow

Connect the agents in a sequential workflow using the builder:

        # Build the workflow with agents as executors
        workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

Step 5: Execute with Streaming

Run the workflow with streaming to observe real-time updates from both agents:

    last_author: str | None = None

    events = workflow.run("Create a slogan for a new electric SUV that is affordable and fun to drive.", stream=True)
    async for event in events:
        if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
            update = event.data
            author = update.author_name
            if author != last_author:
                if last_author is not None:
                    print()
                print(f"{author}: {update.text}", end="", flush=True)
                last_author = author
            else:
                print(update.text, end="", flush=True)

Step 6: Complete Main Function

Wrap everything in the main function with proper async execution:

if __name__ == "__main__":
    asyncio.run(main())

How It Works

Client Setup: Uses one FoundryChatClient with Azure CLI credentials for authentication.
Agent Creation: Creates Writer and Reviewer agents from the same client configuration.
Sequential Processing: Writer agent generates content first, then passes it to the Reviewer agent.
Streaming Updates: Output events (type="output") with AgentResponseUpdate data provide real-time token updates as agents generate responses.

Key Concepts

FoundryChatClient: Shared client used to create workflow agents with consistent configuration.
WorkflowEvent: Output events (type="output") contain agent output data (AgentResponseUpdate for streaming, AgentResponse for non-streaming).
Sequential Workflow: Agents connected in a pipeline where output flows from one to the next.

Complete Implementation

For the complete working implementation, see azure_ai_agents_streaming.py in the Agent Framework repository.

What You'll Build

You'll create a workflow that:

Uses Microsoft Foundry model-deployment agents as workflow executors
Implements a French translation agent
Implements a Spanish translation agent
Implements an English translation agent
Connects agents in a sequential workflow pipeline
Streams real-time updates as agents process requests

Concepts Covered

Prerequisites

Go 1.26 or later
Microsoft Foundry project endpoint and model deployment configured
Azure CLI authentication or another Azure credential source

In a new directory, initialize a Go module and install Agent Framework:

go mod init example.com/agent-workflow
go get github.com/microsoft/agent-framework-go

Set the project endpoint and the name of your deployed model:

$env:FOUNDRY_PROJECT_ENDPOINT = "https://<resource>.services.ai.azure.com/api/projects/<project>"
$env:FOUNDRY_MODEL = "<deployment-name>"

Step 1: Set Up Foundry Configuration

The following sections build the runExample function in the complete implementation. Start with its imports and configuration:

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
if endpoint == "" {
    return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
}
model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
if model == "" {
    return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
}

token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
    return err
}

Warning

azidentity.NewDefaultAzureCredential is convenient for development but requires careful consideration in production. In production, consider using a specific credential, such as azidentity.NewManagedIdentityCredential, to avoid latency issues, unintended credential probing, and potential security risks from fallback mechanisms.

Step 2: Create Agent Factory Function

Create agents with specific translation instructions:

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: fmt.Sprintf(
            "Translate the user's text to %s. Return only the translation.",
            language,
        ),
        Config: agent.Config{Name: language + "Agent"},
    })
}

Step 3: Create Specialized Foundry Agents

frenchAgent := newTranslationAgent("French")
spanishAgent := newTranslationAgent("Spanish")
englishAgent := newTranslationAgent("English")

Step 4: Build the Workflow

Agents can be used as workflow executors, enabling AI-powered workflow steps.

Bind each agent as a workflow executor, then connect the executors with edges:

cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
french := agentworkflow.New(frenchAgent, cfg)
spanish := agentworkflow.New(spanishAgent, cfg)
english := agentworkflow.New(englishAgent, cfg)

wf, err := workflow.NewBuilder(french).
    AddEdge(french, spanish).
    AddEdge(spanish, english).
    WithIntermediateOutputFrom(french, spanish).
    WithOutputFrom(english).
    Build()
if err != nil {
    return err
}

ForwardIncomingMessages: new(false) passes each agent's response downstream without also forwarding that agent's input. Each hosted executor manages its own agent session.

Register French and Spanish as intermediate output sources so callers can observe all three agents. English remains the terminal output source. Output designation controls what the caller sees, not which messages flow along the edges.

Step 5: Execute with Streaming

Run the workflow and enable update events with a workflow.TurnToken. Handle workflow failure events as well as errors from the event iterator:

run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
if err != nil {
    return err
}
defer func() {
    err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()

emitEvents := true
if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
} else if !accepted {
    return fmt.Errorf("workflow doesn't accept turn tokens")
}

lastExecutorID := ""
for evt, streamErr := range run.WatchStream(ctx) {
    if streamErr != nil {
        return streamErr
    }
    switch event := evt.(type) {
    case workflow.OutputEvent:
        update, ok := event.Output.(*agent.ResponseUpdate)
        if !ok || update.String() == "" {
            continue
        }
        if event.ExecutorID != lastExecutorID {
            if lastExecutorID != "" {
                fmt.Println()
            }
            fmt.Printf("%s: ", event.ExecutorID)
            lastExecutorID = event.ExecutorID
        }
        fmt.Print(update.String())
    case workflow.ErrorEvent:
        return event.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w",
            event.ExecutorID, event.Error)
    }
}
if lastExecutorID != "" {
    fmt.Println()
}

The output contains French, Spanish, and English translations, grouped by executor. The complete example returns from runExample before logging a fatal error, so deferred workflow cleanup still runs.

Unlike the C# example, this Go example doesn't create stored agent versions that need deletion in Foundry.

How It Works

Client Setup: Uses a Foundry project endpoint, model deployment, and Azure credential.
Agent Creation: Creates specialized agents with language-specific instructions.
Agent Hosting: Uses agentworkflow.New to bind each agent as a workflow executor.
Sequential Processing: The French executor runs first, then Spanish, then English.
Turn Token Pattern: Hosted agents buffer messages and run when they receive a workflow.TurnToken.
Streaming Updates: Intermediate and terminal output registrations expose updates from all three agents.
Failure Handling: The host checks ErrorEvent and ExecutorFailedEvent and closes the workflow run before reporting failure.

Key Concepts

Foundry agent: An agent.Agent backed by a Foundry model deployment.
agentworkflow.New: Adapts an agent for use as a workflow executor.
workflow.TurnToken: Signal that triggers hosted agents to process buffered messages.
Workflow OutputEvent: Carries agent response updates and final workflow outputs.
Sequential Workflow: Agents connected in a pipeline where output flows from one to the next.

Complete Implementation

Save this example as main.go in your Go module:

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func main() {
    if err := runExample(context.Background()); err != nil {
        log.Fatal(err)
    }
}

func runExample(ctx context.Context) (err error) {
    endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
    if endpoint == "" {
        return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
    }
    model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
    if model == "" {
        return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
    }

    token, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        return err
    }

    newTranslationAgent := func(language string) *agent.Agent {
        return foundryprovider.NewAgent(
            endpoint,
            token,
            foundryprovider.ModelDeployment(model),
            foundryprovider.AgentConfig{
                Instructions: fmt.Sprintf(
                    "Translate the user's text to %s. Return only the translation.",
                    language,
                ),
                Config: agent.Config{Name: language + "Agent"},
            },
        )
    }

    cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
    french := agentworkflow.New(newTranslationAgent("French"), cfg)
    spanish := agentworkflow.New(newTranslationAgent("Spanish"), cfg)
    english := agentworkflow.New(newTranslationAgent("English"), cfg)

    wf, err := workflow.NewBuilder(french).
        AddEdge(french, spanish).
        AddEdge(spanish, english).
        WithIntermediateOutputFrom(french, spanish).
        WithOutputFrom(english).
        Build()
    if err != nil {
        return err
    }

    run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    emitEvents := true
    if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
        return err
    } else if !accepted {
        return fmt.Errorf("workflow doesn't accept turn tokens")
    }

    lastExecutorID := ""
    for evt, streamErr := range run.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := evt.(type) {
        case workflow.OutputEvent:
            update, ok := event.Output.(*agent.ResponseUpdate)
            if !ok || update.String() == "" {
                continue
            }
            if event.ExecutorID != lastExecutorID {
                if lastExecutorID != "" {
                    fmt.Println()
                }
                fmt.Printf("%s: ", event.ExecutorID)
                lastExecutorID = event.ExecutorID
            }
            fmt.Print(update.String())
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    if lastExecutorID != "" {
        fmt.Println()
    }
    return nil
}

Run the example:

go run .

Warning

Tip

See the agents in workflows sample for another runnable translation pipeline.

Next Steps

Human-in-the-Loop

Feedback

Was this page helpful?

Last updated on 2026-10-28

Agents in Workflows

What You'll Build

Concepts Covered

Prerequisites

Step 1: Install NuGet packages

Step 2: Set Up Azure Foundry Client

Step 3: Create Agent Factory Method

Step 4: Create Specialized Azure Foundry Agents

Step 5: Build the Workflow

Step 6: Execute with Streaming

Step 7: Resource Cleanup

How It Works

Key Concepts

Complete Implementation

What You'll Build

Concepts Covered

Prerequisites

Step 1: Import Required Dependencies

Step 2: Create Azure OpenAI Responses Client

Step 3: Create Specialized Agents

Step 4: Build the Workflow

Step 5: Execute with Streaming

Step 6: Complete Main Function

How It Works

Key Concepts

Complete Implementation

What You'll Build

Concepts Covered

Prerequisites

Step 1: Set Up Foundry Configuration

Step 2: Create Agent Factory Function

Step 3: Create Specialized Foundry Agents

Step 4: Build the Workflow

Step 5: Execute with Streaming

How It Works

Key Concepts

Complete Implementation

Next Steps

Feedback

Additional resources