Files
microsoft--agent-framework/python
Vaibhav Patel 3a5d00be54 Python: add checkpointing support to AgentFrameworkWorkflow.run() in agent-framework-ag-ui (#6646)
* Python: add checkpointing support to AgentFrameworkWorkflow.run() in ag-ui

The ag-ui AgentFrameworkWorkflow.run() previously accepted only a
RunAgentInput payload and exposed no way to use the core workflow's
checkpointing/state-persistence, unlike the core agent-framework workflow
implementations. This left ag-ui workflows without resumable execution.

Add optional checkpoint_storage and checkpoint_id keyword arguments to
run(), threaded through run_workflow_stream() into the core Workflow.run().
This delegates to the existing core capability instead of reinventing it and
keeps the public surface consistent with Workflow.run():

- checkpoint_storage enables checkpoint creation at each superstep boundary.
- checkpoint_id resumes a run from a persisted checkpoint; incoming messages
  are forwarded only as request-info responses (never as a new start-executor
  message) to honor the core's message/checkpoint_id mutual exclusivity, and
  responses + checkpoint_id performs a restore-then-send in one call.

Both can also be supplied via the input_data keys __ag_ui_checkpoint_storage
and __ag_ui_checkpoint_id so the FastAPI endpoint (which calls run(input_data)
positionally) can opt in without changing its call site; explicit keyword
arguments take precedence. Checkpoint resume bypasses the AG-UI thread snapshot
hydration early-returns so it always reaches the core restore path.

Backward compatible: run(input_data) keeps working unchanged, and the
non-checkpoint path still calls run_workflow_stream(input_data, workflow) with
its original two-argument convention. Adds focused tests covering checkpoint
creation, resume-from-checkpoint, input-data-keyed params, and the unchanged
default path.

Fixes #6632.

* Import Executor from the public agent_framework API in ag-ui workflow test

* Fix ag-ui checkpoint resume: preserve thread snapshot, coerce resume responses; fix CI lint/typing

A checkpoint-only resume no longer clobbers the stored AG-UI thread snapshot:
the snapshot builder is seeded with the prior stored history so the saved
snapshot keeps the earlier replayable transcript plus the newly produced output.

Resume responses are now coerced against the post-restore pending requests on a
checkpoint restore, so a JSON function_approval_response resumes through AG-UI
after a cold restore instead of failing with a response-type mismatch.

Also update the test-double workflow run() overrides to match the new keyword-only
parent signature and re-sort the workflow test imports so ruff and the typing
checkers pass.

* Coerce ag-ui resume responses without a second checkpoint restore

Reading pending request_info events for resume-response coercion previously
restored the checkpoint into the live workflow, which invoked every executor's
on_checkpoint_restore hook. workflow.run(checkpoint_id=...) then restored again,
running those hooks a second time. Custom restore hooks are not required to be
idempotent, so this could duplicate restoration work or break workflows that
expect exactly one restore per resume.

Load the persisted WorkflowCheckpoint directly from storage (runtime override
or the workflow's build-time context storage) and read its
pending_request_info_events instead. This exposes the same post-restore pending
set for the resume contract and response coercion without mutating workflow
state or running any restore hook, leaving workflow.run(checkpoint_id=...) as
the single restore per resume.

Add a regression test asserting on_checkpoint_restore runs exactly once on a
checkpointed ag-ui resume.

* Python: rework AG-UI workflow checkpointing onto public configuration surfaces

Checkpoint storage is now configured on AgentFrameworkWorkflow (or the
FastAPI endpoint) instead of being smuggled through input_data keys, and
a run resumes by supplying its checkpoint id in the AG-UI forwarded
props. With storage always in hand, resume-response coercion reads the
pending request set straight from the persisted checkpoint via the
public CheckpointStorage.load(), replacing the private runner-context
fallback, and the core run call forwards checkpoint arguments directly,
relying on core validation for conflicting parameters. Requesting a
resume without configured storage now fails with a clear error.

* Assign endpoint checkpoint storage in a single place

The raw-workflow branch assigned checkpoint_storage at construction and
the wiring block assigned it again. Construct the wrapper bare and let
the wiring block own the assignment; the existing-storage guard keeps
allowing a pre-wrapped runner without storage to adopt the endpoint's.

---------

Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>
Co-authored-by: Evan Mattson <evan.mattson@microsoft.com>
2026-08-12 22:07:52 +00:00
..
2026-07-14 06:44:26 +00:00

Get Started with Microsoft Agent Framework for Python Developers

Quick Install

We recommend two common installation paths depending on your use case.

1. Development mode

If you are exploring or developing locally, install the entire framework with all sub-packages:

pip install agent-framework

This installs the core and every integration package, making sure that all features are available without additional steps. This is the simplest way to get started.

2. Selective install

If you only need specific integrations, you can install at a more granular level. This keeps dependencies lighter and focuses on what you actually plan to use. Some examples:

# Core only
# includes Azure OpenAI and OpenAI support by default
# also includes workflows and orchestrations
pip install agent-framework-core

# Core + Microsoft Foundry integration
pip install agent-framework-foundry

# Core + Microsoft Copilot Studio integration (preview package)
pip install agent-framework-copilotstudio --pre

# Core + both Microsoft Copilot Studio and Microsoft Foundry integration
pip install --pre agent-framework-copilotstudio agent-framework-foundry

This selective approach is useful when you know which integrations you need, and it is the recommended way to set up lightweight environments. Released packages such as agent-framework, agent-framework-core, and agent-framework-foundry no longer require --pre, while preview connectors such as agent-framework-copilotstudio still do.

Supported Platforms:

  • Python: 3.10+
  • OS: Windows, macOS, Linux

1. Setup API Keys

Set as environment variables, or create a .env file at your project root:

OPENAI_API_KEY=sk-...
OPENAI_MODEL=...
...
AZURE_OPENAI_API_KEY=...
AZURE_OPENAI_ENDPOINT=...
AZURE_OPENAI_MODEL=...
...
FOUNDRY_PROJECT_ENDPOINT=...
FOUNDRY_MODEL=...

For the generic OpenAI clients (OpenAIChatClient and OpenAIChatCompletionClient), configuration resolves in this order:

  1. Explicit Azure inputs such as credential or azure_endpoint
  2. OPENAI_API_KEY / explicit OpenAI API-key parameters
  3. Azure environment fallback such as AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY

This means mixed shells default to OpenAI when OPENAI_API_KEY is present. To force Azure routing, pass an explicit Azure input such as credential=AzureCliCredential().

You can also override environment variables by explicitly passing configuration parameters to the chat client constructor:

from agent_framework.openai import OpenAIChatClient

client = OpenAIChatClient(
    api_key='',
    azure_endpoint='',
    model='',
    api_version='',
)

See the following setup guide for more information.

2. Create a Simple Agent

Create agents and invoke them directly:

import asyncio
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient

async def main():
    agent = Agent(
        client=OpenAIChatClient(),
        instructions="""
        1) A robot may not injure a human being...
        2) A robot must obey orders given it by human beings...
        3) A robot must protect its own existence...

        Give me the TLDR in exactly 5 words.
        """
    )

    result = await agent.run("Summarize the Three Laws of Robotics")
    print(result)

asyncio.run(main())
# Output: Protect humans, obey, self-preserve, prioritized.

3. Directly Use Chat Clients (No Agent Required)

You can use the chat client classes directly for advanced workflows:

import asyncio
from agent_framework import Message
from agent_framework.openai import OpenAIChatClient

async def main():
    client = OpenAIChatClient()

    messages = [
        Message("system", ["You are a helpful assistant."]),
        Message("user", ["Write a haiku about Agent Framework."])
    ]

    response = await client.get_response(messages)
    print(response.messages[0].text)

    """
    Output:

    Agents work in sync,
    Framework threads through each task—
    Code sparks collaboration.
    """

asyncio.run(main())

4. Build an Agent with Tools and Functions

Enhance your agent with custom tools and function calling:

import asyncio
from typing import Annotated
from random import randint
from pydantic import Field
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient


def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


def get_menu_specials() -> str:
    """Get today's menu specials."""
    return """
    Special Soup: Clam Chowder
    Special Salad: Cobb Salad
    Special Drink: Chai Tea
    """


async def main():
    agent = Agent(
        client=OpenAIChatClient(),
        instructions="You are a helpful assistant that can provide weather and restaurant information.",
        tools=[get_weather, get_menu_specials]
    )

    response = await agent.run("What's the weather in Amsterdam and what are today's specials?")
    print(response)

    """
    Output:
    The weather in Amsterdam is sunny with a high of 22°C. Today's specials include
    Clam Chowder soup, Cobb Salad, and Chai Tea as the special drink.
    """

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

You can explore additional agent samples here.

5. Multi-Agent Orchestration

Coordinate multiple agents to collaborate on complex tasks using orchestration patterns:

import asyncio
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient


async def main():
    # Create specialized agents
    writer = Agent(
        client=OpenAIChatClient(),
        name="Writer",
        instructions="You are a creative content writer. Generate and refine slogans based on feedback."
    )

    reviewer = Agent(
        client=OpenAIChatClient(),
        name="Reviewer",
        instructions="You are a critical reviewer. Provide detailed feedback on proposed slogans."
    )

    # Sequential workflow: Writer creates, Reviewer provides feedback
    task = "Create a slogan for a new electric SUV that is affordable and fun to drive."

    # Step 1: Writer creates initial slogan
    initial_result = await writer.run(task)
    print(f"Writer: {initial_result}")

    # Step 2: Reviewer provides feedback
    feedback_request = f"Please review this slogan: {initial_result}"
    feedback = await reviewer.run(feedback_request)
    print(f"Reviewer: {feedback}")

    # Step 3: Writer refines based on feedback
    refinement_request = f"Please refine this slogan based on the feedback: {initial_result}\nFeedback: {feedback}"
    final_result = await writer.run(refinement_request)
    print(f"Final Slogan: {final_result}")

    # Example Output:
    # Writer: "Charge Forward: Affordable Adventure Awaits!"
    # Reviewer: "Good energy, but 'Charge Forward' is overused in EV marketing..."
    # Final Slogan: "Power Up Your Adventure: Premium Feel, Smart Price!"

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

For more advanced orchestration patterns including Sequential, Concurrent, Group Chat, Handoff, and Magentic orchestrations, see the orchestration samples.

More Examples & Samples

Agent Framework Documentation