How to use Co-Pilot via API

Question

Hi,

I am trying to figure out if it is possible to query co-pilot and grab the generated answer via api. What I mean by this is for example say in Jira or any Ticketing System that can make calls to an API is it possible to make a call to co pilot with some type of prompt or query and in turn receive the response. Then the response data would be used to update the Jira ticket or in general be used by a third party application. I would also like to be able to build applications in .net that also can query co pilot.

I am looking to see if this is possible,
Thank you.

Answer 1

#The below is for Azure Functions to invoke the copilot bot via code. understand Activities, Events, invoke received, invoked copilot triggers.

import azure.functions as func
import logging
import requests
import time
import os
import json

# Load environment variables
from dotenv import load_dotenv
load_dotenv()

TOKEN_ENDPOINT = os.getenv('TOKEN_ENDPOINT')
BOT_SECRET = os.getenv('BOT_SECRET')

# Global variables to store conversation state
direct_line_token = None
conversation_id = None

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.function_name(name='copilot_api')
@app.route(route="copilot_api")  # Define the HTTP trigger route
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    global direct_line_token, conversation_id

    logging.info('Python HTTP trigger function processed a request.')

    try:
        # Parse the JSON body sent by Bot Framework Emulator
        req_body = req.get_json()
        
        # Extract the user message (from "text" field)
        user_message = req_body.get('text')

        # Ensure a message is present in the request body
        if not user_message:
            return func.HttpResponse(
                "User message is missing in the request body",
                status_code=400
            )

        # Get a Direct Line token if not already available
        if not direct_line_token:
            direct_line_token = get_direct_line_token()
            if not direct_line_token:
                return func.HttpResponse(
                    "Failed to obtain Direct Line token",
                    status_code=500
                )

        # Start a new conversation if not already started
        if not conversation_id:
            conversation_id = start_conversation(direct_line_token)
            if not conversation_id:
                return func.HttpResponse(
                    "Failed to start a conversation",
                    status_code=500
                )

        # Send the user's message and get the bot's response
        bot_response = send_message_and_get_response(direct_line_token, conversation_id, user_message)
        if bot_response:
            return func.HttpResponse(bot_response)
        else:
            return func.HttpResponse(
                "Failed to get a response from the bot",
                status_code=500
            )

    except ValueError:
        return func.HttpResponse(
            "Invalid request body",
            status_code=400
        )


# Function to get the Direct Line token
def get_direct_line_token():
    headers = {
        'Authorization': f'Bearer {BOT_SECRET}'
    }
    try:
        response = requests.get(TOKEN_ENDPOINT, headers=headers)
        response.raise_for_status()
        data = response.json()
        return data.get('token')
    except requests.exceptions.RequestException as e:
        logging.error(f"Error fetching Direct Line token: {e}")
        if e.response is not None:
            logging.error(f"Error details: {e.response.text}")
        return None


# Function to start a conversation
def start_conversation(direct_line_token):
    headers = {
        'Authorization': f'Bearer {direct_line_token}',
        'Content-Type': 'application/json'
    }
    try:
        response = requests.post('https://directline.botframework.com/v3/directline/conversations', headers=headers)
        response.raise_for_status()
        data = response.json()
        return data.get('conversationId')
    except requests.exceptions.RequestException as e:
        logging.error(f"Error starting conversation: {e}")
        return None


# Function to send a message and retrieve the bot's response
def send_message_and_get_response(direct_line_token, conversation_id, message):
    headers = {
        'Authorization': f'Bearer {direct_line_token}',
        'Content-Type': 'application/json'
    }
    data = {
        'type': 'message',
        'from': {'id': 'user'},
        'text': message
    }
    try:
        send_response = requests.post(
            f'https://directline.botframework.com/v3/directline/conversations/{conversation_id}/activities',
            headers=headers, json=data
        )
        send_response.raise_for_status()

        # Poll for the bot's response
        activity_id = send_response.json()['id']
        timeout = 10
        start_time = time.time()

        while time.time() - start_time < timeout:
            get_response = requests.get(
                f'https://directline.botframework.com/v3/directline/conversations/{conversation_id}/activities',
                headers=headers
            )
            get_response.raise_for_status()

            activities = get_response.json()['activities']
            for activity in activities:
                if (activity['type'] == 'message' and
                        activity['from']['id'] != 'user' and
                        'replyToId' in activity and
                        activity['replyToId'] == activity_id):
                    return activity['text']

            time.sleep(0.5)

        logging.warning("Timeout waiting for bot's response.")
        return None

    except requests.exceptions.RequestException as e:
        logging.error(f"Error interacting with Direct Line: {e}")
        if e.response is not None:
            logging.error(f"Error details: {e.response.text}")
        return None