Unlocking Analytics with MCP: Building a Prompt-Driven MCP Server with Azure and OpenAI

 In the age of AI and cloud analytics, organizations are seeking ways to empower users to ask questions in natural language and get actionable insights from their data — securely, flexibly, and at scale. Enter the Model Context Protocol (MCP): a pattern for building servers that translate user prompts into data queries, fetch results from cloud sources, and leverage AI models for further analysis.

In this article, I’ll walk you through a real-world example of building an MCP server in Python. This server connects to Azure Application Insights, dynamically generates KQL queries based on user prompts, fetches telemetry data, and sends it to Azure OpenAI for further processing. Along the way, we’ll cover prompt parsing, secure configuration, caching, and even a simple UI for end users.

What is a Model Context Protocol (MCP) Server?

An MCP server is a backend service that:

• Accepts natural language prompts from users (via API, UI, or other clients)
• Parses the prompt to extract intent, parameters, and context (e.g., user, time period)
• Dynamically generates and executes queries against data sources (e.g., Azure Application Insights)
• Optionally enriches or summarizes results using AI models (e.g., Azure OpenAI)
• Returns the final answer or insight to the user

This pattern enables “prompt-driven analytics” — users can simply ask, “Show me failed requests for user alice@example.com in the last 2 months,” and the server handles the rest.

Key Capabilities Demonstrated

• Natural Language Prompt Parsing: Extract user, time period, and intent from free-form prompts using regex and flexible logic.
• Dynamic Query Generation: Build KQL queries on the fly, supporting both static mappings and fully dynamic, prompt-driven analytics.
• Secure Cloud Integration: Connect to Azure Application Insights and Azure OpenAI using environment variables and secure API keys.
• Local Caching: Cache query results by user and period to reduce latency and API costs.
• AI-Powered Insights: Send both the prompt and raw data to Azure OpenAI for further summarization or analysis.
• Simple UI: A sample HTML frontend for entering prompts and displaying results.
• Extensibility: Easily adapt to new data sources, prompt types, or AI models.

Architecture Overview

[User Prompt] → [MCP Server] → [Azure App Insights] → [MCP Server] → [Azure OpenAI] → [MCP Server] → [User/UI]

• The user enters a prompt (e.g., via UI or client).
• The MCP server parses the prompt, generates a KQL query, and fetches data from Azure Application Insights.
• The server sends both the prompt and the data to Azure OpenAI for further processing.
• The AI-generated answer is returned to the user.

Prerequisites

1. You should have Open AI license, in this example I’m using Azure Open AI with gpt-4o-mini LLM model.
2. Application Insights API Access configuration, if you want to use the example here to run and test otherwise not necessary for understanding the concept.
3. You should have python basic knowledge for this article but you can create this example with any programming language of your choice.

Lets Build the Code

The code which I did for this example, here are the step by step explanation.

1. First we will start with securing configuration for all our secrets related to open AI and Application insights app keys and let’s create a env file.
   All sensitive keys and endpoints are loaded from a .env file using python-dotenv:

# Azure Application Insights
APP_INSIGHTS_APP_ID={AppInsight_API_Id}
APP_INSIGHTS_API_KEY={AppInsight_API_Key}

# Azure OpenAI
AZURE_OPENAI_ENDPOINT=https://{Azure_OpenAI_Name}.openai.azure.com
AZURE_OPENAI_KEY={Azure_OpenAI_Key}
AZURE_OPENAI_DEPLOYMENT={Deployment_LLM_Model}

2. Create the MCP server by creating a file named ‘mcp_server.py’ add below code which is complete code for my example followed by detail explanation.

import socket
import threading
import requests
import json
import re
import os
from datetime import datetime
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

APP_INSIGHTS_APP_ID = os.getenv('APP_INSIGHTS_APP_ID')
APP_INSIGHTS_API_KEY = os.getenv('APP_INSIGHTS_API_KEY')
HOST = '127.0.0.1'
PORT = 5000

# Define prompt-to-query mapping here
PROMPT_QUERY_MAP = {
    "failed requests": "requests | where success == false | summarize count()",
    "total requests": "requests | summarize count()",
    "average response time": "requests | summarize avg(duration)",
    "custom events": (
        "customEvents|project "
        "viewName=tostring(name),"
        "action_name=tostring(customDimensions['name']),"
        "action_performed=tostring(customDimensions['action']),"
        "group=tostring(customDimensions['group']),"
        "category=tostring(customDimensions['category']),"
        "subcategory=tostring(customDimensions['subcategory']),"
        "deviceType=tostring(customDimensions['deviceType']),"
        "kpi=tostring(customDimensions['kpi']),"
        "kpiview=tostring(customDimensions['kpiview']),"
        "pagerecipe=tostring(customDimensions['pagerecipe']),"
        "product=tostring(customDimensions['product']),"
        "userid=tostring(customDimensions['userid']),"
        "timestamp=tostring(timestamp)"
    ),
    # ...add more prompt-query pairs as needed...
}

# Local cache for custom events by user and month
custom_events_cache = {}

# Azure OpenAI config (set these as env vars for security)
AZURE_OPENAI_ENDPOINT = os.getenv('AZURE_OPENAI_ENDPOINT')  # e.g. 'https://<your-resource>.openai.azure.com/'
AZURE_OPENAI_KEY = os.getenv('AZURE_OPENAI_KEY')
AZURE_OPENAI_DEPLOYMENT = os.getenv('AZURE_OPENAI_DEPLOYMENT', 'gpt-35-turbo')

def fetch_app_insights(query):
    url = f"https://api.applicationinsights.io/v1/apps/{APP_INSIGHTS_APP_ID}/query"
    headers = {
        "x-api-key": APP_INSIGHTS_API_KEY
    }
    params = {
        "query": query
    }
    try:
        print('query:', query)  # Debugging line to see the query being sent
        response = requests.get(url, headers=headers, params=params, timeout=15)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        return {"error": "Request to App Insights timed out."}
    except requests.exceptions.HTTPError as e:
        return {"error": f"HTTP error: {e.response.status_code} - {e.response.text}"}
    except requests.exceptions.RequestException as e:
        return {"error": f"Request failed: {str(e)}"}
    except Exception as e:
        return {"error": f"Unexpected error: {str(e)}"}

# Always fetch all columns as defined in PROMPT_QUERY_MAP['custom events']
def fetch_custom_events(user, months):
    days = months * 30
    cache_key = (user, months, datetime.now().strftime('%Y-%m'))
    if cache_key in custom_events_cache:
        return custom_events_cache[cache_key]
    # Ensure timestamp is cast to datetime for KQL
    query = PROMPT_QUERY_MAP['custom events'] + f" | where userid == '{user}' and todatetime(timestamp) >= ago({days}d)"
    result = fetch_app_insights(query)
    if 'error' not in result:
        custom_events_cache[cache_key] = result
    return result

def call_azure_openai(prompt, data):
    if not AZURE_OPENAI_ENDPOINT or not AZURE_OPENAI_KEY:
        return {'error': 'Azure OpenAI endpoint/key not set in environment variables.'}
    url = f"{AZURE_OPENAI_ENDPOINT}/openai/deployments/{AZURE_OPENAI_DEPLOYMENT}/chat/completions?api-version=2024-02-15-preview"
    headers = {
        'Content-Type': 'application/json',
        'api-key': AZURE_OPENAI_KEY
    }
    messages = [
        {"role": "system", "content": "You are an analytics assistant. Use the provided data to answer the user's question."},
        {"role": "user", "content": f"Prompt: {prompt}\nData: {json.dumps(data)[:12000]}"}  # Truncate if too large
    ]
    payload = {"messages": messages, "max_tokens": 512}
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        return {'error': f'Azure OpenAI call failed: {str(e)}'}

def extract_user_and_days(prompt):
    # Extract user (email)
    user_match = re.search(r"([\w\.-]+@[\w\.-]+)", prompt)
    user = user_match.group(1) if user_match else None
    # Extract period (months or days)
    number_words = {'one': 1, 'two': 2, 'three': 3, 'four': 4, 'five': 5, 'six': 6, 'seven': 7, 'eight': 8, 'nine': 9, 'ten': 10, 'eleven': 11, 'twelve': 12}
    days = None
    months_match = re.search(r"(last|past) ((\d+)|([a-z]+)) month", prompt, re.IGNORECASE)
    days_match = re.search(r"(last|past) ((\d+)|([a-z]+)) day", prompt, re.IGNORECASE)
    if months_match:
        months_str = months_match.group(2)
        months = int(months_str) if months_str.isdigit() else number_words.get(months_str.lower())
        days = months * 30 if months else 90
    elif days_match:
        days_str = days_match.group(2)
        days = int(days_str) if days_str.isdigit() else number_words.get(days_str.lower(), 90)
    else:
        days = 90  # default to 90 days
    return user, days

def handle_client(conn, addr):
    print(f"Connected by {addr}")
    with conn:
        while True:
            try:
                data = conn.recv(4096)
                if not data:
                    break
                prompt = data.decode('utf-8')
                # Extract user and days from prompt
                user, days = extract_user_and_days(prompt)
                if user:
                    data_result = fetch_custom_events(user, days // 30)
                else:
                    cache_key = ("ALL", days, datetime.now().strftime('%Y-%m'))
                    if cache_key in custom_events_cache:
                        data_result = custom_events_cache[cache_key]
                    else:
                        query = PROMPT_QUERY_MAP['custom events'] + f" | where todatetime(timestamp) >= ago({days}d)"
                        data_result = fetch_app_insights(query)
                        if 'error' not in data_result:
                            custom_events_cache[cache_key] = data_result
                result = call_azure_openai(prompt, data_result)
                conn.sendall(json.dumps(result).encode('utf-8'))
            except Exception as e:
                error_result = {"error": f"Server error: {str(e)}"}
                try:
                    conn.sendall(json.dumps(error_result).encode('utf-8'))
                except Exception:
                    break

def start_server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        s.listen()
        print(f"Server listening on {HOST}:{PORT}")
        while True:
            conn, addr = s.accept()
            threading.Thread(target=handle_client, args=(conn, addr), daemon=True).start()

if __name__ == "__main__":
    start_server()

In the initial few lines of code, I’m loading the secrets from env file and define the prompt query mapping (PROMPT_QUERY_MAP) as mcp server example here is specifically created to handle the request of application insights log data by creating KQL dynamically. You may talk to openAI here it to create the KQL dynamically too.

Below code here is for your MCP server to start with the defined host and port.

def start_server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        s.listen()
        print(f"Server listening on {HOST}:{PORT}")
        while True:
            conn, addr = s.accept()
            threading.Thread(target=handle_client, args=(conn, addr), daemon=True).start()

if __name__ == "__main__":
    start_server()

With above line of code, server starts the TCP server. It binds to the specified host and port, listens for incoming client connections, and spawns a new thread for each client using handle_client.

def handle_client(conn, addr):
    print(f"Connected by {addr}")
    with conn:
        while True:
            try:
                data = conn.recv(4096)
                if not data:
                    break
                prompt = data.decode('utf-8')
                # Extract user and days from prompt
                user, days = extract_user_and_days(prompt)
                if user:
                    data_result = fetch_custom_events(user, days // 30)
                else:
                    cache_key = ("ALL", days, datetime.now().strftime('%Y-%m'))
                    if cache_key in custom_events_cache:
                        data_result = custom_events_cache[cache_key]
                    else:
                        query = PROMPT_QUERY_MAP['custom events'] + f" | where todatetime(timestamp) >= ago({days}d)"
                        data_result = fetch_app_insights(query)
                        if 'error' not in data_result:
                            custom_events_cache[cache_key] = data_result
                result = call_azure_openai(prompt, data_result)
                conn.sendall(json.dumps(result).encode('utf-8'))
            except Exception as e:
                error_result = {"error": f"Server error: {str(e)}"}
                try:
                    conn.sendall(json.dumps(error_result).encode('utf-8'))
                except Exception:
                    break

Above handle_clientfunction manages the communication with a single connected client. It receives prompts, processes them (including prompt parsing, data fetching, and AI analysis), and sends back the results.

In over all, start_server manages the server’s lifecycle and connection handling, while handle_client processes each client’s prompt and delivers analytics results. Together, they enable your MCP server to serve multiple users interactively and efficiently.

The handle_client functions process the request as following steps:
Step 1: Prompt Parsing and Dynamic Query Generation: The server uses regex to extract user emails and time periods (days/months) from prompts, converting months to days for KQL:

def extract_user_and_days(prompt):
    user_match = re.search(r"([\w\.-]+@[\w\.-]+)", prompt)
    # ...extract months/days...
    return user, days

It then builds a KQL query dynamically:

query = f"customEvents | where userid == '{user}' and todatetime(timestamp) >= ago({days}d) | summarize count() by bin(timestamp, 1d)"

Step 2: Fetching Data from Azure Application Insights: The server securely calls the App Insights REST API:

def fetch_app_insights(query):
    url = f"https://api.applicationinsights.io/v1/apps/{APP_INSIGHTS_APP_ID}/query"
    headers = {
        "x-api-key": APP_INSIGHTS_API_KEY
    }
    params = {
        "query": query
    }
    try:
        print('query:', query)  # Debugging line to see the query being sent
        response = requests.get(url, headers=headers, params=params, timeout=15)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        return {"error": "Request to App Insights timed out."}
    except requests.exceptions.HTTPError as e:
        return {"error": f"HTTP error: {e.response.status_code} - {e.response.text}"}
    except requests.exceptions.RequestException as e:
        return {"error": f"Request failed: {str(e)}"}
    except Exception as e:
        return {"error": f"Unexpected error: {str(e)}"}

Step 3: AI-Powered Analytics with Azure OpenAI: The server sends both the prompt and the fetched data to Azure OpenAI for further analysis or summarization:

def call_azure_openai(prompt, data):
    if not AZURE_OPENAI_ENDPOINT or not AZURE_OPENAI_KEY:
        return {'error': 'Azure OpenAI endpoint/key not set in environment variables.'}
    url = f"{AZURE_OPENAI_ENDPOINT}/openai/deployments/{AZURE_OPENAI_DEPLOYMENT}/chat/completions?api-version=2024-02-15-preview"
    headers = {
        'Content-Type': 'application/json',
        'api-key': AZURE_OPENAI_KEY
    }
    messages = [
        {"role": "system", "content": "You are an analytics assistant. Use the provided data to answer the user's question."},
        {"role": "user", "content": f"Prompt: {prompt}\nData: {json.dumps(data)[:12000]}"}  # Truncate if too large
    ]
    payload = {"messages": messages, "max_tokens": 512}
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        return {'error': f'Azure OpenAI call failed: {str(e)}'}

Step 4: Local Caching: To reduce redundant queries, results are cached by user and period locally which you can replace with your own caching mechanism in real life scenario project:

custom_events_cache = {}
cache_key = (user, months, datetime.now().strftime('%Y-%m'))
if cache_key in custom_events_cache:
    return custom_events_cache[cache_key]

Step 5: Simple UI or Client: You can create a UI client to test the above code but in my example here I have create a simple console client to run this, and here is the code.

file name: mcp_client.py

import socket
import json

HOST = '127.0.0.1'
PORT = 5000

def main():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((HOST, PORT))
        while True:
            prompt = input("Enter your prompt (or 'exit'): ")
            if prompt.lower() == 'exit':
                break
            s.sendall(prompt.encode('utf-8'))
            data = s.recv(65536)
            result = json.loads(data.decode('utf-8'))
            print("Result:", json.dumps(result, indent=2))

if __name__ == "__main__":
    main()

That’s all. Lets see what you get from this example.

• Real-World Integration: Connects to actual cloud services (Azure App Insights, Azure OpenAI).
• Prompt-Driven: No rigid forms or dashboards — just ask in natural language.
• Secure and Configurable: All secrets are managed via environment variables.
• Extensible: Add new prompt types, data sources, or AI models with minimal code changes.
• User-Centric: Supports both technical and non-technical users via prompt input.

Conclusion

The Model Context Protocol (MCP) pattern unlocks a new era of analytics — where users can simply ask questions and get answers, powered by cloud data and AI. This example MCP server shows how easy it is to build a secure, prompt-driven analytics backend using Python, Azure, and OpenAI.

If you’re interested in building your own MCP server, check out the full code on GitHub (or adapt the snippets above), and start empowering your users with natural language analytics today!

That wraps up this article! I hope you found the information valuable. If you’re interested in staying updated with similar content, don’t forget to follow. Thanks for reading, and happy coding!