Building an AI Agent with AutoGen (Now Microsoft Agent Framework)

1pip install agent-framework agent-framework-anthropic --pre
2export ANTHROPIC_API_KEY="sk-ant-..."

1from typing import Annotated
2from agent_framework import tool
3
4VENDORS = {
5    "VEN-220": {"name": "Acme Supply Co", "terms": "Net 30", "ytd_spend": 48000},
6}
7
8@tool(approval_mode="never_require")
9def look_up_vendor_contract(
10    vendor_id: Annotated[str, "The vendor ID, e.g. VEN-220"],
11) -> str:
12    """Look up a vendor's contract terms and year-to-date spend."""
13    vendor = VENDORS.get(vendor_id)
14    if not vendor:
15        return f"No vendor found for {vendor_id}"
16    return f"{vendor['name']}: terms={vendor['terms']}, ytd_spend=${vendor['ytd_spend']}"

1@tool(approval_mode="always_require")
2def approve_purchase_order(
3    vendor_id: Annotated[str, "The vendor ID"],
4    amount: Annotated[float, "Purchase order amount in dollars"],
5    reason: Annotated[str, "What the purchase order is for"],
6) -> str:
7    """Approve a purchase order for a vendor. Always requires human approval before it runs."""
8    # In production this posts to your procurement/ERP system
9    return f"Approved PO for {vendor_id}: ${amount:.2f} ({reason})"

1from agent_framework import Agent
2from agent_framework.anthropic import AnthropicClient
3
4agent = Agent(
5    client=AnthropicClient(model="claude-sonnet-5-5"),
6    name="ProcurementAgent",
7    instructions=(
8        "You are a procurement agent. Always call look_up_vendor_contract before making any claim "
9        "about a vendor. Never tell someone a purchase order was approved without actually calling "
10        "approve_purchase_order."
11    ),
12    tools=[look_up_vendor_contract, approve_purchase_order],
13)

1import asyncio
2
3async def handle_approvals(query: str, agent, session) -> str:
4    result = await agent.run(query, session=session)
5    while result.user_input_requests:
6        new_inputs = []
7        for request in result.user_input_requests:
8            call = request.function_call
9            print(f"[APPROVAL NEEDED] {call.name}({call.arguments})")
10            approved = True  # swap for a real Slack/ticket check
11            new_inputs.append({"role": "user", "contents": [request.to_function_approval_response(approved)]})
12        result = await agent.run(new_inputs, session=session)
13    return result.text
14
15async def main():
16    session = agent.create_session()
17    text = await handle_approvals(
18        "VEN-220 needs a $4,200 PO for a replacement HVAC unit. Check their terms and approve it.",
19        agent,
20        session,
21    )
22    print(text)
23
24asyncio.run(main())

1from agent_framework import WorkflowBuilder, WorkflowContext, Executor, handler, response_handler
2from dataclasses import dataclass
3
4@dataclass
5class POApprovalRequest:
6    vendor_id: str
7    amount: float
8
9class FinanceReviewer(Executor):
10    @handler
11    async def review(self, po: POApprovalRequest, ctx: WorkflowContext) -> None:
12        await ctx.request_info(request_data=po, response_type=str)
13
14    @response_handler
15    async def on_decision(self, original: POApprovalRequest, decision: str, ctx: WorkflowContext) -> None:
16        await ctx.yield_output(f"{'APPROVED' if decision == 'approved' else 'REJECTED'}: {original.vendor_id}")
17
18# workflow = WorkflowBuilder(start_executor=procurement_agent_executor).add_edge(
19#     procurement_agent_executor, FinanceReviewer(id="finance-reviewer")
20# ).build()

1@tool(approval_mode="always_require")
2def approve_purchase_order(
3    vendor_id: Annotated[str, "The vendor ID"],
4    amount: Annotated[float, "Purchase order amount in dollars"],
5    reason: Annotated[str, "What the purchase order is for"],
6) -> str:
7    """Approve a purchase order for a vendor. Always requires human approval before it runs."""
8    try:
9        po = erp_client.create_purchase_order(vendor_id=vendor_id, amount=amount, memo=reason)
10        return f"Approved PO {po.id} for {vendor_id}: ${amount:.2f}"
11    except Exception as e:
12        return f"PO creation failed: {e}"