Coverage for fastagency/runtimes/ag2/tools/whatsapp.py: 100%

1from typing import Union 9 ctx1bfgahicde

2 

3from autogen import ConversableAgent 9 ctx1bfgahicde

4 

5from fastagency.api.openapi import OpenAPI 9 ctx1bfgahicde

6from fastagency.api.openapi.security import APIKeyHeader 9 ctx1bfgahicde

7from fastagency.runtimes.ag2.ag2 import Toolable 9 ctx1bfgahicde

8 

9WHATSAPP_OPENAPI_URL = "https://dev.infobip.com/openapi/products/whatsapp.json" 9 ctx1bfgahicde

10WHATSAPP_API_SERVER = "https://api.infobip.com" 9 ctx1bfgahicde

11WHATSAPP_FUNCTIONS = ["send_whatsapp_text_message"] 9 ctx1bfgahicde

12 

13 

14class WhatsAppTool(Toolable): 9 ctx1bfgahicde

15 def __init__( 9 ctx1bfgahicde

16 self, 

17 whatsapp_api_key: str, 

18 whatsapp_openapi_url: str = WHATSAPP_OPENAPI_URL, 

19 whatsapp_api_server: str = WHATSAPP_API_SERVER, 

20 ): 

21 """Create a new WhatsAppTool instance. 

22 

23 Args: 

24 whatsapp_api_key (str): The WhatsApp API key. 

25 whatsapp_openapi_url (str): Url of the openapi schema for Infobip WhatsApp API, defaults to https://dev.infobip.com/openapi/products/whatsapp.json 

26 whatsapp_api_server (str): Url of the Infobip WhatsApp API server, defaults to https://api.infobip.com 

27 """ 

28 self.whatsapp_api = OpenAPI.create( 5 ctx1bacde

29 openapi_url=whatsapp_openapi_url, 

30 servers=[{"url": whatsapp_api_server}], 

31 ) 

32 

33 header_authorization = f"App {whatsapp_api_key}" 5 ctx1bacde

34 self.whatsapp_api.set_security_params( 5 ctx1bacde

35 APIKeyHeader.Parameters(value=header_authorization) 

36 ) 

37 

38 def register( 9 ctx1bfgahicde

39 self, 

40 *, 

41 caller: ConversableAgent, 

42 executor: Union[ConversableAgent, list[ConversableAgent]], 

43 ) -> None: 

44 executors = executor if isinstance(executor, list) else [executor] 1 ctx1a

45 

46 self.whatsapp_api._register_for_llm(caller, functions=WHATSAPP_FUNCTIONS) 1 ctx1a

47 

48 for executor in executors: 1 ctx1a

49 self.whatsapp_api._register_for_execution( 1 ctx1a

50 executor, functions=WHATSAPP_FUNCTIONS 

51 )