microsoft--ai-agents-for-beginners/11-agentic-protocols/code_samples/github-mcp/app.py

import os
import json
import logging
logging.getLogger("agent_framework.foundry").setLevel(logging.ERROR)
from typing import Annotated
from dotenv import load_dotenv
import requests
import re

import chainlit as cl
from mcp import ClientSession

from agent_framework import tool, AgentResponseUpdate, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from azure.core.credentials import AzureKeyCredential

from azure.search.documents import SearchClient
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import SearchIndex, SimpleField, SearchFieldDataType, SearchableField


# Load environment variables
load_dotenv()

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


# Initialize Azure AI Search with persistent storage
search_service_endpoint = os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT")
search_api_key = os.getenv("AZURE_SEARCH_API_KEY")
index_name = "event-descriptions"

search_client = SearchClient(
    endpoint=search_service_endpoint,
    index_name=index_name,
    credential=AzureKeyCredential(search_api_key)
)

index_client = SearchIndexClient(
    endpoint=search_service_endpoint,
    credential=AzureKeyCredential(search_api_key)
)

# Define the index schema
fields = [
    SimpleField(name="id", type=SearchFieldDataType.String, key=True),
    SearchableField(name="content", type=SearchFieldDataType.String)
]

index = SearchIndex(name=index_name, fields=fields)

# Check if index already exists if not, create it
try:
    existing_index = index_client.get_index(index_name)
    print(f"Index '{index_name}' already exists, using the existing index.")
except Exception as e:
    # Create the index if it doesn't exist
    print(f"Creating new index '{index_name}'...")
    index_client.create_index(index)

# Always read event descriptions from markdown file
current_dir = os.path.dirname(os.path.abspath(__file__))
event_descriptions_path = os.path.join(current_dir, "event-descriptions.md")

try:
    with open(event_descriptions_path, "r", encoding='utf-8') as f:
        markdown_content = f.read()
except FileNotFoundError:
    logger.warning(f"Could not find {event_descriptions_path}")
    markdown_content = ""

# Split the markdown content into individual event descriptions
event_descriptions = markdown_content.split("---")  # You can change the delimiter

# Create documents for Azure Search
documents = []
for i, description in enumerate(event_descriptions):
    description = description.strip()  # Remove leading/trailing whitespace
    if description:  # Avoid empty descriptions
        documents.append({"id": str(i + 1), "content": description})

# Add documents to the index (only if we have documents)
if documents:
    # Delete existing documents first to avoid duplicates
    try:
        search_client.delete_documents(documents=[{"id": doc["id"]} for doc in documents])
        print("Cleared existing documents")
    except Exception as e:
        print(f"Warning: Failed to clear existing documents: {str(e)}")
    
    # Upload new documents
    search_client.upload_documents(documents)
    print(f"Uploaded {len(documents)} documents to index")


# RAG tool for event search
@tool
def search_events(
    query: Annotated[str, "The search query to find relevant events"]
) -> str:
    """Searches for relevant events based on a query using Azure Search and a live API."""
    context_strings = []
    try:
        results = search_client.search(query, top=5)
        for result in results:
            if 'content' in result:
                context_strings.append(f"Event: {result['content']}")
    except Exception as e:
        context_strings.append(f"Error searching Azure Search: {str(e)}")
    # Live API (example: Devpost hackathons)
    try:
        api_resp = requests.get(f"https://devpost.com/api/hackathons?search={query}", timeout=5)
        if api_resp.ok:
            data = api_resp.json()
            for event in data.get('hackathons', [])[:5]:
                context_strings.append(f"Live Event: {event.get('title')} - {event.get('url')}")
    except Exception as e:
        context_strings.append(f"Error fetching live events: {str(e)}")
    if context_strings:
        return "\n\n".join(context_strings)
    else:
        return "No relevant events found."


def flatten(xss):
    return [x for xs in xss for x in xs]


GITHUB_INSTRUCTIONS = """
You are an expert on GitHub repositories. When answering questions, you **must** use the provided GitHub username to find specific information about that user's repositories, including:

*   Who created the repositories
*   The programming languages used
*   Information found in files and README.md files within those repositories
*   Provide links to each repository referenfced in your answers

**Important:** Never perform general searches for repositories. Always use the given GitHub username to find the relevant information. If a GitHub username is not provided, state that you need a username to proceed.
"""

HACKATHON_AGENT = """
You are an AI Agent Hackathon Strategist specializing in recommending winning project ideas.

Your task:
1. Analyze the GitHub activity of users to understand their technical skills
2. Suggest creative AI Agent projects tailored to their expertise. 
3. Focus on projects that align with Microsoft's AI Agent Hackathon prize categories

When making recommendations:
- Base your ideas strictly on the user's GitHub repositories, languages, and tools
- Give suggestions on tools, languages and frameworks to use to build it. 
- Provide detailed project descriptions including architecture and implementation approach
- Explain why the project has potential to win in specific prize categories
- Highlight technical feasibility given the user's demonstrated skills by referencing the specific repositories or languages used.

Formatting your response:
- Provide a clear and structured response that includes:
    - Suggested Project Name
    - Project Description 
    - Potential languages and tools to use
    - Link to each relevant GitHub repository you based your recommendation on

Hackathon prize categories:
- Best Overall Agent ($20,000)
- Best Agent in Python ($5,000)
- Best Agent in C# ($5,000)
- Best Agent in Java ($5,000)
- Best Agent in JavaScript/TypeScript ($5,000)
- Best Copilot Agent using Microsoft Copilot Studio or Microsoft 365 Agents SDK ($5,000)
- Best Microsoft Foundry Agent Service Usage ($5,000)
        
"""

EVENTS_AGENT = """
You are an Event Recommendation Agent specializing in suggesting relevant tech events.

Your task:
1. Review the project idea recommended by the Hackathon Agent
2. Use the search_events function to find relevant events based on the technologies mentioned.
3. NEVER suggest and event that the where there is not a relevant technology that the user has used.
3. ONLY recommend events that were returned by the search_events functionf

When making recommendations:
- IMPORTANT: You must first call the search_events function with appropriate technology keywords from the project
- Only recommend events that were explicitly returned by the search_events function
- Do not make up or suggest events that weren't in the search results
- Construct search queries using specific technologies mentioned (e.g., "Python AI workshop" or "JavaScript hackathon")
- Try multiple search queries if needed to find the most relevant events


For each recommended event:
- Only include events found in the search_events results
- Explain the direct connection between the event and the specific project requirements
- Highlight relevant workshops, sessions, or networking opportunities

Formatting your response:
- Start with "Based on the hackathon project idea, here are relevant events that I found:"
- Only list events that were returned by the search_events function
- For each event, include the exact event details as returned by search_events
- Explain specifically how each event relates to the project technologies

If no relevant events are found, acknowledge this and suggest trying different search terms instead of making up events.
"""


@cl.on_mcp_connect
async def on_mcp(connection, session: ClientSession):
    logger.info(f"MCP Connection established: {connection.name}")
    result = await session.list_tools()
    tools = [{
        "name": t.name,
        "description": t.description,
        "input_schema": t.inputSchema,
    } for t in result.tools]

    mcp_tools = cl.user_session.get("mcp_tools", {})
    mcp_tools[connection.name] = tools
    cl.user_session.set("mcp_tools", mcp_tools)
    
    # Log available tools
    print(f"Available MCP tools for {connection.name}:")
    for t in tools:
        print(f"  - {t['name']}: {t['description']}")

@cl.step(type="tool")
async def call_tool(tool_use):
    tool_name = tool_use.name
    tool_input = tool_use.input

    current_step = cl.context.current_step
    current_step.name = tool_name

    # Identify which mcp is used
    mcp_tools = cl.user_session.get("mcp_tools", {})
    mcp_name = None

    for connection_name, tools in mcp_tools.items():
        if any(t.get("name") == tool_name for t in tools):
            mcp_name = connection_name
            break

    if not mcp_name:
        current_step.output = json.dumps(
            {"error": f"Tool {tool_name} not found in any MCP connection"})
        return current_step.output

    mcp_session, _ = cl.context.session.mcp_sessions.get(mcp_name)

    if not mcp_session:
        current_step.output = json.dumps(
            {"error": f"MCP {mcp_name} not found in any MCP connection"})
        return current_step.output

    try:
        current_step.output = await mcp_session.call_tool(tool_name, tool_input)
    except Exception as e:
        current_step.output = json.dumps({"error": str(e)})

    return current_step.output


@cl.on_chat_start
async def on_chat_start():

    # Create the Microsoft Foundry Agent Service provider
    provider = FoundryChatClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        credential=AzureCliCredential(),
    )

    # Create agents using MAF
    github_agent = provider.as_agent(
        name="GithubAgent",
        instructions=GITHUB_INSTRUCTIONS,
    )

    hackathon_agent = provider.as_agent(
        name="HackathonAgent",
        instructions=HACKATHON_AGENT,
    )

    events_agent = provider.as_agent(
        name="EventsAgent",
        instructions=EVENTS_AGENT,
    )

    # Build a sequential workflow: GitHub → Hackathon → Events
    workflow = WorkflowBuilder(start_executor=github_agent) \
        .add_edge(github_agent, hackathon_agent) \
        .add_edge(hackathon_agent, events_agent) \
        .build()

    # Store in user session
    cl.user_session.set("provider", provider)
    cl.user_session.set("github_agent", github_agent)
    cl.user_session.set("hackathon_agent", hackathon_agent)
    cl.user_session.set("events_agent", events_agent)
    cl.user_session.set("workflow", workflow)
    cl.user_session.set("mcp_tools", {})
    cl.user_session.set("conversation_history", [])


# Add a cleanup handler for when the session ends
@cl.on_chat_end
async def on_chat_end():
    pass


def route_user_input(user_input: str):
    """
    Analyze user input and return a list of agent names to invoke.
    Returns: list of agent names (e.g., ["GitHubAgent", "HackathonAgent", "EventsAgent"])
    """
    user_input_lower = user_input.lower()
    agents = []
    # Example patterns (expand as needed)
    if re.search(r"github|repo|repository|commit|pull request", user_input_lower):
        agents.append("GitHubAgent")
    if re.search(r"hackathon|project idea|competition|challenge|win", user_input_lower):
        agents.append("HackathonAgent")
    if re.search(r"event|conference|meetup|workshop|webinar", user_input_lower):
        agents.append("EventsAgent")
    if not agents:
        agents = ["GitHubAgent", "HackathonAgent", "EventsAgent"]
    return agents


@cl.on_message
async def on_message(message: cl.Message):
    workflow = cl.user_session.get("workflow")
    github_agent = cl.user_session.get("github_agent")
    hackathon_agent = cl.user_session.get("hackathon_agent")
    events_agent = cl.user_session.get("events_agent")
    conversation_history = cl.user_session.get("conversation_history", [])

    user_input = message.content
    agent_names = route_user_input(user_input)
    conversation_history.append({"role": "user", "content": user_input})

    # If more than one agent is selected, use the workflow
    if len(agent_names) > 1:
        answer = cl.Message(content="Processing your request using: {}...\n\n".format(", ".join(agent_names)))
        await answer.send()
        agent_responses = []
        try:
            events = workflow.run(user_input, stream=True, tools=[search_events])
            last_author = None
            async for event in events:
                if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
                    update = event.data
                    author = update.author_name or "Agent"
                    if author != last_author:
                        if last_author is not None:
                            await answer.stream_token("\n\n")
                        await answer.stream_token(f"**{author}**: ")
                        last_author = author
                    if update.text:
                        await answer.stream_token(update.text)
                        agent_responses.append(f"**{author}**: {update.text}")
            full_response = "".join(agent_responses) if agent_responses else answer.content
            conversation_history.append({"role": "assistant", "content": full_response})
            cl.user_session.set("conversation_history", conversation_history)
            answer.content = full_response
            await answer.update()
        except Exception as e:
            await answer.stream_token(f"\n\n❌ Error: {str(e)}\n\n")
            conversation_history.append({"role": "assistant", "content": f"Error: {str(e)}"})
            cl.user_session.set("conversation_history", conversation_history)
            answer.content += f"\n\n❌ Error: {str(e)}"
            await answer.update()
    else:
        # Single agent: route to the appropriate agent
        agent_name = agent_names[0]
        agent_map = {
            "GitHubAgent": github_agent,
            "HackathonAgent": hackathon_agent,
            "EventsAgent": events_agent,
        }
        agent = agent_map.get(agent_name, github_agent)

        answer = cl.Message(content=f"Processing your request using {agent_name}...\n\n")
        await answer.send()
        try:
            tools_for_agent = [search_events] if agent_name == "EventsAgent" else []
            response = await agent.run(user_input, tools=tools_for_agent)
            answer.content = str(response)
            conversation_history.append({"role": "assistant", "content": answer.content})
            cl.user_session.set("conversation_history", conversation_history)
            await answer.update()
        except Exception as e:
            await answer.stream_token(f"\n\n❌ Error: {str(e)}\n\n")
            conversation_history.append({"role": "assistant", "content": f"Error: {str(e)}"})
            cl.user_session.set("conversation_history", conversation_history)
            answer.content += f"\n\n❌ Error: {str(e)}"
            await answer.update()