Building Your First AI Agent with Microsoft Agent Framework (Python) – Part 3

# Create project directory
mkdir maf-research-agent
cd maf-research-agent

# Create and activate virtual environment
python -m venv .venv

# Windows
.venv\Scripts\activate

# macOS/Linux
source .venv/bin/activate

# Install the full Agent Framework package
pip install agent-framework --pre

# Or install specific components
pip install agent-framework-azure-ai --pre
pip install agent-framework-openai --pre

# Additional dependencies
pip install python-dotenv aiohttp

# .env
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-4o
AZURE_OPENAI_API_VERSION=2024-08-01-preview

import asyncio
import os
from dotenv import load_dotenv
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

# Load environment variables
load_dotenv()

async def main():
    # Create Azure OpenAI client with CLI credential
    agent = AzureOpenAIResponsesClient(
        credential=AzureCliCredential(),
        endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        deployment_name=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
    ).create_agent(
        name="ResearchAssistant",
        instructions="""
            You are a research assistant that helps users find and understand
            information on various topics. You provide well-sourced, accurate
            answers and always cite your reasoning.
        """
    )
    
    # Run the agent
    print("🔬 Research Assistant Ready!")
    result = await agent.run("What are the key benefits of agentic AI in enterprise?")
    print(f"\nAssistant: {result.text}")

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

async def stream_response():
    print("\nAssistant: ", end="", flush=True)
    
    async for update in agent.run_stream("Explain quantum computing in simple terms"):
        if update.text:
            print(update.text, end="", flush=True)
    
    print("\n")  # New line after response

from typing import Annotated
from pydantic import Field
from agent_framework import ai_function
import aiohttp

@ai_function
async def search_web(
    query: Annotated[str, Field(description="The search query")],
    num_results: Annotated[int, Field(description="Number of results to return")] = 5
) -> str:
    """
    Search the web for information on a topic.
    Returns a summary of the top search results.
    """
    # In production, integrate with Bing Search API, Google, etc.
    # For demo, we'll return mock results
    return f"""
    Search results for '{query}':
    1. Enterprise AI Adoption Trends 2025 - Microsoft Research
    2. How Agentic AI is Transforming Business Processes - Harvard Business Review
    3. Building Production AI Agents - Azure AI Blog
    4. The Future of Autonomous AI Systems - MIT Technology Review
    5. Agentic AI Best Practices - Gartner
    """

@ai_function
def summarize_document(
    content: Annotated[str, Field(description="The document content to summarize")],
    max_length: Annotated[int, Field(description="Maximum summary length in words")] = 100
) -> str:
    """
    Summarize a document or article content.
    Returns a concise summary within the specified word limit.
    """
    # In production, this could call another LLM or use extractive summarization
    words = content.split()
    if len(words) <= max_length:
        return content
    return ' '.join(words[:max_length]) + '...'

@ai_function
def add_citation(
    source_title: Annotated[str, Field(description="Title of the source")],
    source_url: Annotated[str, Field(description="URL of the source")],
    author: Annotated[str, Field(description="Author name")] = "Unknown",
    year: Annotated[int, Field(description="Publication year")] = 2025
) -> str:
    """
    Format a citation in APA style.
    """
    return f"{author} ({year}). {source_title}. Retrieved from {source_url}"

@ai_function
def get_current_date() -> str:
    """
    Get the current date. Useful for time-sensitive research.
    """
    from datetime import datetime
    return datetime.now().strftime("%B %d, %Y")

# Create agent with tools
agent = AzureOpenAIResponsesClient(
    credential=AzureCliCredential(),
).create_agent(
    name="ResearchAssistant",
    instructions="""
        You are a research assistant that helps users find and understand
        information. You have access to:
        - search_web: Find information online
        - summarize_document: Create concise summaries
        - add_citation: Format citations properly
        - get_current_date: Get today's date
        
        Always cite your sources and provide accurate information.
    """,
    tools=[search_web, summarize_document, add_citation, get_current_date]
)

async def research_session():
    # Create a thread for the research session
    thread = agent.get_new_thread()
    
    print("\n═══════════════════════════════════════════════════════════")
    print("  🔬 Research Assistant - Multi-Turn Session")
    print("═══════════════════════════════════════════════════════════")
    print("  Type 'exit' to end the session.\n")
    
    while True:
        user_input = input("You: ").strip()
        
        if not user_input:
            continue
        if user_input.lower() == 'exit':
            break
        
        print("\nResearching...", end="", flush=True)
        
        # Run with thread - maintains conversation history
        async for update in agent.run_stream(user_input, thread):
            if update.text:
                print(f"\r\033[KAssistant: {update.text}", end="", flush=True)
        
        print("\n")

"""
Research Assistant Agent
Built with Microsoft Agent Framework

Author: Nithin Mohan T K
"""

import asyncio
import os
import logging
from typing import Annotated
from datetime import datetime

from dotenv import load_dotenv
from pydantic import Field
from azure.identity import AzureCliCredential

from agent_framework import ai_function
from agent_framework.azure import AzureOpenAIResponsesClient

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# ============================================================
# TOOLS
# ============================================================

@ai_function
async def search_web(
    query: Annotated[str, Field(description="The search query")],
    num_results: Annotated[int, Field(description="Number of results")] = 5
) -> str:
    """Search the web for information on a topic."""
    logger.info(f"Searching web for: {query}")
    return f"""
    Top {num_results} results for '{query}':
    1. [Enterprise AI Guide] - Microsoft Research
    2. [Agentic AI Patterns] - Harvard Business Review
    3. [Building AI Agents] - Azure AI Blog
    4. [AI Agent Architecture] - MIT Technology Review
    5. [Production AI Systems] - Google AI Blog
    """

@ai_function
def summarize_text(
    content: Annotated[str, Field(description="Content to summarize")],
    max_words: Annotated[int, Field(description="Max words")] = 100
) -> str:
    """Create a concise summary of the provided content."""
    logger.info(f"Summarizing content ({len(content)} chars)")
    words = content.split()
    if len(words) <= max_words:
        return content
    return ' '.join(words[:max_words]) + '...'

@ai_function
def format_citation(
    title: Annotated[str, Field(description="Source title")],
    url: Annotated[str, Field(description="Source URL")],
    author: Annotated[str, Field(description="Author")] = "Unknown",
    year: Annotated[int, Field(description="Year")] = 2025
) -> str:
    """Format a citation in APA style."""
    return f"{author} ({year}). {title}. Retrieved from {url}"

@ai_function
def get_date() -> str:
    """Get the current date."""
    return datetime.now().strftime("%B %d, %Y")

# ============================================================
# MAIN APPLICATION
# ============================================================

async def main():
    try:
        # Initialize agent
        agent = AzureOpenAIResponsesClient(
            credential=AzureCliCredential(),
            endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        ).create_agent(
            name="ResearchAssistant",
            instructions="""
                You are an expert research assistant. Your capabilities:
                
                1. Search the web for current information
                2. Summarize complex content concisely
                3. Provide properly formatted citations
                4. Give accurate, well-sourced answers
                
                Always:
                - Cite your sources
                - Be objective and accurate
                - Acknowledge uncertainty when appropriate
                - Use tools to find real information
            """,
            tools=[search_web, summarize_text, format_citation, get_date]
        )
        
        # Create conversation thread
        thread = agent.get_new_thread()
        
        print("\n" + "=" * 60)
        print("  Research Assistant v1.0")
        print("  Powered by Microsoft Agent Framework")
        print("=" * 60)
        print("  Commands: 'exit' to quit, 'clear' for new session\n")
        
        while True:
            try:
                user_input = input("\nYou: ").strip()
                
                if not user_input:
                    continue
                
                if user_input.lower() == 'exit':
                    print("\nThank you for using Research Assistant!")
                    break
                
                if user_input.lower() == 'clear':
                    thread = agent.get_new_thread()
                    print("\nSession cleared. Starting fresh!")
                    continue
                
                # Stream the response
                print("\nAssistant: ", end="", flush=True)
                
                start_time = datetime.now()
                response_text = ""
                
                async for update in agent.run_stream(user_input, thread):
                    if update.text:
                        print(update.text, end="", flush=True)
                        response_text += update.text
                
                elapsed = (datetime.now() - start_time).total_seconds()
                print(f"\n\n[Response time: {elapsed:.2f}s]")
                
            except KeyboardInterrupt:
                print("\n\nInterrupted. Type 'exit' to quit.")
                
    except Exception as e:
        logger.error(f"Application error: {e}", exc_info=True)
        raise

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

📝 You: What are the latest trends in enterprise AI adoption?

🤖 Assistant: [Uses search_web tool]
Based on my research, here are the key trends in enterprise AI adoption for 2025:

1. **Agentic AI** - Moving beyond chatbots to autonomous agents...
2. **Multi-Agent Systems** - Organizations deploying specialized agents...
3. **Observability & Governance** - Focus on AI transparency...

Citations:
- Microsoft Research (2025). Enterprise AI Adoption Trends...

📝 You: Can you summarize the first point in 50 words?

🤖 Assistant: [Uses summarize_text tool with conversation context]
Agentic AI represents the shift from reactive chatbots to proactive autonomous agents that can reason, plan, and execute multi-step tasks with minimal human intervention, fundamentally changing how enterprises automate complex workflows.