main.py

class DIDx_Chatbot:
    def __init__(self):
        import openai

        def get_balance():
            import requests
            response = requests.get(
                f'https://newapi.didx.net/DidxApis/api/WebGetAccountBalance.php?UserID={user_id}&Password={password}')
            return response.text

        def get_available_dids_country():
            import requests
            response = requests.get(
                f'https://newapi.didx.net/DidxApis/api/getDIDCountry.php?UserID=700290&Password={password}')
            return response.text

        def get_available_dids():
            import requests
            response = requests.get(
                f'https://newapi.didx.net/DidxApis/api/getAvailableDIDS.php?UserID=700290&Password={password}&CountryCode=1&AreaCode=-1')
            return response.text

        self.get_balance = get_balance
        self.get_available_dids_country = get_available_dids_country
        self.get_available_dids = get_available_dids

        function_balance = {
            "type": "function",
            "function": {
                "name": "get_balance",
                "description": "Retrieve the available balance present in the account",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "Balance": {
                            "type": "string",
                            "description": "A balance in the amount of dollars"
                        },
                        "Status": {
                            "type": "string",
                            "description": "A balance status due or something else"

                        }
                    },
                    "required": ["account balance", "Status"]
                }
            }
        }
        function_dids_country = {
            "type": "function",
            "function": {
                "name": "get_available_dids_country",
                "description": "Retrieve the available DIDs country information",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "country_code": {
                            "type": "string",
                            "description": "Country code for available DIDs"
                        },
                        "description": {
                            "type": "string",
                            "description": "Country/City/County code for available DIDs"

                        },
                        "country_id": {
                            "type": "string",
                            "description": "Country/City/County code for available DIDs"
                        }

                    },
                    "required": ["country_code", "description", "country_id"]
                }
            }
        }
        function_dids = {
            "type": "function",
            "function": {
                "name": "get_available_dids",
                "description": "Retrieve all available DIDs list",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "CountryCode": {
                            "type": "string",
                            "description": "Country code for available DIDs. Default CountryCode is 1 for US"
                        },
                        "description": {
                            "type": "string",
                            "description": "A list of available DIDs"

                        },
                        "AreaCode": {
                            "type": "string",
                            "description": "Area Code for the available DIDs. Default AreaCode is -1 that will show all AreaCode"
                        }

                    },
                    "required": ["CountryCode", "description", "AreaCode"]
                }
            }
        }

        import os
        self.client = openai.OpenAI(api_key=os.environ['openai_api_key'])

        # Step 1: Create an Assistant
        self.assistant = self.client.beta.assistants.create(
            name="DIDx Customer Support Chatbot",
            instructions="You are a personal DIDx customer support chatbot.",
            tools=[function_balance, function_dids_country, function_dids],
            model="gpt-4-1106-preview",
        )

    def user_auth(self, user, passw):
        global user_id
        global password

        user_id = user
        password = passw

    def user_chat(self, query):
        import time
        # Step 2: Create a Thread
        thread = self.client.beta.threads.create()

        # Step 3: Add a Message to a Thread
        message = self.client.beta.threads.messages.create(
            thread_id=thread.id,
            role="user",
            content=query
        )

        # Step 4: Run the Assistant
        run = self.client.beta.threads.runs.create(
            thread_id=thread.id,
            assistant_id=self.assistant.id,
            instructions=""
        )
        answer = None
        while True:
            # Retrieve the run status
            run_status = self.client.beta.threads.runs.retrieve(
                thread_id=thread.id,
                run_id=run.id
            )
            # print(run_status.model_dump_json(indent=4))
            run_status.model_dump_json(indent=4)

            # If run is completed, get messages
            if run_status.status == 'completed':
                messages = self.client.beta.threads.messages.list(
                    thread_id=thread.id
                )
                # Loop through messages and print content based on role
                for msg in messages.data:
                    role = msg.role
                    content = msg.content[0].text.value
                    print(f"{role.capitalize()}: {content}")
                    answer = f"{role.capitalize()}: {content}"
                    break
                break
            elif run_status.status == 'requires_action':
                # print("Function Calling")
                required_actions = run_status.required_action.submit_tool_outputs.model_dump()
                # print('required action test: ',required_actions)
                tool_outputs = []
                import json
                for action in required_actions["tool_calls"]:
                    func_name = action['function']['name']
                    arguments = json.loads(action['function']['arguments'])

                    if func_name == "get_balance":
                        output = self.get_balance()
                        tool_outputs.append({
                            "tool_call_id": action['id'],
                            "output": output
                        })

                    elif func_name == 'get_available_dids_country':
                        output = self.get_available_dids_country()
                        tool_outputs.append({
                            "tool_call_id": action['id'],
                            "output": output
                        })
                    elif func_name == 'get_available_dids':
                        output = self.get_available_dids()
                        tool_outputs.append({
                            "tool_call_id": action['id'],
                            "output": output
                        })

                    else:
                        raise ValueError(f"Unknown function: {func_name}")

                print("Submitting outputs back to the Assistant...")
                self.client.beta.threads.runs.submit_tool_outputs(
                    thread_id=thread.id,
                    run_id=run.id,
                    tool_outputs=tool_outputs
                )
            else:
                print("Waiting for the Assistant to process...")
                time.sleep(5)

        if answer is not None:
            print(f'this is my answer : ', answer)
            return answer