Microsoft Agent Framework

The mental model

Two layers: a single agent, and a Workflow of them.

Agent Framework has exactly two abstractions you build on. An Agent wraps a chat client (a model provider) with instructions, tools, and a session for memory: it runs the model-call → tool-call loop for you. A Workflow is a graph of executors connected by edges, where each executor can be a plain function, a class, or an AgentExecutor wrapping an Agent. Agents are for one reasoning loop; Workflows are for orchestrating many, with explicit topology, fan-out/fan-in, and checkpointing.

The framework is the merger of two predecessors. Semantic Kernel brought enterprise plumbing (connectors, filters, telemetry, .NET-first design); AutoGen brought event-driven multi-agent conversation. Agent Framework unifies both into one supported path, in Python and .NET (with a Go SDK also in the repo). If you were on either, this is where the community is converging.

flowchart LR
  U(["user input"]) --> A["Agent
instructions plus loop"]
  A <--> C["chat client
model provider"]
  A <--> T["function tools
your Python code"]
  A <--> S[("AgentSession
conversation memory")]
  A --> R(["AgentResponse"])
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class A,C model;
  class T,S code;
    

The lineage, at a glance

Install & setup

One meta-package, then pick a provider.

The Python distribution is agent-framework (Python 3.10+). The umbrella package pulls in the core plus optional provider extras. You add a credential library for whichever backend you target: azure-identity for Azure and Microsoft Foundry, or an OPENAI_API_KEY for OpenAI direct. A .NET distribution ships in parallel as the Microsoft.Agents.AI NuGet packages; this guide teaches Python.

# core + all provider extras
pip install agent-framework

# Azure / Microsoft Foundry path also wants a credential library
pip install azure-identity
az login                       # AzureCliCredential reads this session

# OpenAI-direct path just wants a key
export OPENAI_API_KEY="sk-..."

Hello world

A running agent in a handful of lines.

Everything is async. You build a chat client, wrap it in an Agent, and await agent.run(...). The same run call streams when you pass stream=True. Here we use FoundryChatClient, the Microsoft Foundry (Azure AI) provider, authenticating with your az login session.

import asyncio
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

async def main() -> None:
    client = FoundryChatClient(
        project_endpoint="https://your-project.services.ai.azure.com",
        model="gpt-4o",
        credential=AzureCliCredential(),
    )

    agent = Agent(
        client=client,
        name="HelloAgent",
        instructions="You are a friendly assistant. Keep your answers brief.",
    )

    result = await agent.run("What is the capital of France?")
    print(result.text)

asyncio.run(main())

no graph, no state schema, no boilerplate: the Agent runs the whole tool loop for you

Agents & chat clients

The Agent is provider-agnostic. Swap the client.

An Agent is the same object no matter who serves the model. What changes is the chat client you pass in. The provider extras live under agent_framework.*: OpenAIChatClient for OpenAI direct, the same client pointed at an Azure endpoint for Azure OpenAI, and FoundryChatClient for a Microsoft Foundry project. Instructions, tools, and sessions are provider-independent, so you can move an agent between backends by changing one constructor.

# OpenAI direct
from agent_framework.openai import OpenAIChatClient

agent = Agent(
    client=OpenAIChatClient(model="gpt-4o-mini", api_key=os.environ["OPENAI_API_KEY"]),
    name="WeatherAgent",
    instructions="You are a helpful weather agent.",
)

# Azure OpenAI: same client, Azure endpoint + credential
agent = Agent(
    client=OpenAIChatClient(
        model=os.environ["AZURE_OPENAI_MODEL"],
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        credential=AzureCliCredential(),
    ),
    name="WeatherAgent",
    instructions="You are a helpful weather agent.",
)

There is also a factory shorthand: many chat clients expose client.as_agent(name=..., instructions=...), which returns the same kind of Agent. Use whichever reads better. The constructor form is what the current samples lead with.

Tools

A tool is a decorated Python function.

Decorate a function with @tool and pass it in tools=[...]. The docstring becomes the tool description; type hints, especially Annotated[str, Field(description=...)], become the parameter schema the model sees. The model can then request a call, and the framework runs it and feeds the result back, looping until the model stops asking.

from typing import Annotated
from agent_framework import Agent, tool
from pydantic import Field

@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""   # docstring IS the tool description
    return f"The weather in {location} is sunny with a high of 24C."

agent = Agent(
    client=OpenAIChatClient(model="gpt-4o-mini"),
    name="WeatherAgent",
    instructions="Use the get_weather tool to answer questions.",
    tools=[get_weather],          # a single function or a list
)

result = await agent.run("What's the weather like in Seattle?")

Tools are not limited to your own functions. The framework speaks MCP (Model Context Protocol), so you can attach hosted or local MCP servers as tool sources; see the provider samples such as client_with_hosted_mcp.py and client_with_local_mcp.py for the current wiring. Providers also expose hosted tools like code interpreter, file search, and web search where the backend supports them.

Sessions & memory

Conversation memory is one object: a session.

By default each run is stateless. To carry history across turns, create an AgentSession with agent.create_session() and pass it on every call. The session accumulates the conversation so the agent remembers what came before. This replaces the older AgentThread / get_new_thread() naming you may see in earlier posts.

agent = Agent(
    client=OpenAIChatClient(model="gpt-4o-mini"),
    name="ConversationAgent",
    instructions="You are a friendly assistant. Keep your answers brief.",
)

session = agent.create_session()          # one conversation

await agent.run("My name is Alice and I love hiking.", session=session)
result = await agent.run("What do you remember about me?", session=session)
# -> the agent recalls "Alice" and "hiking" from the session

For memory that is richer than raw history (a running profile, retrieved facts, long-term recall across sessions) implement a ContextProvider. Its before_run hook injects instructions or context before the model sees the turn, and after_run writes back into session.state. That is where you plug in a vector store or a database.

from typing import Any
from agent_framework import AgentSession, ContextProvider, SessionContext

class UserMemoryProvider(ContextProvider):
    async def before_run(self, *, agent, session: AgentSession | None,
                          context: SessionContext, state: dict[str, Any]) -> None:
        if (name := state.get("user_name")):
            context.extend_instructions(self.source_id, f"The user's name is {name}.")

    async def after_run(self, *, agent, session, context: SessionContext,
                         state: dict[str, Any]) -> None:
        state["user_name"] = extract_name(context.input_messages)   # your code

agent = Agent(client=client, name="MemoryAgent",
              instructions="You are a friendly assistant.",
              context_providers=[UserMemoryProvider()])

Streaming

Same run method, one flag.

Pass stream=True and run returns an async stream of AgentResponseUpdate objects instead of a single response. Each update carries a slice of the output; read update.text for the token text, or drill into update.contents for tool calls, reasoning, and usage. This is what makes an agent feel responsive.

# iterate updates as they arrive, for a typing effect
async for update in agent.run("Tell me a one-sentence fun fact.", stream=True):
    if update.text:
        print(update.text, end="", flush=True)

# or grab the aggregated result after streaming, no re-iteration
stream = agent.run("Tell me a story", stream=True)
async for update in stream:
    print(update.text, end="")
final = await stream.get_final_response()      # full AgentResponse
print(final.text, len(final.messages))

The non-streaming run returns an AgentResponse with .text (the aggregated answer) and .messages (every message produced, including tool calls and results). Both shapes surface the same content; streaming just hands it to you incrementally.

Structured output

Hand it a Pydantic model, read a typed value.

Pass a Pydantic BaseModel as response_format in the run options and the agent returns parsed, type-safe data. The parsed object is on result.value; fall back to result.text if parsing failed. This is provider-backed structured output where the model supports it.

from pydantic import BaseModel
from agent_framework import Agent, AgentResponse

class CityInfo(BaseModel):
    city: str
    description: str

agent = Agent(client=OpenAIChatClient(), name="CityAgent",
              instructions="Describe cities in a structured format.")

result = await agent.run("Tell me about Paris, France",
                         options={"response_format": CityInfo})

if (data := result.value):          # parsed CityInfo instance
    print(data.city, data.description)

# streaming variant collects updates, then parses:
result = await AgentResponse.from_update_generator(
    agent.run("Tell me about Tokyo", stream=True, options={"response_format": CityInfo}),
    output_format_type=CityInfo,
)

Workflows · graph orchestration

When one agent is not enough, build a graph.

A Workflow is Agent Framework's multi-agent engine: a graph of executors joined by edges. An executor is a unit of work: a class subclassing Executor with @handler methods, a function decorated with @executor, or an AgentExecutor that wraps an Agent. Executors talk by passing typed messages: a WorkflowContext[T] lets a node send_message to the next executor or yield_output as the final result. Routing is type-based and edge-based: the graph moves a message to the executors wired to receive it.

from agent_framework import Executor, WorkflowBuilder, WorkflowContext, executor, handler
from typing_extensions import Never

class UpperCase(Executor):
    @handler
    async def to_upper(self, text: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message(text.upper())          # forward to next node

@executor(id="reverse_text")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    await ctx.yield_output(text[::-1])                # emit final output

upper = UpperCase(id="upper_case")
workflow = WorkflowBuilder(start_executor=upper).add_edge(upper, reverse_text).build()

events = await workflow.run("hello world")
print(events.get_outputs())          # ['DLROW OLLEH']

Fan-out and fan-in

The reason to reach for a Workflow is parallelism and joins. add_fan_out_edges sends one message to several executors that run concurrently; add_fan_in_edges collects their results into a list for a single aggregator. Execution follows a superstep (Pregel-style) model: each round routes pending messages, runs the targeted executors concurrently, and waits for all of them before advancing, which is what makes the join deterministic.

flowchart TB
  U(["user prompt"]) --> D["dispatcher
fan out"]
  D --> R["researcher
AgentExecutor"]
  D --> M["marketer
AgentExecutor"]
  D --> L["legal
AgentExecutor"]
  R --> AG["aggregator
fan in"]
  M --> AG
  L --> AG
  AG --> O(["consolidated report"])
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class R,M,L model;
  class D,AG code;
    

from agent_framework import AgentExecutor, WorkflowBuilder

# wrap each domain Agent as an executor node
researcher = AgentExecutor(Agent(client=client, name="researcher", instructions="..."))
marketer   = AgentExecutor(Agent(client=client, name="marketer",   instructions="..."))
legal      = AgentExecutor(Agent(client=client, name="legal",      instructions="..."))

workflow = (
    WorkflowBuilder(start_executor=dispatcher)
    .add_fan_out_edges(dispatcher, [researcher, marketer, legal])   # parallel
    .add_fan_in_edges([researcher, marketer, legal], aggregator)    # join
    .build()
)

An AgentExecutor exchanges AgentExecutorRequest and AgentExecutorResponse messages; the aggregator reads response.agent_response.text from each. Because the graph is a first-class object, Workflows also support checkpointing (resume a long run after a crash) and human-in-the-loop: a request-info step emits a request, the workflow pauses, and it resumes when your code supplies the answer (see the human-in-the-loop samples for the current request/response executor API). Higher-level orchestration presets (sequential, concurrent, group-chat, handoff, and the Magentic pattern) build on this same graph engine.

Agent or Workflow?

Middleware

Wrap every run and every tool call.

Middleware is a chain of interceptors around agent runs and function invocations, the place for cross-cutting concerns: auth checks, logging, timing, redaction, retries, result rewriting. A middleware is an async function that receives a context and a call_next; you do work, call await call_next() to continue, or return early to short-circuit. Register them with the middleware=[...] argument on the Agent.

from collections.abc import Awaitable, Callable
from agent_framework import Agent, AgentContext, FunctionInvocationContext

async def security_middleware(context: AgentContext,
                              call_next: Callable[[], Awaitable[None]]) -> None:
    last = context.messages[-1] if context.messages else None
    if last and last.text and "password" in last.text.lower():
        print("[security] blocked")
        return                          # do not call_next -> run is halted
    await call_next()

async def timing_middleware(context: FunctionInvocationContext,
                            call_next: Callable[[], Awaitable[None]]) -> None:
    print(f"calling {context.function.name}")   # wraps each tool call
    await call_next()

agent = Agent(client=client, name="GuardedAgent", instructions="...",
              tools=[get_weather],
              middleware=[security_middleware, timing_middleware])

The framework tells agent middleware (AgentContext) from function middleware (FunctionInvocationContext) by the parameter type, or you can be explicit with the @agent_middleware / @function_middleware decorators. Class-based middleware works too when you need state across calls.

Observability

OpenTelemetry, on by one call.

Agent Framework emits OpenTelemetry traces, logs, and metrics out of the box. Call configure_otel_providers() once and it wires up tracing, logging, and metrics from your OTLP environment variables, so every agent run, tool call, and Workflow step becomes a span you can inspect. Point the standard OTEL_EXPORTER_OTLP_ENDPOINT at a collector, Azure Monitor / Application Insights, or the Foundry tracing UI.

from agent_framework.observability import configure_otel_providers, get_tracer
from opentelemetry.trace import SpanKind

configure_otel_providers(enable_sensitive_data=True)   # reads OTEL_* env vars

with get_tracer().start_as_current_span("Scenario: Agent Chat", kind=SpanKind.CLIENT):
    agent = Agent(client=client, name="WeatherAgent",
                  instructions="You are a weather assistant.", tools=[get_weather])
    session = agent.create_session()
    async for update in agent.run("Weather in Amsterdam?", session=session, stream=True):
        if update.text:
            print(update.text, end="")

Putting it together

A support-triage Workflow, end to end.

Deterministic redaction in code before the model sees anything, a triage agent that classifies and can look up an order, and a fan-out to two specialist agents joined by an aggregator. One graph, code on the boundaries, models inside the boxes.

import asyncio
from typing import Annotated
from agent_framework import (
    Agent, AgentExecutor, Executor, WorkflowBuilder, WorkflowContext,
    executor, handler, tool,
)
from agent_framework.openai import OpenAIChatClient
from pydantic import Field
from typing_extensions import Never

client = OpenAIChatClient(model="gpt-4o-mini")

@tool(approval_mode="never_require")
def lookup_order(order_id: Annotated[str, Field(description="Order ID.")]) -> dict:
    """Look up an order status by ID."""
    return {"id": order_id, "status": "delayed"}

@executor(id="redact")
async def redact(text: str, ctx: WorkflowContext[str]) -> None:
    await ctx.send_message(strip_pii(text))     # code owns redaction, first

class Consolidate(Executor):
    @handler
    async def join(self, results: list, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output("\n\n".join(r.agent_response.text for r in results))

triage = AgentExecutor(Agent(client=client, name="triage", tools=[lookup_order],
    instructions="Classify the ticket and look up the order if an ID is present."))
billing = AgentExecutor(Agent(client=client, name="billing",
    instructions="Answer billing and refund questions."))
support = AgentExecutor(Agent(client=client, name="support",
    instructions="Answer shipping and product questions."))
consolidate = Consolidate(id="consolidate")

workflow = (
    WorkflowBuilder(start_executor=redact)
    .add_edge(redact, triage)
    .add_fan_out_edges(triage, [billing, support])
    .add_fan_in_edges([billing, support], consolidate)
    .build()
)

events = await workflow.run("My order 12345 is late and I want a refund.")
print(events.get_outputs())

Production & Azure

From script to service.

• Use a scoped credential in production. AzureCliCredential and DefaultAzureCredential are for dev; prefer ManagedIdentityCredential to avoid credential probing, latency, and fallback surprises.
• Set approval_mode="always_require" on any tool with a side effect, and gate it behind your own approval handler so a model can never issue a refund or delete unattended.
• Turn on OpenTelemetry with configure_otel_providers() and export to Azure Monitor / Application Insights or Foundry tracing; keep enable_sensitive_data off unless the backend is in your trust boundary.
• For long or multi-agent runs, use Workflow checkpointing so a crash resumes from the last superstep rather than restarting, and use request-info steps for human approval gates.
• Foundry gives you service-managed, persistent agents (server-side history and identity); the same Agent code targets it by swapping in FoundryChatClient. A parallel .NET stack (Microsoft.Agents.AI) exists for teams standardized on C#.
• Pin the version. On a 1.x line that is still moving fast, pin agent-framework==<exact> and read the release notes before you bump.

Gotchas

What trips people up.

Reach for it when

• You are a .NET or Azure shop and want first-class provider, identity, and Foundry integration.
• You need enterprise governance: managed identity, OpenTelemetry, approval gates, and a supported roadmap.
• You were on Semantic Kernel or AutoGen and want the consolidated, forward path.
• Your problem is genuinely multi-agent: parallel branches and joins that the Workflow graph models cleanly.

Skip it when

• It is a single OpenAI-native agent with one or two tools: the vendor SDK or a lighter framework is less to carry.
• You want a TypeScript-first stack: Python and .NET are the first-class languages here, with Go in the repo.
• You need API stability today: on a fast-moving 1.x line, pin hard and expect churn between minors.

The principle it teaches

Where it fits your labs

A Week 4 reference for the enterprise end of the multi-agent spectrum. The orchestrator-workers pattern you build by hand is an Agent Framework Workflow with fan-out and fan-in; the threat-modeling and approval-gating discipline you apply everywhere is approval_mode="always_require" and middleware here. Carry it for awareness: when a stakeholder says "but it has to run on Azure with managed identity," this is the answer you should recognise.