Remove network calls and only export ACTIVITY_SPECS list for consumption

Author: jnkrmg

application_sdk/application/__init__.py

@@ -5,7 +5,6 @@
 from application_sdk.clients.base import BaseClient
 from application_sdk.clients.utils import get_workflow_client
 from application_sdk.constants import ENABLE_MCP
-from application_sdk.decorators.automation_activity import flush_activity_registrations
 from application_sdk.events.models import EventRegistration
 from application_sdk.handlers.base import BaseHandler
 from application_sdk.observability.logger_adaptor import get_logger
@@ -153,9 +152,6 @@ async def setup_workflow(
                 workflow_and_activities_classes=workflow_and_activities_classes
             )
 
-        # Register activities via HTTP API for automation engine
-        flush_activity_registrations(app_name=self.application_name)
-
     async def start_workflow(self, workflow_args, workflow_class) -> Any:
         """
         Start a new workflow execution.

application_sdk/constants.py

@@ -49,14 +49,6 @@
 SQL_SERVER_MIN_VERSION = os.getenv("ATLAN_SQL_SERVER_MIN_VERSION")
 #: Path to the SQL queries directory
 SQL_QUERIES_PATH = os.getenv("ATLAN_SQL_QUERIES_PATH", "app/sql")
-#: Host address for the automation engine API server
-AUTOMATION_ENGINE_API_HOST = os.getenv("AUTOMATION_ENGINE_API_HOST", "localhost")
-#: Port number for the automation engine API server
-AUTOMATION_ENGINE_API_PORT = os.getenv("AUTOMATION_ENGINE_API_PORT", "8080")
-#: Base URL for automation engine API server (constructed from host and port)
-AUTOMATION_ENGINE_API_URL = (
-    f"http://{AUTOMATION_ENGINE_API_HOST}:{AUTOMATION_ENGINE_API_PORT}"
-)
 
 # Output Path Constants
 #: Output path format for workflows.

application_sdk/decorators/__init__.py

@@ -0,0 +1,6 @@
+"""Decorators module for application SDK."""
+
+from application_sdk.decorators.automation_activity import ACTIVITY_SPECS
+
+__all__: list[str] = ["ACTIVITY_SPECS"]
+

application_sdk/decorators/automation_activity.py

@@ -7,16 +7,14 @@
 from typing import Any, Callable, Dict, List, Union, get_args, get_origin
 
 from pydantic import BaseModel
-import requests
 
-from application_sdk.constants import AUTOMATION_ENGINE_API_URL
 from application_sdk.observability.logger_adaptor import get_logger
 
 logger = get_logger(__name__)
 
 
 # Global registry to collect decorated activities
-ACTIVITY_REGISTRY: List[dict[str, Any]] = []
+ACTIVITY_SPECS: List[dict[str, Any]] = []
 
 
 def _type_to_json_schema(annotation: Any) -> Dict[str, Any]:
@@ -117,7 +115,7 @@ def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
         output_schema: dict[str, Any] = _generate_output_schema(func)
 
         logger.info(f"Collected automation activity: {name}")
-        ACTIVITY_REGISTRY.append(
+        ACTIVITY_SPECS.append(
             {
                 "name": name,
                 "description": description,
@@ -129,78 +127,3 @@ def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
         return func
 
     return decorator
-
-
-def flush_activity_registrations(
-    app_name: str,
-) -> None:
-    """Flush all collected registrations by calling the activities create API via HTTP."""
-    if not ACTIVITY_REGISTRY:
-        logger.info("No activities to register")
-        return
-
-    if not AUTOMATION_ENGINE_API_URL:
-        logger.warning(
-            "Automation engine API URL not configured. Skipping activity registration."
-        )
-        return
-
-    # Perform health check first
-    try:
-        health_check_url: str = f"{AUTOMATION_ENGINE_API_URL}/api/health"
-        health_response: requests.Response = requests.get(health_check_url, timeout=5.0)
-        health_response.raise_for_status()
-        logger.info("Automation engine health check passed")
-    except Exception as e:
-        logger.warning(
-            f"Automation engine health check failed: {e}. "
-            "Skipping activity registration. "
-            "Check if the automation engine is deployed and accessible."
-        )
-        return
-
-    logger.info(
-        f"Registering {len(ACTIVITY_REGISTRY)} activities with automation engine"
-    )
-
-    # Generate app qualified name
-    app_qualified_name: str = f"default/apps/{app_name}"
-
-    # Build tools payload without function objects (not JSON serializable)
-    tools = [
-        {
-            "name": item["name"],
-            "description": item["description"],
-            "input_schema": item["input_schema"],
-            "output_schema": item["output_schema"],
-        }
-        for item in ACTIVITY_REGISTRY
-    ]
-
-    payload = {
-        "app_qualified_name": app_qualified_name,
-        "app_name": app_name,
-        "tools": tools,
-    }
-
-    try:
-        response: requests.Response = requests.post(
-            f"{AUTOMATION_ENGINE_API_URL}/api/tools",
-            json=payload,
-            timeout=30.0,
-        )
-        response.raise_for_status()
-        result = response.json()
-
-        if result.get("status") == "success":
-            logger.info(
-                f"Successfully registered {len(tools)} activities with automation engine"
-            )
-        else:
-            logger.warning(
-                f"Failed to register activities with automation engine: {result.get('message')}"
-            )
-    except Exception as e:
-        raise Exception(
-            f"Failed to register activities with automation engine: {e}"
-        ) from e