🔹 Introduction: Giving Bots a Brain
To go beyond basic replies, your bot needs memory, logic, and intent recognition. Enter Copilot Studio via Direct Line API.
🧠 What is Copilot Studio (formerly PVA)?
Copilot Studio allows you to build conversational bots powered by Microsoft’s AI and natural language understanding. It can handle logic, flows, and memory.
🌐 Why Use Direct Line API?
Direct Line API allows external apps (e.g., Azure Functions or custom backends) to securely send and receive messages with your Copilot Studio bot.
🔁 Token Expiry & Caching Tip:
Direct Line tokens typically expire in 24 hours. Caching these tokens per user session reduces overhead and improves responsiveness.
Token Management and Caching
def _get_cached_token(from_user: str) -> dict | None:
current_time = datetime.now()
expired_users = [
user for user, data in token_cache.items()
if current_time - data['created_at'] >= timedelta(hours=24)
]
for expired_user in expired_users:
del token_cache[expired_user]
if from_user in token_cache:
cached_data = token_cache[from_user]
if current_time - cached_data['created_at'] < timedelta(hours=24):
return cached_data
return None
Creating a Direct Line Token and Conversation
def get_directline_token(self, from_user: str) -> dict:
user_id = f'dl_{uuid.uuid4()}'
direct_line_endpoint = f"{self.PVA_TOKEN_ENDPOINT}conversations/"
headers = {
'Authorization': f'Bearer {self.DIRECT_LINE_SECRET}',
'Content-Type': 'application/json'
}
payload = {'user': {'id': from_user}}
response = requests.post(direct_line_endpoint, headers=headers, json=payload)
response.raise_for_status()
return response.json()
Posting Context and Messages to Copilot Studio
def set_mobile_number(self, token: str, conv_id: str, user_id: str, text: str):
url = f"{self.PVA_TOKEN_ENDPOINT}conversations/{conv_id}/activities"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
payload = {
"type": "event",
"name": "UserContext",
"from": { "id": user_id },
"value": { "userMobileNumber": text }
}
requests.post(url, headers=headers, json=payload)
def post_message(
self,
token: str,
conv_id: str,
user_id: str,
text: str,
product_name: str = None,
used_info_template: bool = False,
location: str = None
):
url = f"{self.PVA_TOKEN_ENDPOINT}conversations/{conv_id}/activities"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
data = {
"type": "message",
"from": {"id": user_id},
"text": text,
"value": {
"userMobileNumber": user_id,
"productName": product_name if product_name else "Unknown Product",
"usedInfoTemplate": used_info_template,
"location": location if location else "Not Selected"
}
}
requests.post(url, headers=headers, json=data)
Fetching Bot Replies
def get_reply(self, token, conv_id, watermark=None, processed_watermarks=None, timeout_sec=20):
url = f"{self.PVA_TOKEN_ENDPOINT}conversations/{conv_id}/activities"
headers = {"Authorization": f"Bearer {token}"}
params = {"watermark": watermark} if watermark else {}
start = time.time()
bot_activities = []
last_watermark = watermark or "0"
processed_watermarks = processed_watermarks or set()
while time.time() - start < timeout_sec:
r = requests.get(url, headers=headers, params=params, timeout=10)
r.raise_for_status()
data = r.json()
activities = data.get("activities", [])
if not activities:
time.sleep(0.5)
continue
current_watermark = data.get("watermark")
new_activities = [
{
"message": act["text"],
"suggested_actions": [
{"title": action["title"], "value": action["value"]}
for action in act.get("suggestedActions", {}).get("actions", [])
] if "suggestedActions" in act else [],
"attachments": act.get("attachments", []),
"watermark": extract_watermark_number(act.get("id", "0"))
}
for act in activities
if (
act.get("from", {}).get("role") == "bot"
and act.get("text")
and extract_watermark_number(act.get("id", "0")) not in processed_watermarks
)
]
if new_activities:
processed_watermarks.update(act["watermark"] for act in new_activities)
bot_activities.extend(new_activities)
last_watermark = current_watermark
return bot_activities, last_watermark
time.sleep(0.5)
return bot_activities, last_watermark
Tips:
- Use watermarks to avoid duplicate processing.
- Pass user context (e.g., mobile number, product selection) for personalized responses.
- Implement token and session caching for efficiency.
🔹 Engagement Tip:
Your bot can now recognize “reschedule my order” and respond intelligently. Think of it as your AI-powered front desk.
Section Summary
In this part, we added intelligence to your bot using Copilot Studio (formerly Power Virtual Agents) and the Direct Line API. You learned how to manage authentication tokens, initiate conversations, and post contextual messages to the bot. These capabilities give your chatbot the ability to understand intent, respond logically, and handle complex dialogs.
