Building a Web Search Workflow Using Microsoft Agent Framework (Without Any AI Services) -Part II

Introduction

In our previous blogs, we explored what Microsoft Agent Framework is, what AI Agents are, and how workflows differ from them.

1. https://dev.to/sreeni5018/microsoft-agent-framework-combining-semantic-kernel-autogen-for-advanced-ai-agents-2i4i

2. https://dev.to/sreeni5018/building-smart-workflows-with-the-microsoft-agent-framework-part-i-2fej

Now, let’s take the next step we’ll build our first workflow using the Microsoft Agent Framework, without using any AI or LLM based services.

This post marks hands-on series where we’ll build:

1. Workflows without using LLMs or AI services
2. AI Agents inside workflows (Workflow + AI Agent)
3. Conditional workflows
4. Checkpointing and resuming workflows (Hydration and Dehydration)

Today, we’ll start simple and build a Web Search Workflow that connects executors and functions to perform a multi-step task no AI involved.

Understanding Microsoft Agent Workflows

Using the Microsoft Agent Framework, we can create two main types of AI-powered applications:

AI Agents: These use LLMs to process inputs, reason about tasks, call external tools or MCP servers, perform actions, and generate responses.

Graph-based Workflows: These connect multiple agents and non-agentic functions to perform multi-step business tasks.

Interestingly, not all workflows require an LLM. Some can execute logical or data-processing tasks purely through function execution, such as calculating tax, formatting data, or performing API-based searches.

The real strength of AI Agent + Workflow combination lies in building complex, end-to-end business processes and modernizing legacy systems with modular, reusable components.

Key Benefits of Microsoft Agent Workflows

Modularity- Build reusable, smaller components that are easy to manage and update.

Agent Integration – Combine multiple AI agents with regular logic for complex orchestration.

Type Safety – Ensure correct message flow with strong typing and validation.

Flexible Flow – Model workflows visually using graphs with executors and edges.

External Integration – Easily connect external APIs or include human-in-the-loop scenarios.

Checkpointing – Save workflow states and resume long-running processes. [Hydration & Dehydration]

Multi-Agent Orchestration – Coordinate agents in sequential or parallel execution.

Composability – Nest or combine workflows for more advanced use cases.

In this tutorial, I’ll walk through building a web search workflow using the Microsoft Agent Framework integrated with the Tavily API.

This example demonstrates how to design workflows, manage data flow between executors, edges to connect each steps and produce clean, readable outputs all without relying on any AI model.

Overview of the Workflow

Our Sreeni Web Search Workflow follows a 3-step process:

1. Search the web using Tavily API.
2. Format the search results into readable content.
3. Print the final formatted results.

Workflow Design Principles

We’ll follow Microsoft Agent Framework best practices:

Separation of Concerns – Each executor handles one specific task.
Data Flow – Data moves smoothly between executors.
Error Handling – Every step handles errors gracefully.
Clean Output – Results are formatted and easy to read.

Technology Stack

Microsoft Agent Framework – Workflow orchestration

1. Tavily API – Web search functionality
2. Python – Implementation language
3. Environment Variables – Secure API key handling

Implementation

Step 1: Web Search Executor

@executor(id="sreeni_web_searchworkflow")
async def sreeni_web_search(input_data: str, ctx: WorkflowContext[str]) -> None:
    """
    First executor: Performs web search and passes raw results to next executor.
    """
    print(f"🚀 Step 1: Performing web search for: {input_data}")
    search_results = await perform_tavily_search(input_data)
    await ctx.send_message(str(search_results))

Responsibilities:

1. Performs a web search using Tavily API
2. Returns raw results
3. Passes data to the next executor

Step 2: Format Executor

@executor(id="sreeni_web_search_formatted_workflow")
async def sreeni_web_search_formatted(input_data: str, ctx: WorkflowContext[str]) -> None:
    """
    Second executor: Formats the search results and passes formatted data to next executor.
    """
    print(f"🎨 Step 2: Formatting search results...")

    import ast
    try:
        search_data = ast.literal_eval(input_data)
        formatted_results = format_search_results(search_data, "Python programming best practices")
    except:
        formatted_results = f"❌ Error parsing search results: {input_data}"

    await ctx.send_message(formatted_results)

Responsibilities:

1. Receives raw results from Step 1
2. Formats them into a readable structure
3. Passes the formatted output to the final step

Step 3: Print Executor

@executor(id="sreeni_web_search_formatted_print_workflow")
async def sreeni_web_search_formatted_print_workflow(input_data: str, ctx: WorkflowContext[Never,str]) -> None:
    """
    Third executor: Prints the formatted results to console and yields output.
    """
    print(f"🖨️  Step 3: Printing formatted results...")

    print("\n" + "="*80)
    print("📊 FINAL FORMATTED SEARCH RESULTS")
    print("="*80)
    print(input_data)
    print("="*80 + "\n")

    await ctx.yield_output(input_data)

Responsibilities:

1. Receives formatted data from Step 2
2. Prints the final, cleaned output
3. Yields the result back through the framework

Workflow Execution and Output

Complete Implementation

Sreeni Web Search Workflow - Dependencies

Microsoft Agent Framework

agent-framework>=1.0.0b251001

Tavily API for web search

tavily-python>=0.3.0

Environment variable management

python-dotenv>=1.0.0

Additional dependencies (if needed)

requests>=2.31.0

import asyncio 
import os
from dotenv import load_dotenv
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor
from typing_extensions import Never
from tavily import TavilyClient

# Load environment variables from .env file
load_dotenv()

async def perform_tavily_search(query: str) -> dict:
    """
    Perform web search using Tavily API and return raw results.
    """
    try:
        print(f"🔍 Searching with Tavily for: {query}")

        # Initialize Tavily client with API key from .env file
        api_key = os.getenv("TAVILY_API_KEY")
        if not api_key:
            return {"error": "TAVILY_API_KEY not found in environment variables"}

        client = TavilyClient(api_key=api_key)

        # Perform search using Tavily
        response = client.search(
            query=query,
            search_depth="basic",
            max_results=10
        )

        if not response.get('results'):
            return {"error": f"No search results found for: {query}"}

        return response

    except Exception as e:
        return {"error": f"Error performing Tavily search: {str(e)}"}

def format_search_results(search_data: dict, query: str) -> str:
    """
    Format search results into readable format.
    """
    if "error" in search_data:
        return f"❌ {search_data['error']}"

    results = search_data.get('results', [])
    if not results:
        return f"❌ No search results found for: {query}"

    # Format the results in a clean, readable format
    formatted_results = f"\n🔍 SEARCH RESULTS FOR: {query.upper()}\n"
    formatted_results += "=" * 80 + "\n\n"

    # Add clean results with better spacing (limit to 3 results)
    for i, result in enumerate(results[:3], 1):
        title = result.get('title', 'N/A')
        url = result.get('url', 'N/A')
        score = result.get('score', 0)
        content = result.get('content', 'N/A')

        # Clean up content - remove HTML tags, links, and excessive whitespace
        import re
        clean_content = re.sub(r'<[^>]+>', '', content)  # Remove HTML tags
        clean_content = re.sub(r'\[.*?\]\(.*?\)', '', clean_content)  # Remove markdown links
        clean_content = re.sub(r'https?://\S+', '', clean_content)  # Remove URLs
        clean_content = re.sub(r'[^\w\s.,!?-]', '', clean_content)  # Remove special chars
        clean_content = re.sub(r'\s+', ' ', clean_content).strip()  # Clean whitespace

        formatted_results += f"📌 RESULT #{i}\n"
        formatted_results += f"   Title: {title}\n"
        formatted_results += f"   URL: {url}\n"
        formatted_results += f"   Relevance: {score:.1%}\n"
        formatted_results += f"   Summary: {clean_content[:200]}...\n"
        formatted_results += "\n" + "-" * 60 + "\n\n"

    return formatted_results

@executor(id="sreeni_web_searchworkflow")
async def sreeni_web_search(input_data: str, ctx: WorkflowContext[str]) -> None:
    """
    First executor: Performs web search and passes raw results to next executor.
    """
    print(f"🚀 Step 1: Performing web search for: {input_data}")
    search_results = await perform_tavily_search(input_data)
    # Pass the search results to the next executor
    await ctx.send_message(str(search_results))

@executor(id="sreeni_web_search_formatted_workflow")
async def sreeni_web_search_formatted(input_data: str, ctx: WorkflowContext[str]) -> None:
    """
    Second executor: Formats the search results and passes formatted data to next executor.
    """
    print(f"🎨 Step 2: Formatting search results...")

    # Parse the search results from the previous executor
    import ast
    try:
        search_data = ast.literal_eval(input_data)
        formatted_results = format_search_results(search_data, "Python programming best practices")
    except:
        formatted_results = f"❌ Error parsing search results: {input_data}"

    # Pass the formatted results to the next executor
    await ctx.send_message(formatted_results)

@executor(id="sreeni_web_search_formatted_print_workflow")
async def sreeni_web_search_formatted_print_workflow(input_data: str, ctx: WorkflowContext[Never,str]) -> None:
    """
    Third executor: Prints the formatted results to console and yields output.
    """
    print(f"🖨️  Step 3: Printing formatted results...")

    # Print the formatted results to console
    print("\n" + "="*80)
    print("📊 FINAL FORMATTED SEARCH RESULTS")
    print("="*80)
    print(input_data)
    print("="*80 + "\n")

    # Yield the output through the framework
    await ctx.yield_output(input_data)

async def main():
    print("=" * 60)
    print("Sreeni Web Search Workflow")
    print("=" * 60)

    workflow = (
        WorkflowBuilder()
        .add_edge(sreeni_web_search, sreeni_web_search_formatted)
        .add_edge(sreeni_web_search_formatted, sreeni_web_search_formatted_print_workflow)
        .set_start_executor(sreeni_web_search)
        .build()
    )

    # Test the workflow with a sample search query
    test_query = "Seenivasa Ramadurai"
    print(f"Testing workflow with query: {test_query}")

    # Run the workflow
    await workflow.run(test_query)

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

Conclusion

This example shows how you can use the Microsoft Agent Framework to build functional and efficient workflows even without using any LLM or AI model.

By chaining simple, well-defined executors, we can automate repetitive data tasks, integrate APIs, and prepare the foundation for more advanced workflows that later include AI reasoning, conditional paths, or checkpointing.

In the next blog, we’ll take this a step further and build a Workflow + AI Agent combination that uses an LLM to reason and enhance results dynamically. Stay tuned!

Thanks
Sreeni Ramadorai