Merge pull request #4711 from Agenta-AI/add-code-hook-workflow-config

[feat] Enable full configuration of `code` and `hook` `workflows` and their `schemas`

Author: bekossy

sdks/python/agenta/sdk/decorators/running.py

@@ -43,14 +43,27 @@
     retrieve_handler,
     retrieve_interface,
     retrieve_configuration,
+    parse_uri,
 )
+from agenta.sdk.engines.running.handlers import remote_forward_v0
 
 import agenta as ag
 
 
 log = get_module_logger(__name__)
 
 
+def _is_custom_hook(uri: Optional[str]) -> bool:
+    """True for a custom hook URI (any provider/version), e.g. agenta:custom:hook:v0."""
+    if not uri:
+        return False
+    try:
+        _provider, kind, key, _version = parse_uri(uri)
+    except Exception:
+        return False
+    return kind == "custom" and key == "hook"
+
+
 class InvokeFn(Protocol):
     async def __call__(
         self,
@@ -133,6 +146,8 @@ def __init__(
         #
         revision: Optional[dict] = None,
         # -------------------------------------------------------------------- #
+        remote: bool = False,
+        # -------------------------------------------------------------------- #
         **kwargs,
     ):
         # -------------------------------------------------------------------- #
@@ -194,6 +209,8 @@ def __init__(
 
         self.handler = None
 
+        self.remote = remote
+
         self.middlewares = [
             VaultMiddleware(),
             ResolverMiddleware(),
@@ -205,22 +222,25 @@ def __init__(
         self.uri = _data.uri
 
         if self.uri is not None:
-            self._retrieve_handler(self.uri)
-
-            if self.handler:
-                registered = retrieve_interface(self.uri)
-                if registered:
-                    # merge registered interface into revision data, keeping caller overrides
-                    merged = registered.model_dump(exclude_none=True)
-                    merged.update(self.revision.data.model_dump(exclude_none=True))
-                    self.revision.data = WorkflowRevisionData(**merged)
-                    self.uri = self.revision.data.uri
-
-                registered_config = retrieve_configuration(self.uri)
-                if registered_config and not self.revision.data.parameters:
-                    self.revision.data.parameters = registered_config.parameters
-
-                self.parameters = self.revision.data.parameters
+            # A user custom hook must run its own installed handler (local) or
+            # forward to its url (remote); the URI must not resolve to a managed
+            # handler that would shadow the function attached by the decorator.
+            if not _is_custom_hook(self.uri):
+                self._retrieve_handler(self.uri)
+
+            registered = retrieve_interface(self.uri)
+            if registered:
+                # merge registered interface into revision data, keeping caller overrides
+                merged = registered.model_dump(exclude_none=True)
+                merged.update(self.revision.data.model_dump(exclude_none=True))
+                self.revision.data = WorkflowRevisionData(**merged)
+                self.uri = self.revision.data.uri
+
+            registered_config = retrieve_configuration(self.uri)
+            if registered_config and not self.revision.data.parameters:
+                self.revision.data.parameters = registered_config.parameters
+
+            self.parameters = self.revision.data.parameters
 
     def __call__(self, handler: Optional[Callable[..., Any]] = None) -> Workflow:
         if self.handler is None and handler is not None:
@@ -373,6 +393,21 @@ async def invoke(
                 )
                 running_ctx.parameters = self.parameters
 
+                # remote=True forwards to the workflow url; otherwise run the
+                # installed handler. Seeding it here lets the resolver keep the
+                # decorator's handler instead of re-resolving by URI.
+                running_ctx.handler = remote_forward_v0 if self.remote else self.handler
+                log.info(
+                    "workflow handler bound",
+                    uri=self.uri,
+                    remote=self.remote,
+                    handler=getattr(
+                        running_ctx.handler,
+                        "__name__",
+                        type(running_ctx.handler).__name__,
+                    ),
+                )
+
                 async def terminal(req: WorkflowInvokeRequest):
                     return None
 

sdks/python/agenta/sdk/engines/running/errors.py

@@ -250,6 +250,21 @@ def __init__(self, message: str, stacktrace: Optional[str] = None):
         )
 
 
+class CustomHookHandlerNotDefinedV0Error(ErrorStatus):
+    code: int = 500
+    type: str = f"{ERRORS_BASE_URL}#v0:workflows:custom-hook-handler-not-defined"
+
+    def __init__(self) -> None:
+        super().__init__(
+            code=self.code,
+            type=self.type,
+            message=(
+                "Custom hook has no handler. Define a local handler on the workflow, "
+                "or set remote=True to forward to its configured url."
+            ),
+        )
+
+
 class CustomCodeServerV0Error(ErrorStatus):
     code: int = 500
     type: str = f"{ERRORS_BASE_URL}#v0:workflows:custom-code-server-error"

sdks/python/agenta/sdk/engines/running/handlers.py

@@ -41,6 +41,7 @@
 from agenta.sdk.engines.running.templates import EVALUATOR_TEMPLATES
 from agenta.sdk.engines.running.errors import (
     CustomCodeServerV0Error,
+    CustomHookHandlerNotDefinedV0Error,
     ErrorStatus,
     InvalidConfigurationParametersV0Error,
     InvalidConfigurationParameterV0Error,
@@ -2226,8 +2227,15 @@ async def chat_v0(
     )
 
 
-@instrument(ignore_inputs=["parameters"])
-async def hook_v0(
+def _extract_revision_field(value: Optional[Data], field: str) -> Optional[Any]:
+    if isinstance(value, dict):
+        data = value.get("data") if "data" in value else value
+        if isinstance(data, dict):
+            return data.get(field)
+    return None
+
+
+async def remote_forward_v0(
     request: Optional[Data] = None,
     revision: Optional[Data] = None,
     #
@@ -2238,37 +2246,21 @@ async def hook_v0(
     trace: Optional[Data] = None,
     testcase: Optional[Data] = None,
 ) -> Any:
-    """
-    Webhook-based application handler for CUSTOM app types.
+    """Run a workflow remotely by forwarding the request to its ``url``.
 
-    Forwards the request to an external webhook URL and returns the response.
-    The webhook URL is read from the workflow interface (``url`` field in
-    revision data), not from ``parameters``.
-
-    Args:
-        request: Optional canonical request envelope.
-        revision: Optional revision data containing the webhook URL.
-        parameters: Configuration parameters forwarded to the webhook.
-        inputs: Inputs to forward to the webhook.
-        outputs: Optional outputs to forward to the webhook.
-        trace: Optional trace data to forward to the webhook.
-        testcase: Optional testcase data to forward to the webhook.
-
-    Returns:
-        The response from the webhook.
+    Selected when a workflow declares ``remote=True``. Reads ``url`` (and optional
+    ``headers``) from the revision data, POSTs to ``{url}/invoke``, and returns the
+    response. This is execution-location logic, not a per-URI handler.
     """
     from agenta.sdk.contexts.running import RunningContext
 
-    def _extract_webhook_url(value: Optional[Data]) -> Optional[str]:
-        if isinstance(value, dict):
-            data = value.get("data") if "data" in value else value
-            if isinstance(data, dict):
-                url = data.get("url")
-                return str(url) if url else None
-        return None
-
     ctx = RunningContext.get()
-    webhook_url = _extract_webhook_url(revision) or _extract_webhook_url(ctx.revision)
+    webhook_url = _extract_revision_field(revision, "url") or _extract_revision_field(
+        ctx.revision, "url"
+    )
+    headers = _extract_revision_field(revision, "headers") or _extract_revision_field(
+        ctx.revision, "headers"
+    )
 
     if not webhook_url:
         raise MissingConfigurationParameterV0Error(path="url")
@@ -2283,6 +2275,12 @@ def _extract_webhook_url(value: Optional[Data]) -> Optional[str]:
             got=webhook_url,
         ) from exc
 
+    # The stored url is the service base (pre-/invoke); the invoke surface lives
+    # at /invoke and is always appended.
+    target_url = f"{webhook_url.rstrip('/')}/invoke"
+
+    log.info("remote_forward_v0 POST", url=target_url)
+
     json_payload = {
         "inputs": inputs or {},
         "parameters": parameters or {},
@@ -2294,11 +2292,19 @@ def _extract_webhook_url(value: Optional[Data]) -> Optional[str]:
     if testcase is not None:
         json_payload["testcase"] = testcase
 
+    # httpx requires str->str headers; coerce values from revision data.
+    request_headers = (
+        {str(k): str(v) for k, v in headers.items()}
+        if isinstance(headers, dict)
+        else None
+    )
+
     async with httpx.AsyncClient() as client:
         try:
             response = await client.post(
-                url=webhook_url,
+                url=target_url,
                 json=json_payload,
+                headers=request_headers,
                 timeout=httpx.Timeout(30.0, connect=5.0),
             )
         except Exception as e:
@@ -2330,6 +2336,26 @@ def _extract_webhook_url(value: Optional[Data]) -> Optional[str]:
             return response_bytes.decode("utf-8")
 
 
+async def hook_v0(
+    request: Optional[Data] = None,
+    revision: Optional[Data] = None,
+    #
+    parameters: Optional[Data] = None,
+    inputs: Optional[Data] = None,
+    outputs: Optional[Union[Data, str]] = None,
+    #
+    trace: Optional[Data] = None,
+    testcase: Optional[Data] = None,
+) -> Any:
+    """Placeholder for the custom-hook URI. Reaching it is a misconfiguration.
+
+    A custom hook must run its own installed handler (local) or forward to its url
+    (``remote=True``). The URI never resolves to this function in either path, so
+    being here means a custom hook was invoked without a defined handler.
+    """
+    raise CustomHookHandlerNotDefinedV0Error()
+
+
 def _resolve_reference_value(reference: Any, request: Dict[str, Any]) -> Any:
     """Resolve a reference that may be a JSONPath/Pointer selector or a literal value.
 
@@ -2839,11 +2865,28 @@ async def code_v0(
             got=parameters,
         )
 
-    if "code" not in parameters:
-        raise MissingConfigurationParameterV0Error(path="code")
+    from agenta.sdk.contexts.running import RunningContext
 
-    code = str(parameters["code"])
-    runtime = str(parameters.get("runtime") or "python")
+    ctx = RunningContext.get()
+
+    # Canonical source is the revision data; legacy `parameters["code"]` is the fallback.
+    code = (
+        _extract_revision_field(revision, "script")
+        or _extract_revision_field(ctx.revision, "script")
+        or parameters.get("code")
+    )
+    if code is None:
+        raise MissingConfigurationParameterV0Error(path="script")
+
+    runtime = (
+        _extract_revision_field(revision, "runtime")
+        or _extract_revision_field(ctx.revision, "runtime")
+        or parameters.get("runtime")
+        or "python"
+    )
+
+    code = str(code)
+    runtime = str(runtime)
 
     if runtime not in ["python", "javascript", "typescript"]:
         raise InvalidConfigurationParameterV0Error(

sdks/python/agenta/sdk/engines/running/runners/restricted.py

@@ -9,13 +9,12 @@
     full_write_guard,
 )
 
-from agenta.sdk.engines.running.runners.base import CodeRunner
+from agenta.sdk.engines.running.runners.base import CodeRunner, normalize_result
 
 
 # Pure data/iteration builtins that RestrictedPython's safe_builtins omits but
 # evaluators routinely need. All operate on data only — none reach the host or
-# the class graph, so adding them does not widen the sandbox (escapes go through
-# attribute access, which safer_getattr blocks).
+# the class graph.
 _SAFE_EXTRA_BUILTINS = (
     "dict",
     "list",
@@ -38,16 +37,21 @@
 # pathlib, socket, importlib, io, shutil, ...) or the network (httpx, urllib,
 # requests, ...) is excluded. Operators who need unrestricted execution must opt
 # into the `local` runner; hostile multi-tenant should use `daytona`.
+#
+# EXCLUDED intentionally even though they look safe:
+#   - `typing`    — exposes `typing.sys`, giving access to `sys.modules` and
+#                   therefore every already-loaded module (including `os`).
+#   - `datetime`  — same escape: `datetime.sys.modules`.
+#   - `statistics`— same escape: `statistics.sys.modules`.
+# safer_getattr only blocks underscore-prefixed names; plain public attributes
+# like `sys` and `modules` on imported module objects are not blocked.
 _ALLOWED_IMPORTS = frozenset(
     {
         "math",
-        "statistics",
-        "datetime",
         "json",
         "re",
         "random",
         "string",
-        "typing",
         "collections",
         "itertools",
         "functools",
@@ -129,16 +133,21 @@ def run(
             trace: Full trace data (v2 only)
 
         Returns:
-            Float score between 0 and 1, or None if execution fails
+            Versions "1"/"2": float score between 0 and 1.
+            Version "3": any JSON-serializable value (dict, list, str, float, bool).
         """
         # Normalize runtime: None means python
         runtime = runtime or "python"
 
         # The restricted sandbox runs in-process and only supports Python.
+        # JavaScript and TypeScript require the Daytona runner
+        # (AGENTA_SERVICES_CODE_SANDBOX_RUNNER=daytona).
         if runtime != "python":
             raise ValueError(
-                f"RestrictedRunner only supports 'python' runtime, got: {runtime}. "
-                "Use the Daytona runner for javascript/typescript."
+                f"Runtime '{runtime}' is not supported by the default sandbox. "
+                "JavaScript and TypeScript evaluators require the Daytona runner. "
+                "Set AGENTA_SERVICES_CODE_SANDBOX_RUNNER=daytona, or change the "
+                "runtime to 'python'."
             )
 
         try:
@@ -153,24 +162,12 @@ def run(
 
             fn = environment["evaluate"]
 
-            if version == "2":
+            if version in ("2", "3"):
                 result = fn(inputs, output, trace)
             else:
                 result = fn(app_params, inputs, output, correct_answer)
 
-            # Attempt to convert result to float
-            if isinstance(result, (float, int, str)):
-                try:
-                    result = float(result)
-                except ValueError as e:
-                    raise ValueError(f"Result cannot be converted to float: {e}")
-
-            if not isinstance(result, float):
-                raise TypeError(
-                    f"Result is not a float after conversion: {type(result)}"
-                )
-
-            return result
+            return normalize_result(result, version)
 
         except KeyError as e:
             raise KeyError(f"Missing expected key in environment: {e}")