feat(vefaas): get application endpoint (#35)

Author: zakahan

veadk/cli/services/vefaas/template/src/requirements.txt

@@ -1,5 +1,6 @@
 veadk-python[eval] @ git+https://github.com/volcengine/veadk-python.git # extra eval for prompt optimization in veadk studio
 opensearch-py==2.8.0
 agent-pilot-sdk>=0.0.9 # extra dep for prompt optimization in veadk studio
+typer>=0.16.0
 uvicorn[standard]
 fastapi

veadk/cli/services/vefaas/vefaas.py

@@ -202,6 +202,20 @@ def _list_application(self):
         )
         return response["Result"]["Items"]
 
+    def get_application_details(self, app_id: str = None, app_name: str = None):
+        if not app_id and not app_name:
+            raise ValueError("app_id and app_name cannot be both empty.")
+        apps = self._list_application()
+        if app_id:
+            for app in apps:
+                if app["Id"] == app_id:
+                    return app
+            return None
+        else:
+            for app in apps:
+                if app["Name"] == app_name:
+                    return app
+
     def find_app_id_by_name(self, name: str):
         apps = self._list_application()
         for app in apps:

veadk/cloud/cloud_app.py

@@ -15,6 +15,7 @@
 from typing import Any
 from uuid import uuid4
 
+import json
 import httpx
 from a2a.client import A2ACardResolver, A2AClient
 from a2a.types import AgentCard, Message, MessageSendParams, SendMessageRequest
@@ -36,9 +37,9 @@ class CloudApp:
 
     def __init__(
         self,
-        vefaas_application_name: str,
-        vefaas_endpoint: str,
-        vefaas_application_id: str,
+        vefaas_application_name: str = "",
+        vefaas_endpoint: str = "",
+        vefaas_application_id: str = "",
         use_agent_card: bool = False,
     ):
         self.vefaas_endpoint = vefaas_endpoint
@@ -73,18 +74,24 @@ def __init__(
 
         self.httpx_client = httpx.AsyncClient()
 
-    def _get_vefaas_endpoint(self) -> str:
-        vefaas_endpoint = ""
-
-        if self.vefaas_application_id:
-            # TODO(zakahan): get endpoint from vefaas
-            vefaas_endpoint = ...
-            return vefaas_endpoint
+    def _get_vefaas_endpoint(
+        self,
+        volcengine_ak: str = getenv("VOLCENGINE_ACCESS_KEY"),
+        volcengine_sk: str = getenv("VOLCENGINE_SECRET_KEY"),
+    ) -> str:
+        from veadk.cli.services.vefaas.vefaas import VeFaaS
 
-        if self.vefaas_application_name:
-            # TODO(zakahan): get endpoint from vefaas
-            vefaas_endpoint = ...
-            return vefaas_endpoint
+        vefaas_client = VeFaaS(access_key=volcengine_ak, secret_key=volcengine_sk)
+        app = vefaas_client.get_application_details(
+            app_id=self.vefaas_application_id,
+            app_name=self.vefaas_application_name,
+        )
+        cloud_resource = json.loads(app["CloudResource"])
+        try:
+            vefaas_endpoint = cloud_resource["framework"]["url"]["system_url"]
+        except Exception as e:
+            raise ValueError(f"VeFaaS cloudAPP could not get endpoint. Error: {e}")
+        return vefaas_endpoint
 
     def _get_vefaas_application_id_by_name(self) -> str:
         if not self.vefaas_application_name: