### Copy Environment File (Bash/PowerShell)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Copies the example environment file (.env.example) to a new file named .env. This .env file will store sensitive credentials and configuration settings.

```powershell
Copy-Item -Path .env.example -Destination .env
```

```bash
cp .env.example .env
```

--------------------------------

### Install Project Dependencies (Bash)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Installs all the Python packages required by the project, as listed in the 'requirements.txt' file. This command should be run after activating the virtual environment.

```bash
pip install -r requirements.txt
```

--------------------------------

### Clone Repository (Bash)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Clones the project repository from GitHub to your local machine. This is the first step in setting up the project.

```bash
git clone https://github.com/Azure-Samples/azure-ai-agent-service-enterprise-demo.git
```

--------------------------------

### Run Jupyter Notebook (VS Code)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Instructions to open and run the 'enterprise-streaming-agent.ipynb' Jupyter Notebook within VS Code. This notebook orchestrates the agent's setup, tool integration, and launches the demo UI.

```text
Open the 'enterprise-streaming-agent.ipynb' in VS Code.
Step through the cells to connect to Azure AI Foundry, configure tools, and launch the Gradio UI.
```

--------------------------------

### Configure .env File Variables (Plaintext)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Example configuration variables to be set in the .env file. These include Azure AI Foundry connection string, model name, Bing search connection, and OpenWeather API keys.

```plaintext
PROJECT_CONNECTION_STRING="<HostName>;<AzureSubscriptionId>;<ResourceGroup>;<ProjectName>"
MODEL_NAME="YOUR_MODEL_NAME"
BING_CONNECTION_NAME="YOUR_CONNECTION_NAME"
OPENWEATHER_GEO_API_KEY="YOUR_OPENWEATHER_GEOCODING_API_KEY"
OPENWEATHER_ONE_API_KEY="YOUR_OPENWEATHER_ONE_CALL_API_KEY"
```

--------------------------------

### Copy Environment File - Bash

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/infra/azure-deployment/README.md

Copies the example environment file to a new file, which will then be edited with specific deployment values. This is a standard practice for managing configuration in different environments.

```bash
cp .env.example .env
```

--------------------------------

### Create Python Virtual Environment (Bash)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Creates a Python virtual environment named '.venv' for isolating project dependencies. This ensures that project packages do not conflict with system-wide Python packages.

```bash
python -m venv .venv
```

--------------------------------

### Run Deployment Script - Bash

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/infra/azure-deployment/README.md

Executes the shell script responsible for creating Azure resources and deploying the web application. This script automates the infrastructure setup and application deployment process.

```bash
cd azure-deployment
./deploy.sh
```

--------------------------------

### Activate Python Virtual Environment (Bash/PowerShell)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Activates the Python virtual environment created in the previous step. Activation modifies the shell's PATH to prioritize the virtual environment's Python executable and packages.

```bash
source .venv/bin/activate
```

```powershell
.venv\Scripts\activate
```

--------------------------------

### Deploy Logic App using Azure CLI

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Deploys a logic app from a JSON template using the Azure CLI. It requires a resource group, the template file name, and a logic app name parameter. This enables the logic app to trigger on an HTTP request.

```bash
az deployment group create \
  --resource-group <your-resource-group> \
  --template-file send_email_logic_app.template.json \
  --parameters logicAppName=send_email_logic_app
```

--------------------------------

### VS Code: Select Python Interpreter (Command Palette)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Instructions for selecting the project's virtual environment interpreter within Visual Studio Code. This ensures that VS Code uses the correct Python environment for running and debugging the project.

```text
Press Ctrl+Shift+P (or Cmd+Shift+P on Mac)
Choose Python: Select Interpreter and select the .venv environment.
```

--------------------------------

### Configure Direct Azure AI Search Integration in .env

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Sets environment variables for direct integration with an existing Azure AI Search index. This allows for custom control over the search index used by the agent.

```dotenv
#AZURE_SEARCH_CONNECTION_NAME="YOUR_AZURE_SEARCH_CONNECTION_NAME"
#AZURE_SEARCH_INDEX_NAME="YOUR_AZURE_SEARCH_INDEX_NAME"
```

--------------------------------

### Configure Logic App Send Email URL in .env

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/README.md

Sets the endpoint URL for the deployed logic app in the environment file. This URL is used by the agent to send emails via the logic app.

```dotenv
LOGIC_APP_SEND_EMAIL_URL="https://<your-logic-app-endpoint>"
```

--------------------------------

### Build Gradio UI for Azure AI Agent in Python

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

Constructs a Gradio interface for the Azure AI Agent Service. It features a chat interface, a text input for user queries, and predefined examples. The UI is styled with a custom theme and includes functions to manage the conversation thread and clear the chat.

```python
brand_theme = gr.themes.Default(
    primary_hue="blue",
    secondary_hue="blue",
    neutral_hue="gray",
    font=["Segoe UI", "Arial", "sans-serif"],
    font_mono=["Courier New", "monospace"],
    text_size="lg",
).set(
    button_primary_background_fill="#0f6cbd",
    button_primary_background_fill_hover="#115ea3",
    button_primary_background_fill_hover_dark="#4f52b2",
    button_primary_background_fill_dark="#5b5fc7",
    button_primary_text_color="#ffffff",
    button_secondary_background_fill="#e0e0e0",
    button_secondary_background_fill_hover="#c0c0c0",
    button_secondary_background_fill_hover_dark="#a0a0a0",
    button_secondary_text_color="#000000",
    body_background_fill="#f5f5f5",
    block_background_fill="#ffffff",
    body_text_color="#242424",
    body_text_color_subdued="#616161",
    block_border_color="#d1d1d1",
    block_border_color_dark="#333333",
    input_background_fill="#ffffff",
    input_border_color="#d1d1d1",
    input_border_color_focus="#0f6cbd",
)

with gr.Blocks(theme=brand_theme, css="footer {visibility: hidden;}", fill_height=True) as demo:

    def clear_thread():
        global thread
        thread = project_client.agents.create_thread()
        return []

    def on_example_clicked(evt: gr.SelectData):
        return evt.value["text"]  # Fill the textbox with that example text

    gr.HTML("<h1 style=\"text-align: center;\">Azure AI Agent Service</h1>")

    chatbot = gr.Chatbot(
        type="messages",
        examples=[
            {"text": "What's my company's remote work policy?"},
            {"text": "Check if it will rain tomorrow?"},
            {"text": "How is Contoso's stock doing today?"},
            {"text": "Send my direct report a summary of the HR policy."},
        ],
        show_label=False,
        scale=1,
    )

    textbox = gr.Textbox(
        show_label=False,
        lines=1,
        submit_btn=True,
    )

    # Populate textbox when an example is clicked
    chatbot.example_select(fn=on_example_clicked, inputs=None, outputs=textbox)

    # On submit: call azure_enterprise_chat, then clear the textbox
    (textbox
     .submit(
         fn=azure_enterprise_chat,
         inputs=[textbox, chatbot],
         outputs=[chatbot, textbox],
     )
     .then(
         fn=lambda: "",
         outputs=textbox,
     )
    )

    # A "Clear" button that resets the thread and the Chatbot
    chatbot.clear(fn=clear_thread, outputs=chatbot)

# Launch your Gradio app
if __name__ == "__main__":
    demo.launch()
```

--------------------------------

### Set Up BingGroundingTool and FileSearchTool

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

Configures and sets up essential tools for the AI agent, specifically `BingGroundingTool` for search grounding and `FileSearchTool` for document retrieval. It handles checking for existing connections and vector stores, and creates them if they don't exist. This involves uploading local documents and creating a vector store for efficient searching.

```python
try:
    bing_connection = project_client.connections.get(connection_name=os.environ["BING_CONNECTION_NAME"])
    conn_id = bing_connection.id
    bing_tool = BingGroundingTool(connection_id=conn_id)
    print("bing > connected")
except Exception:
    bing_tool = None
    print("bing failed > no connection found or permission issue")

FOLDER_NAME = "enterprise-data"
VECTOR_STORE_NAME = "hr-policy-vector-store"
all_vector_stores = project_client.agents.list_vector_stores().data
existing_vector_store = next(
    (store for store in all_vector_stores if store.name == VECTOR_STORE_NAME),
    None
)

vector_store_id = None
if existing_vector_store:
    vector_store_id = existing_vector_store.id
    print(f"reusing vector store > {existing_vector_store.name} (id: {existing_vector_store.id})")
else:
    # If you have local docs to upload
    import os
    if os.path.isdir(FOLDER_NAME):
        file_ids = []
        for file_name in os.listdir(FOLDER_NAME):
            file_path = os.path.join(FOLDER_NAME, file_name)
            if os.path.isfile(file_path):
                print(f"uploading > {file_name}")
                uploaded_file = project_client.agents.upload_file_and_poll(
                    file_path=file_path,
                    purpose=FilePurpose.AGENTS
                )
                file_ids.append(uploaded_file.id)

        if file_ids:
            print(f"creating vector store > from {len(file_ids)} files.")
            vector_store = project_client.agents.create_vector_store_and_poll(
                file_ids=file_ids,
                name=VECTOR_STORE_NAME
            )
            vector_store_id = vector_store.id
            print(f"created > {vector_store.name} (id: {vector_store_id})")

file_search_tool = None
if vector_store_id:
    file_search_tool = FileSearchTool(vector_store_ids=[vector_store_id])
```

--------------------------------

### Create LoggingToolSet and Combine Tools

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code defines a LoggingToolSet that extends ToolSet to log function call inputs and outputs. It then creates a ToolSet, adds various tools (Bing, File Search, custom functions), and iterates through the added tools to print their names and descriptions.

```python
class LoggingToolSet(ToolSet):
    def execute_tool_calls(self, tool_calls: List[Any]) -> List[dict]:
        """
        Execute the upstream calls, printing only two lines per function:
        1) The function name + its input arguments
        2) The function name + its output result
        """

        # For each function call, print the input arguments
        for c in tool_calls:
            if hasattr(c, "function") and c.function:
                fn_name = c.function.name
                fn_args = c.function.arguments
                print(f"{fn_name} inputs > {fn_args} (id:{c.id})")

        # Execute the tool calls (superclass logic)
        raw_outputs = super().execute_tool_calls(tool_calls)

        # Print the output of each function call
        for item in raw_outputs:
            print(f"output > {item['output']}")

        return raw_outputs

custom_functions = FunctionTool(enterprise_fns)

toolset = LoggingToolSet()
if bing_tool:
    toolset.add(bing_tool)
if file_search_tool:
    toolset.add(file_search_tool)
toolset.add(custom_functions)

for tool in toolset._tools:
    tool_name = tool.__class__.__name__
    print(f"tool > {tool_name}")
    for definition in tool.definitions:
        if hasattr(definition, "function"):
            fn = definition.function
            print(f"{fn.name} > {fn.description}")
        else:
            pass
```

--------------------------------

### Import Libraries for Azure AI Agent Service Demo

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

Imports all necessary Python libraries and modules for the Azure AI Agent Service enterprise demo. This includes Azure AI SDKs for client interaction, Gradio for UI development, and custom functions for enterprise-specific tasks. It also loads environment variables using dotenv.

```python
import os
import re
from datetime import datetime as pydatetime
from typing import Any, List, Dict
from dotenv import load_dotenv

# (Optional) Gradio app for UI
import gradio as gr
from gradio import ChatMessage

# Azure AI Projects
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
    AgentEventHandler,
    RunStep,
    RunStepDeltaChunk,
    ThreadMessage,
    ThreadRun,
    MessageDeltaChunk,
    BingGroundingTool,
    FilePurpose,
    FileSearchTool,
    FunctionTool,
    ToolSet
)

# Your custom Python functions (for "fetch_weather","fetch_stock_price","send_email","fetch_datetime", etc.)
from enterprise_functions import enterprise_fns

load_dotenv()
```

--------------------------------

### Create Azure AI Client and Load Project

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

Initializes the Azure AI client using `DefaultAzureCredential` for authentication and connects to the Azure AI service via a project connection string loaded from environment variables. This client is essential for interacting with Azure AI resources.

```python
credential = DefaultAzureCredential()
project_client = AIProjectClient.from_connection_string(
    credential=credential,
    conn_str=os.environ["PROJECT_CONNECTION_STRING"]
)
```

--------------------------------

### Integrate Azure AI Search Tool

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code snippet conditionally adds an Azure AI Search tool to the existing toolset. It first checks if a FileSearchTool is already present. If not, it attempts to connect to Azure AI Search using environment variables for connection and index names, then adds the search tool to the toolset. Errors during connection are caught and reported.

```python
# (Optional) Direct Azure AI Search Integration
# If the default File Search Tool is available, we add it and skip the Azure AI Search integration.
if any(tool.__class__.__name__ == "FileSearchTool" for tool in toolset._tools):
    print("file_search tool exists > skipping ai_search tool add")
else:
    try:
        # Get the connection ID for your Azure AI Search resource
        connections = project_client.connections.list()
        conn_id = next(
            c.id for c in connections if c.name == os.environ.get("AZURE_SEARCH_CONNECTION_NAME")
        )
        
        # Initialize Azure AI Search tool for direct index access
        from azure.ai.projects.models import AzureAISearchTool
        search_tool = AzureAISearchTool(
            index_connection_id=conn_id,
            index_name=os.environ.get("AZURE_SEARCH_INDEX_NAME")
        )
        
        # Add the Azure AI Search tool to our toolset
        toolset.add(search_tool)
        print("azure ai search > connected directly to index")
        
        # Verify the tool was added by iterating through the toolset
        for tool in toolset._tools:
            tool_name = tool.__class__.__name__
            print(f"tool > {tool_name}")
            for definition in tool.definitions:
                if hasattr(definition, "function"):
                    fn = definition.function
                    print(f"{fn.name} > {fn.description}")
                else:
                    pass

    except Exception as e:
        print(f"azure ai search > skipped (no connection configured): {str(e)}")
```

--------------------------------

### Verify Web App Deployment URL - Text

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/infra/azure-deployment/README.md

Provides the base URL format to access the deployed web application. Replace the placeholder with the actual web app name configured during deployment.

```text
https://<YOUR_WEB_APP_NAME>.azurewebsites.net
```

--------------------------------

### Manage Run Steps and Tool Calls in Azure AI Agent Service Event Stream (Python)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code handles 'run_step' events from the Azure AI Agent Service stream. It manages tool calls in different states (in_progress, completed) and message creation steps. It clears relevant tracking dictionaries and appends new assistant messages when needed. Dependencies include 'ChatMessage', 'in_progress_tools', 'partial_calls_by_id', 'partial_calls_by_index', 'call_id_for_index', and 'finalize_tool_call'.

```python
            elif event_type == "run_step":
                step_type = event_data["type"]
                step_status = event_data["status"]

                if step_type == "tool_calls" and step_status == "in_progress":
                    for tcall in event_data["step_details"].get("tool_calls", []):
                        upsert_tool_call(tcall)
                    yield conversation, ""

                elif step_type == "tool_calls" and step_status == "completed":
                    for cid, msg_obj in in_progress_tools.items():
                        msg_obj.metadata["status"] = "done"
                    in_progress_tools.clear()
                    partial_calls_by_id.clear()
                    partial_calls_by_index.clear()
                    call_id_for_index.clear()
                    yield conversation, ""

                elif step_type == "message_creation" and step_status == "in_progress":
                    msg_id = event_data["step_details"].get("message_creation", {}).get("message_id")
                    if msg_id:
                        conversation.append(ChatMessage(role="assistant", content=""))
                    yield conversation, ""

                elif step_type == "message_creation" and step_status == "completed":
                    yield conversation, ""
```

--------------------------------

### Environment Variables for Azure Deployment - Env

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/infra/azure-deployment/README.md

Defines essential environment variables required for deploying the Azure AI Agent Service Enterprise Demo web app. These include connection strings, resource names, and API keys for various Azure services and external APIs.

```env
PROJECT_CONNECTION_STRING="<HostName>;<AzureSubscriptionId>;<ResourceGroup>;<ProjectName>"
RESOURCE_GROUP="YOUR_RESOURCE_GROUP_NAME"
APP_SERVICE_PLAN="YOUR_APP_SERVICE_PLAN_NAME"
WEB_APP_NAME="YOUR_WEB_APP_NAME"
LOCATION="YOUR_APPSERVICEPLAN_LOCATION"
AGENT_NAME="YOUR_AGENT_NAME"
BING_CONNECTION_NAME="YOUR_CONNECTION_NAME"
VECTOR_STORE_NAME="YOUR_VECTOR_STORE_NAME"
OPENWEATHER_ONE_API_KEY="YOUR_OPENWEATHER_ONE_CALL_API_KEY"
OPENWEATHER_GEO_API_KEY="YOUR_OPENWEATHER_GEOCODING_API_KEY"
```

--------------------------------

### Initialize and Append Tool Call Messages in Azure AI Agent Service (Python)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code initializes and appends chat messages for different tool types (tool, file_search) within an Azure AI Agent Service context. It creates 'ChatMessage' objects with specific roles, content, and metadata, including status and an ID. It also manages 'in_progress_tools' for tracking ongoing tool operations. Dependencies include 'ChatMessage', 'get_function_title', 'conversation', and 'in_progress_tools'.

```python
            msg_obj = ChatMessage(
                role="assistant",
                content="searching docs...",
                metadata={
                    "title": get_function_title("file_search"),
                    "status": "pending",
                    "id": f"tool-{call_id}" if call_id else "tool-noid"
                }
            )
            conversation.append(msg_obj)
            if call_id:
                in_progress_tools[call_id] = msg_obj
            return

        elif t_type != "function":
            return
```

--------------------------------

### Rebase and Force Push for PR Updates

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/CONTRIBUTING.md

This shell command sequence is used to update a pull request after suggested changes. It rebases the local branch onto the master branch interactively and then force pushes the changes to the GitHub repository, ensuring the PR reflects the latest updates.

```shell
git rebase master -i
git push -f
```

--------------------------------

### Create or Reuse Enterprise Agent

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code defines an enterprise agent, either creating a new one or updating an existing one. It retrieves a list of agents, checks if an agent with the name 'my-enterprise-agent' exists, and then either updates the found agent with the current toolset and model or creates a new agent with specified model, name, instructions, and toolset. The current date and time are dynamically included in the agent's instructions.

```python
AGENT_NAME = "my-enterprise-agent"
found_agent = None
all_agents_list = project_client.agents.list_agents().data
for a in all_agents_list:
    if a.name == AGENT_NAME:
        found_agent = a
        break

model_name = os.environ.get("MODEL_DEPLOYMENT_NAME", "gpt-4o")
instructions = (
    "You are a helpful enterprise assistant at Contoso. "
    f"Today's date is {pydatetime.now().strftime('%A, %b %d, %Y, %I:%M %p')}. "
    "You have access to hr documents in file_search, the grounding engine from bing and custom python functions like fetch_weather, "
    "fetch_stock_price, send_email, etc. Provide well-structured, concise, and professional answers."
)

if found_agent:
    # Update the existing agent to use new tools
    agent = project_client.agents.update_agent(
        assistant_id=found_agent.id,
        model=found_agent.model,
        instructions=found_agent.instructions,
        toolset=toolset,
    )
    print(f"reusing agent > {agent.name} (id: {agent.id})")
else:
    agent = project_client.agents.create_agent(
        model=model_name,
        name=AGENT_NAME,
        instructions=instructions,
        toolset=toolset
    )
    print(f"creating agent > {agent.name} (id: {agent.id})")
```

--------------------------------

### Implement Main Chat Functions in Python

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code defines the main chat functions for processing user messages and tool interactions. It manages conversation history using agent threads and streams partial responses. Key functions include extracting Bing search queries, converting dictionary-based messages to ChatMessage objects, and handling the overall chat flow with tool call management.

```python
import re
from typing import List, Dict

# Assuming ChatMessage and project_client, thread are defined elsewhere
# For example purposes, let's define a mock ChatMessage class
class ChatMessage:
    def __init__(self, role: str, content: str, metadata: dict = None):
        self.role = role
        self.content = content
        self.metadata = metadata

def extract_bing_query(request_url: str) -> str:
    """
    Extract the query string from something like:
      https://api.bing.microsoft.com/v7.0/search?q="latest news about Microsoft January 2025"
    Returns: latest news about Microsoft January 2025
    """
    match = re.search(r'q="([^"]+)"', request_url)
    if match:
        return match.group(1)
    # If no match, fall back to entire request_url
    return request_url

def convert_dict_to_chatmessage(msg: dict) -> ChatMessage:
    """
    Convert a legacy dict-based message to a gr.ChatMessage.
    Uses the 'metadata' sub-dict if present.
    """
    return ChatMessage(
        role=msg["role"],
        content=msg["content"],
        metadata=msg.get("metadata", None)
    )

def azure_enterprise_chat(user_message: str, history: List[dict]):
    """
    Accumulates partial function arguments into ChatMessage['content'], sets the
    corresponding tool bubble status from "pending" to "done" on completion,
    and also handles non-function calls like bing_grounding or file_search by appending a
    "pending" bubble. Then it moves them to "done" once tool calls complete.

    This function returns a list of ChatMessage objects directly (no dict conversion).
    Your Gradio Chatbot should be type="messages" to handle them properly.
    """

    # Convert existing history from dict to ChatMessage
    conversation = []
    for msg_dict in history:
        conversation.append(convert_dict_to_chatmessage(msg_dict))

    # Append the user's new message
    conversation.append(ChatMessage(role="user", content=user_message))

    # Immediately yield two outputs to clear the textbox
    # This part assumes a generator function and yields partial results
    # For a direct return, this yield would be replaced by a return statement
    yield conversation, ""

    # Post user message to the thread (for your back-end logic)
    # Assuming project_client and thread are accessible in this scope
    # project_client.agents.create_message(
    #     thread_id=thread.id,
    #     role="user",
    #     content=user_message
    # )

    # Mappings for partial function calls
    call_id_for_index: Dict[int, str] = {}
    partial_calls_by_index: Dict[int, dict] = {}
    partial_calls_by_id: Dict[str, dict] = {}
    in_progress_tools: Dict[str, ChatMessage] = {}

    # Titles for tool bubbles
    function_titles = {
        "fetch_weather": "☁️ fetching weather",
        "fetch_datetime": "🕒 fetching datetime",
        "fetch_stock_price": "📈 fetching financial info",
        "send_email": "✉️ sending mail",
        "file_search": "📄 searching docs",
        "bing_grounding": "🔍 searching bing",
    }

    def get_function_title(fn_name: str) -> str:
        return function_titles.get(fn_name, f"🛠 calling {fn_name}")

    def accumulate_args(storage: dict, name_chunk: str, arg_chunk: str):
        """Accumulates partial JSON data for a function call."""
        if name_chunk:
            storage["name"] += name_chunk
        if arg_chunk:
            storage["args"] += arg_chunk

    def finalize_tool_call(call_id: str):
        """Creates or updates the ChatMessage bubble for a function call."""
        if call_id not in partial_calls_by_id:
            return
        data = partial_calls_by_id[call_id]
        fn_name = data["name"].strip()
        fn_args = data["args"].strip()
        if not fn_name:
            return

        if call_id not in in_progress_tools:
            # Create a new bubble with status="pending"
            msg_obj = ChatMessage(
                role="assistant",
                content=fn_args or "",
                metadata={
                    "title": get_function_title(fn_name),
                    "status": "pending",
                    "id": f"tool-{call_id}"
                }
            )
            conversation.append(msg_obj)
            in_progress_tools[call_id] = msg_obj
        else:
            # Update existing bubble
            msg_obj = in_progress_tools[call_id]
            msg_obj.content = fn_args or ""
            msg_obj.metadata["title"] = get_function_title(fn_name)

    def upsert_tool_call(tcall: dict):
        """
        1) Check the call type
        2) If "function", gather partial name/args
        3) If "bing_grounding" or "file_search", show a pending bubble
        """
        t_type = tcall.get("type", "")
        call_id = tcall.get("id")

        # --- BING GROUNDING ---
        if t_type == "bing_grounding":
            request_url = tcall.get("bing_grounding", {}).get("requesturl", "")
            if not request_url.strip():
                return

            query_str = extract_bing_query(request_url)
            if not query_str.strip():
                return

            msg_obj = ChatMessage(
                role="assistant",
                content=query_str,
                metadata={
                    "title": get_function_title("bing_grounding"),
                    "status": "pending",
                    "id": f"tool-{call_id}"
                }
            )
            # Add or update the message in the conversation history
            # This simplified example assumes appending; a real implementation might search and update
            conversation.append(msg_obj)
            in_progress_tools[call_id] = msg_obj
        # --- FILE SEARCH ---
        elif t_type == "file_search":
            query_str = tcall.get("file_search", {}).get("query", "")
            if not query_str.strip():
                return

            msg_obj = ChatMessage(
                role="assistant",
                content=query_str,
                metadata={
                    "title": get_function_title("file_search"),
                    "status": "pending",
                    "id": f"tool-{call_id}"
                }
            )
            conversation.append(msg_obj)
            in_progress_tools[call_id] = msg_obj
        # --- REGULAR FUNCTION CALL ---
        elif t_type == "function":
            # Logic to handle function calls, accumulating name and args
            # This would involve checking for partial calls and updating them
            pass # Placeholder for function call accumulation logic

    # --- Main loop for processing agent responses (simulated) ---
    # In a real scenario, this would be a loop processing agent output stream
    # For demonstration, let's assume a single tool call is processed

    # Example of a simulated tool call response from the agent:
    simulated_tool_call = {
        "type": "bing_grounding",
        "id": "call-123",
        "bing_grounding": {
            "requesturl": "https://api.bing.microsoft.com/v7.0/search?q=\"latest AI news\""
        }
    }

    upsert_tool_call(simulated_tool_call)

    # Simulate agent processing and updating the tool status
    # In a real app, this would be driven by actual agent responses
    # For example, after the tool call is 'done':
    if simulated_tool_call["id"] in in_progress_tools:
        in_progress_tools[simulated_tool_call["id"]].metadata["status"] = "done"
        # Optionally update content with results if available
        in_progress_tools[simulated_tool_call["id"]].content += " (completed)"

    # Yield the final conversation state
    yield conversation, ""

```

--------------------------------

### Accumulate Partial Function Arguments in Azure AI Agent Service (Python)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code segment handles the accumulation of partial function call arguments. It manages a dictionary 'partial_calls_by_id' to store and append argument chunks for ongoing function calls, ensuring that even fragmented arguments are correctly assembled. It uses an 'accumulate_args' helper function and updates tool call information. Dependencies include 'partial_calls_by_id', 'partial_calls_by_index', 'call_id_for_index', and 'finalize_tool_call'.

```python
        if call_id not in partial_calls_by_id:
            partial_calls_by_id[call_id] = {"name": "", "args": ""}

        if index in partial_calls_by_index:
            old_data = partial_calls_by_index.pop(index)
            partial_calls_by_id[call_id]["name"] += old_data.get("name", "")
            partial_calls_by_id[call_id]["args"] += old_data.get("args", "")

        accumulate_args(partial_calls_by_id[call_id], name_chunk, arg_chunk)

        finalize_tool_call(call_id)
```

--------------------------------

### Handle Parallel Tool Call Errors with Azure SDK

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/KNOWN_ISSUES.md

This snippet demonstrates handling the HttpResponseError that occurs when attempting to add messages to a thread with an active run. It shows how to catch the specific Azure core exception and potentially implement retry logic or user feedback.

```python
from azure.core.exceptions import HttpResponseError

try:
    # Attempt to add a message to an active thread
    # thread.create_message(...) or similar operation
    pass
except HttpResponseError as e:
    if "Can't add messages to thread_" in str(e) and "while a run" in str(e):
        print(f"Error: Cannot add message. A run is currently active on the thread. Please wait or clear the conversation. Details: {e}")
        # Implement workaround: e.g., inform user, clear conversation, or retry after delay
    else:
        print(f"An unexpected Azure error occurred: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")
```

--------------------------------

### Create Conversation Thread using Python

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This snippet shows how to create a new conversation thread using the Azure AI Agent Service client. Threads are essential for managing and tracking ongoing conversations with the agent. It takes a project client object as input and returns a thread object.

```python
thread = project_client.agents.create_thread()
print(f"thread > created (id: {thread.id})")
```

--------------------------------

### Delete Azure Resources (Optional) in Python

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

Provides commented-out Python code for deleting Azure resources like agents, threads, and vector stores. It utilizes `DefaultAzureCredential` for authentication and `AIProjectClient` to interact with Azure AI projects. Ensure the `PROJECT_CONNECTION_STRING` environment variable is set.

```python
#from azure.identity import DefaultAzureCredential
#from azure.ai.projects import AIProjectClient
#import os
#
#credential = DefaultAzureCredential()
#project_client_delete = AIProjectClient.from_connection_string(
#    credential=credential,
#    conn_str=os.environ.get("PROJECT_CONNECTION_STRING")
#)
```

--------------------------------

### Handle Partial Tool Calls in Azure AI Agent Service Event Stream

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python function processes 'thread.run.step.delta' events to handle partial tool calls. It identifies tool call steps and updates the conversation with pending tool messages, managing tool call IDs for tracking. Dependencies include a 'ChatMessage' class and an 'upsert_tool_call' function.

```python
            if event_type == "thread.run.step.delta":
                step_delta = event_data.get("delta", {}).get("step_details", {})
                if step_delta.get("type") == "tool_calls":
                    for tcall in step_delta.get("tool_calls", []):
                        upsert_tool_call(tcall)
                    yield conversation, ""
```

--------------------------------

### Define Custom Event Handler for Agent Debugging in Python

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code defines a custom event handler class, `MyEventHandler`, that inherits from `AgentEventHandler`. It's designed to capture and display various events during the agent's operation, such as message deltas, thread messages, run statuses, and tool calls. This allows for detailed real-time debugging and logging. It handles events like `on_message_delta`, `on_thread_message`, `on_thread_run`, `on_run_step`, and `on_run_step_delta`.

```python
class MyEventHandler(AgentEventHandler):
    def __init__(self):
        super().__init__()
        self._current_message_id = None
        self._accumulated_text = ""

    def on_message_delta(self, delta: MessageDeltaChunk) -> None:
        # If a new message id, start fresh
        if delta.id != self._current_message_id:
            # First, if we had an old message that wasn't completed, finish that line
            if self._current_message_id is not None:
                print()  # move to a new line
            
            self._current_message_id = delta.id
            self._accumulated_text = ""
            print("\nassistant > ", end="")  # prefix for new message

        # Accumulate partial text
        partial_text = ""
        if delta.delta.content:
            for chunk in delta.delta.content:
                partial_text += chunk.text.get("value", "")
        self._accumulated_text += partial_text

        # Print partial text with no newline
        print(partial_text, end="", flush=True)

    def on_thread_message(self, message: ThreadMessage) -> None:
        # When the assistant's entire message is "completed", print a final newline
        if message.status == "completed" and message.role == "assistant":
            print()  # done with this line
            self._current_message_id = None
            self._accumulated_text = ""
        else:
            # For other roles or statuses, you can log if you like:
            print(f"{message.status.name.lower()} (id: {message.id})")

    def on_thread_run(self, run: ThreadRun) -> None:
        print(f"status > {run.status.name.lower()}")
        if run.status == "failed":
            print(f"error > {run.last_error}")

    def on_run_step(self, step: RunStep) -> None:
        print(f"{step.type.name.lower()} > {step.status.name.lower()}")

    def on_run_step_delta(self, delta: RunStepDeltaChunk) -> None:
        # If partial tool calls come in, we log them
        if delta.delta.step_details and delta.delta.step_details.tool_calls:
            for tcall in delta.delta.step_details.tool_calls:
                if getattr(tcall, "function", None):
                    if tcall.function.name is not None:
                        print(f"tool call > {tcall.function.name}")

    def on_unhandled_event(self, event_type: str, event_data):
        print(f"unhandled > {event_type} > {event_data}")

    def on_error(self, data: str) -> None:
        print(f"error > {data}")

    def on_done(self) -> None:
        print("done")
```

--------------------------------

### Delete Azure AI Agent Service Resources using Python SDK

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This snippet shows how to delete an agent, its associated thread, and a vector store using the Azure AI Agent Service Python SDK. It includes basic error handling to catch and report exceptions during the deletion process. Ensure the necessary client objects and resource IDs are correctly initialized before execution.

```python
# try:
#     project_client_delete.agents.delete_agent(agent.id)
#     print("Agent deletion successful.")
#     project_client_delete.agents.delete_thread(thread.id)
#     print("Thread deletion successful.")
#     project_client_delete.agents.delete_vector_store(vector_store_id)
#     print("Vector store deletion successful.")
#     print("All deletions succeeded.")
# except Exception as e:
#     print(f"Error during deletion: {e}")
```

--------------------------------

### Handle Streaming Chat Messages in Python

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

Processes incoming chat messages, appending streamed text to existing messages or creating new ones. It distinguishes between assistant messages, tool calls, and final message completion events. This code manages the conversation history and tool progress.

```python
if matching_msg:
                    # Append newly streamed text
                    matching_msg.content += agent_msg
                else:
                    # Append to last assistant or create new
                    if (
                        not conversation
                        or conversation[-1].role != "assistant"
                        or (
                            conversation[-1].metadata
                            and str(conversation[-1].metadata.get("id", "")).startswith("tool-")
                        )
                    ):
                        conversation.append(ChatMessage(role="assistant", content=agent_msg))
                    else:
                        conversation[-1].content += agent_msg

                yield conversation, ""

            # 4) If entire assistant message is completed
            elif event_type == "thread.message":
                if event_data["role"] == "assistant" and event_data["status"] == "completed":
                    for cid, msg_obj in in_progress_tools.items():
                        msg_obj.metadata["status"] = "done"
                    in_progress_tools.clear()
                    partial_calls_by_id.clear()
                    partial_calls_by_index.clear()
                    call_id_for_index.clear()
                    yield conversation, ""

            # 5) Final done
            elif event_type == "thread.message.completed":
                for cid, msg_obj in in_progress_tools.items():
                    msg_obj.metadata["status"] = "done"
                in_progress_tools.clear()
                partial_calls_by_id.clear()
                partial_calls_by_index.clear()
                call_id_for_index.clear()
                yield conversation, ""
                break

    return conversation, ""
```

--------------------------------

### Process Assistant Text Deltas in Azure AI Agent Service Event Stream (Python)

Source: https://github.com/azure-samples/azure-ai-agent-service-enterprise-demo/blob/main/enterprise-streaming-agent.ipynb

This Python code snippet processes 'thread.message.delta' events to accumulate text chunks from the assistant. It reconstructs the assistant's message content and updates the corresponding message object in the conversation history. It relies on a 'ChatMessage' class and searches the conversation history for a matching message ID and role.

```python
            elif event_type == "thread.message.delta":
                agent_msg = ""
                for chunk in event_data["delta"].get("content", []):
                    agent_msg += chunk["text"].get("value", "")

                message_id = event_data["id"]

                matching_msg = None
                for msg in reversed(conversation):
                    if msg.metadata and msg.metadata.get("id") == message_id and msg.role == "assistant":
                        matching_msg = msg
                        break

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.