microsoft-agent-framework

using Microsoft.Agents.AI;
 
var agent = new Agent(
    name: "assistant",
    model: new ModelClient(model: "gpt-4"),
    instructions: "You are a helpful assistant"
);
 
var response = await agent.RunAsync("Hello!");
Console.WriteLine(response.Content);

from agents_framework import function_tool
 
@function_tool
def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"Weather in {location}: Sunny, 72°F"
 
agent = Agent(
    name="assistant",
    model=ModelClient(model="gpt-4"),
    tools=[get_weather]
)
 
response = await agent.run(message="What's the weather in Seattle?")
# Agent automatically calls get_weather() and responds with result

[FunctionTool]
public static string GetWeather(string location)
{
    return $"Weather in {location}: Sunny, 72°F";
}
 
var agent = new Agent(
    name: "assistant",
    model: new ModelClient(model: "gpt-4"),
    tools: new[] { typeof(Tools).GetMethod("GetWeather") }
);

from agents_framework import Workflow, GraphWorkflow
 
# Define workflow graph
workflow = GraphWorkflow()
 
# Add agents as nodes
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", review_agent)
 
# Define edges (control flow)
workflow.add_edge("researcher", "writer")  # Sequential
workflow.add_edge("writer", "reviewer")
 
# Conditional routing
def should_revise(state):
    return state.get("needs_revision", False)
 
workflow.add_conditional_edge(
    "reviewer",
    should_revise,
    {"revise": "writer", "done": "END"}
)
 
# Execute workflow
result = await workflow.run(initial_message="Research AI trends")

var workflow = new GraphWorkflow();
 
workflow.AddNode("researcher", researchAgent);
workflow.AddNode("writer", writerAgent);
workflow.AddNode("reviewer", reviewAgent);
 
workflow.AddEdge("researcher", "writer");
workflow.AddEdge("writer", "reviewer");
 
var result = await workflow.RunAsync("Research AI trends");

from agents_framework import Thread, ContextProvider
 
# Thread maintains conversation history
thread = Thread()
await agent.run(thread=thread, message="Remember: My name is Alice")
await agent.run(thread=thread, message="What's my name?")  # "Alice"
 
# Custom context provider
class DatabaseContext(ContextProvider):
    async def get_context(self, thread_id: str):
        return await db.fetch_history(thread_id)
 
    async def save_context(self, thread_id: str, messages):
        await db.save_history(thread_id, messages)
 
agent = Agent(model=model, context_provider=DatabaseContext())

from agents_framework import Middleware
from opentelemetry import trace
 
# Custom middleware
class LoggingMiddleware(Middleware):
    async def process(self, message, next_handler):
        print(f"Processing: {message.content}")
        response = await next_handler(message)
        print(f"Response: {response.content}")
        return response
 
# OpenTelemetry integration
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("agent-run"):
    response = await agent.run(message="Hello")

public class LoggingMiddleware : IMiddleware
{
    public async Task<Message> ProcessAsync(Message message, Func<Message, Task<Message>> next)
    {
        Console.WriteLine($"Processing: {message.Content}");
        var response = await next(message);
        Console.WriteLine($"Response: {response.Content}");
        return response;
    }
}

from agents_framework import HumanInTheLoop
 
@function_tool
def delete_file(path: str) -> str:
    """Delete a file (requires approval)."""
    return f"Deleted {path}"
 
# Add approval wrapper
delete_file_with_approval = HumanInTheLoop(
    tool=delete_file,
    approval_prompt="Approve deletion of {path}?"
)
 
agent = Agent(tools=[delete_file_with_approval])

workflow = GraphWorkflow()
 
# Add multiple agents
workflow.add_node("analyst1", analyst_agent)
workflow.add_node("analyst2", analyst_agent)
workflow.add_node("synthesizer", synthesis_agent)
 
# Parallel execution
workflow.add_edge("START", ["analyst1", "analyst2"])  # Both run in parallel
workflow.add_edge(["analyst1", "analyst2"], "synthesizer")  # Wait for both
 
result = await workflow.run(message="Analyze market trends")

from pydantic import BaseModel
 
class WeatherReport(BaseModel):
    location: str
    temperature: float
    conditions: str
 
agent = Agent(
    model=model,
    instructions="Generate weather reports",
    response_format=WeatherReport
)
 
response = await agent.run(message="Weather in Seattle")
report: WeatherReport = response.parsed
print(f"{report.location}: {report.temperature}°F, {report.conditions}")

from agents_framework import RetryPolicy
 
agent = Agent(
    model=model,
    retry_policy=RetryPolicy(
        max_retries=3,
        backoff_factor=2.0,
        exceptions=[TimeoutError, ConnectionError]
    )
)
 
try:
    response = await agent.run(message="Hello")
except Exception as e:
    print(f"Failed after retries: {e}")

# Use amplihack for orchestration
from claude import Agent as ClaudeAgent
 
orchestrator = ClaudeAgent("orchestrator.md")
 
# Delegate to Agent Framework for stateful agents
from agents_framework import Agent, Thread
 
conversational_agent = Agent(
    model=ModelClient(model="gpt-4"),
    instructions="Maintain conversation context"
)
 
thread = Thread()
response1 = await conversational_agent.run(thread=thread, message="Start task")
response2 = await conversational_agent.run(thread=thread, message="Continue")
 
# Use amplihack for final synthesis
result = orchestrator.process({"responses": [response1, response2]})