Microsoft Agent Framework Basics

# pip install agent-framework
import asyncio
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
 
async def main():
    client = OpenAIChatClient(model="gpt-4o")  # model id: verify at build
    agent = client.as_agent(
        name="HelloAgent",
        instructions="You are a friendly assistant. Keep answers brief.",
    )
    result = await agent.run("Name one benefit of multi-agent systems.")
    print(result.text)
 
asyncio.run(main())

async def chat_with_memory(agent):
    session = agent.create_session()
    r1 = await agent.run("Remember the codeword is azure-lark.", session=session)
    r2 = await agent.run("What is the codeword?", session=session)
    print(r1.text, r2.text)

from agent_framework import tool
from typing import Annotated
 
@tool
def word_count(text: Annotated[str, "Text to measure"]) -> str:
    """Count words in text."""
    return str(len(text.split()))
 
agent = client.as_agent(
    name="Counter",
    instructions="Use word_count when the user asks for a count.",
    tools=[word_count],
)
print((await agent.run("How many words are in: multi agent systems rock?")).text)

writer = client.as_agent(
    name="writer",
    instructions="You are a concise copywriter. Output one punchy marketing sentence.",
)
reviewer = client.as_agent(
    name="reviewer",
    instructions="You are a reviewer. Give brief feedback on the previous message only.",
)

from typing import cast
from agent_framework import Message
from agent_framework.orchestrations import SequentialBuilder
 
async def two_agent_handoff(client):
    writer = client.as_agent(
        name="writer",
        instructions="Write one punchy product tagline. No preamble.",
    )
    reviewer = client.as_agent(
        name="reviewer",
        instructions="Critique the tagline in two short bullets.",
    )
    workflow = SequentialBuilder(participants=[writer, reviewer]).build()
    task = "Tagline for an enterprise multi-agent SDK."
    async for event in workflow.run(task, stream=True):
        if event.type == "output":
            history = cast(list[Message], event.data)
            for msg in history:
                print(f"[{msg.author_name or msg.role}]: {msg.text}")

from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
 
client = FoundryChatClient(
    project_endpoint="https://YOUR_PROJECT.services.ai.azure.com/api/projects/YOUR_PROJECT",
    model="gpt-4o",  # deployment / model: verify at build
    credential=AzureCliCredential(),
)
agent = client.as_agent(name="FoundryHello", instructions="Be brief.")

researcher = client.as_agent(
    name="researcher",
    instructions="Answer with three bullet facts only.",
)
research_tool = researcher.as_tool(
    name="research_bullets",
    description="Produce three factual bullets on a topic",
    arg_name="topic",
    arg_description="Topic to research",
)
coordinator = client.as_agent(
    name="coordinator",
    instructions="Use research_bullets, then write a one-sentence summary.",
    tools=[research_tool],
)
print((await coordinator.run("Benefits of checkpointed workflows")).text)

from agent_framework import AgentContext
from collections.abc import Awaitable, Callable
 
async def log_runs(context: AgentContext, call_next: Callable[[], Awaitable[None]]) -> None:
    print("start", context.agent.name if context.agent else "agent")
    await call_next()
    print("done")
 
agent = client.as_agent(
    name="Logged",
    instructions="Reply in one sentence.",
    middleware=[log_runs],
)

# Decision sketch (not framework API)
def choose_shape(task: str) -> str:
    if task in {"tagline_review", "draft_then_edit"}:
        return "sequential_orchestration"
    if task in {"fan_out_research", "approval_gate", "checkpointed_etl"}:
        return "workflow_graph"
    return "single_agent"