refactor(agents): route worker /agent/* HTTP through ApiClient; retire parallel token cache

Framework workers (claude_agent_sdk, langchain, langgraph) posted to Agentspan /agent/* endpoints with raw requests + a standalone token mint/cache in token_utils, a second token authority parallel to ApiClient. New _internal/agent_http.py reconstructs a cached ApiClient per (server_url, auth_key) inside the spawned worker and posts via ApiClient.call_api, reusing the SDK's token management (mint/cache/TTL-refresh/401-retry). _task_client folds into the same cache so /tasks and /agent/* share one token. token_utils is reduced to decode_jwt_exp. Adds test_agent_http.py (in-process server; mint-once-and-cache guard); test_token_utils trimmed to the decoder.

Author: v1r3n

src/conductor/ai/agents/_internal/agent_http.py

@@ -0,0 +1,104 @@
+# Copyright (c) 2025 Agentspan
+# Licensed under the MIT License. See LICENSE file in the project root for details.
+
+"""Authenticated HTTP to Agentspan ``/agent/*`` endpoints via the SDK ``ApiClient``.
+
+Framework workers (claude_agent_sdk, langchain, langgraph) run in spawned worker
+processes and receive ``(server_url, auth_key, auth_secret)`` as plain strings — a
+live ``ApiClient`` can't cross the process boundary. This module reconstructs an
+``ApiClient`` from those strings inside the worker, cached per
+``(server_url, auth_key)`` so the token mint in the constructor happens once per
+worker, and posts through :meth:`ApiClient.call_api`. That reuses the SDK's single
+token authority (mint/cache/TTL-refresh/401-retry) instead of a parallel token
+cache.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Any, Dict, Optional, Tuple
+
+logger = logging.getLogger("conductor.ai.agents.agent_http")
+
+# The auth-setting name the generated clients use to drive X-Authorization
+# injection (see OrkesAgentClient._AUTH_SETTINGS).
+_AUTH_SETTINGS = ["api_key"]
+_JSON_HEADERS = {"Accept": "application/json", "Content-Type": "application/json"}
+
+# One ApiClient per (server_url, auth_key). Building an ApiClient mints a token in
+# its constructor, so caching is mandatory to avoid a /token mint on every call.
+_API_CLIENTS: Dict[Tuple[str, str], Any] = {}
+_API_CLIENTS_LOCK = threading.Lock()
+
+
+def _agent_api_client(server_url: str, auth_key: str, auth_secret: str):
+    """Return a cached ``ApiClient`` for ``(server_url, auth_key)``.
+
+    Imports are lazy so importing this module stays cheap and side-effect free
+    (spawn/pickle safety).
+    """
+    key = (server_url, auth_key or "")
+    client = _API_CLIENTS.get(key)
+    if client is not None:
+        return client
+
+    from conductor.client.configuration.configuration import Configuration
+    from conductor.client.configuration.settings.authentication_settings import (
+        AuthenticationSettings,
+    )
+    from conductor.client.http.api_client import ApiClient
+
+    auth = (
+        AuthenticationSettings(key_id=auth_key, key_secret=auth_secret or "")
+        if auth_key
+        else None
+    )
+    config = Configuration(server_api_url=server_url, authentication_settings=auth)
+    with _API_CLIENTS_LOCK:
+        client = _API_CLIENTS.get(key)
+        if client is None:
+            client = ApiClient(config)
+            _API_CLIENTS[key] = client
+    return client
+
+
+def agent_post(
+    server_url: str,
+    auth_key: str,
+    auth_secret: str,
+    path: str,
+    body: Optional[Dict[str, Any]] = None,
+    *,
+    read_response: bool = False,
+) -> Optional[Any]:
+    """POST to an Agentspan ``/agent/*`` endpoint through the SDK ``ApiClient``.
+
+    ``path`` is relative to the server host (which already ends in ``/api``), so it
+    must start with ``/`` and omit ``/api`` — e.g. ``/agent/events/{id}``.
+
+    Returns the deserialized JSON body when ``read_response`` is True, otherwise
+    ``None``. Any transport/HTTP error (including a non-retryable ``ApiException``)
+    is swallowed and returns ``None`` so callers degrade exactly as the previous
+    raw-``requests`` paths did — a 401 is still auto-retried once inside
+    :meth:`ApiClient.call_api`.
+    """
+    try:
+        client = _agent_api_client(server_url, auth_key, auth_secret)
+        return client.call_api(
+            path,
+            "POST",
+            {},
+            [],
+            dict(_JSON_HEADERS),
+            body=body if body is not None else {},
+            post_params=[],
+            files={},
+            response_type="object" if read_response else None,
+            auth_settings=_AUTH_SETTINGS,
+            _return_http_data_only=True,
+            _preload_content=True,
+        )
+    except Exception as exc:  # ApiException + any transport-level error
+        logger.debug("agent_post failed (path=%s): %s", path, exc)
+        return None

src/conductor/ai/agents/_internal/token_utils.py

@@ -1,24 +1,18 @@
 # Copyright (c) 2025 Agentspan
 # Licensed under the MIT License. See LICENSE file in the project root for details.
 
-"""Auth token helpers shared by the sync/async agent API clients and framework adapters.
+"""JWT helper for the agent runtime.
 
-Secured Conductor hosts (e.g. orkes) authenticate API calls with a JWT in the
-``X-Authorization`` header, minted from an application access key via
-``POST {server}/token``. These helpers centralize that mint (with an expiry-aware
-process-wide cache) so every HTTP path — agent API, SSE streaming, framework event
-pushes — sends the same correct header. Anonymous servers ignore the header.
+Auth-token minting/caching lives in the SDK ``ApiClient`` (a single token
+authority — mint/cache/TTL-refresh/401-retry); the worker HTTP path reuses it via
+``conductor.ai.agents._internal.agent_http``. This module only *decodes* a JWT's
+expiry.
 """
 
 from __future__ import annotations
 
 import base64
 import json
-import logging
-import threading
-from typing import Dict, Optional, Tuple
-
-logger = logging.getLogger("conductor.ai.agents.token_utils")
 
 
 def decode_jwt_exp(token: str) -> float:
@@ -36,65 +30,3 @@ def decode_jwt_exp(token: str) -> float:
         return float(payload.get("exp", 0) or 0)
     except Exception:
         return 0.0
-
-
-# Process-wide mint cache: (server_url, auth_key) -> (token, exp). Framework event
-# pushes run on thread pools, so guard with a lock.
-_TOKEN_CACHE: Dict[Tuple[str, str], Tuple[str, float]] = {}
-_TOKEN_LOCK = threading.Lock()
-
-
-def resolve_agent_api_token(
-    server_url: str,
-    api_key: Optional[str] = None,
-    auth_key: Optional[str] = None,
-    auth_secret: Optional[str] = None,
-) -> Optional[str]:
-    """Resolve the JWT for agent API calls.
-
-    An explicit ``api_key`` is already a token and returned as-is. Otherwise a JWT is
-    minted from ``auth_key``/``auth_secret`` via ``POST {server_url}/token`` and cached
-    until ~expiry. Returns None when no credentials are configured or the mint fails
-    (anonymous / security-disabled servers).
-    """
-    if api_key:
-        return api_key
-    if not auth_key or not auth_secret:
-        return None
-
-    import time
-
-    cache_key = (server_url.rstrip("/"), auth_key)
-    with _TOKEN_LOCK:
-        cached = _TOKEN_CACHE.get(cache_key)
-        if cached:
-            token, exp = cached
-            if exp == 0.0 or time.time() < exp - 30:
-                return token
-
-    import requests
-
-    url = server_url.rstrip("/") + "/token"
-    try:
-        resp = requests.post(url, json={"keyId": auth_key, "keySecret": auth_secret}, timeout=30)
-        resp.raise_for_status()
-        token = resp.json().get("token")
-    except Exception as e:  # pragma: no cover - network/credential failures
-        logger.warning("Failed to mint agent API token: %s", e)
-        return None
-    if not token:
-        return None
-    with _TOKEN_LOCK:
-        _TOKEN_CACHE[cache_key] = (token, decode_jwt_exp(token))
-    return token
-
-
-def agent_api_auth_headers(
-    server_url: str,
-    api_key: Optional[str] = None,
-    auth_key: Optional[str] = None,
-    auth_secret: Optional[str] = None,
-) -> Dict[str, str]:
-    """``X-Authorization`` header dict for agent API calls ({} when anonymous)."""
-    token = resolve_agent_api_token(server_url, api_key, auth_key, auth_secret)
-    return {"X-Authorization": token} if token else {}

src/conductor/ai/agents/frameworks/claude_agent_sdk.py

@@ -20,7 +20,7 @@
 from dataclasses import is_dataclass, replace
 from typing import Any, Dict, List, Optional, Tuple
 
-from conductor.ai.agents._internal.token_utils import agent_api_auth_headers
+from conductor.ai.agents._internal.agent_http import _agent_api_client, agent_post
 from conductor.ai.agents.frameworks.serializer import WorkerInfo
 
 logger = logging.getLogger("conductor.ai.agents.frameworks.claude_agent_sdk")
@@ -888,17 +888,24 @@ def _push_event_nonblocking(
 
     def _do_push():
         try:
-            import requests
-
-            url = f"{server_url}/agent/events/{execution_id}"
-            headers = agent_api_auth_headers(server_url, auth_key=auth_key, auth_secret=auth_secret)
-            requests.post(url, json=event, headers=headers, timeout=5)
+            agent_post(server_url, auth_key, auth_secret, f"/agent/events/{execution_id}", event)
         except Exception as exc:
             logger.debug("Event push failed (execution_id=%s): %s", execution_id, exc)
 
     _EVENT_PUSH_POOL.submit(_do_push)
 
 
+def _task_client(server_url: str, auth_key: str, auth_secret: str):
+    """Return a ``TaskResourceApi`` backed by the shared cached ``ApiClient``.
+
+    Reuses the single per-(server_url, auth_key) ApiClient from ``agent_http`` so
+    the ``/tasks`` progress update and the ``/agent/*`` posts share one token.
+    """
+    from conductor.client.http.api.task_resource_api import TaskResourceApi
+
+    return TaskResourceApi(_agent_api_client(server_url, auth_key, auth_secret))
+
+
 def _update_task_progress_nonblocking(
     task_id: str,
     execution_id: str,
@@ -924,20 +931,15 @@ def _update_task_progress_nonblocking(
 
     def _do_update():
         try:
-            import requests
+            from conductor.client.http.models.task_result import TaskResult
 
-            url = f"{server_url}/tasks"
-            headers: Dict[str, str] = {"Content-Type": "application/json"}
-            headers.update(
-                agent_api_auth_headers(server_url, auth_key=auth_key, auth_secret=auth_secret)
+            result = TaskResult(
+                task_id=task_id,
+                workflow_instance_id=execution_id,
+                status="IN_PROGRESS",
+                output_data=progress_data,
             )
-            body = {
-                "taskId": task_id,
-                "workflowInstanceId": execution_id,
-                "status": "IN_PROGRESS",
-                "outputData": progress_data,
-            }
-            requests.post(url, json=body, headers=headers, timeout=5)
+            _task_client(server_url, auth_key, auth_secret).update_task(result)
         except Exception as exc:
             logger.debug(
                 "Task progress update failed (task_id=%s, execution_id=%s): %s",
@@ -968,26 +970,15 @@ def _create_tracking_workflow(
     Returns the execution ID of the new workflow, or None on failure.
     Uses POST /api/agent/execution (Agentspan custom endpoint).
     """
-    try:
-        import requests
-
-        url = f"{server_url}/agent/execution"
-        headers: Dict[str, str] = {"Content-Type": "application/json"}
-        headers.update(
-            agent_api_auth_headers(server_url, auth_key=auth_key, auth_secret=auth_secret)
-        )
-        body: Dict[str, Any] = {"workflowName": workflow_name, "input": input_data}
-        if parent_workflow_id:
-            body["parentWorkflowId"] = parent_workflow_id
-        if parent_workflow_task_id:
-            body["parentWorkflowTaskId"] = parent_workflow_task_id
-        resp = requests.post(url, json=body, headers=headers, timeout=5)
-        if resp.status_code < 400:
-            return resp.json().get("executionId")
-        return None
-    except Exception as exc:
-        logger.debug("Create tracking workflow failed: %s", exc)
-        return None
+    body: Dict[str, Any] = {"workflowName": workflow_name, "input": input_data}
+    if parent_workflow_id:
+        body["parentWorkflowId"] = parent_workflow_id
+    if parent_workflow_task_id:
+        body["parentWorkflowTaskId"] = parent_workflow_task_id
+    resp = agent_post(
+        server_url, auth_key, auth_secret, "/agent/execution", body, read_response=True
+    )
+    return resp.get("executionId") if isinstance(resp, dict) else None
 
 
 def _inject_tool_task(
@@ -1009,32 +1000,23 @@ def _inject_tool_task(
     For SUB_WORKFLOW tasks, pass sub_workflow_param with keys:
       name, version, executionId (the tracking sub-workflow).
     """
-    try:
-        import requests
-
-        url = f"{server_url}/agent/{execution_id}/tasks"
-        headers: Dict[str, str] = {"Content-Type": "application/json"}
-        headers.update(
-            agent_api_auth_headers(server_url, auth_key=auth_key, auth_secret=auth_secret)
-        )
-        body: Dict[str, Any] = {
-            "taskDefName": tool_name,
-            "referenceTaskName": ref_name,
-            "type": task_type,
-            "inputData": input_data,
-        }
-        if sub_workflow_param:
-            body["subWorkflowParam"] = sub_workflow_param
-        resp = requests.post(url, json=body, headers=headers, timeout=5)
-        return resp.status_code < 400
-    except Exception as exc:
-        logger.debug(
-            "Inject tool task failed (execution_id=%s, ref=%s): %s",
-            execution_id,
-            ref_name,
-            exc,
-        )
-        return False
+    body: Dict[str, Any] = {
+        "taskDefName": tool_name,
+        "referenceTaskName": ref_name,
+        "type": task_type,
+        "inputData": input_data,
+    }
+    if sub_workflow_param:
+        body["subWorkflowParam"] = sub_workflow_param
+    resp = agent_post(
+        server_url,
+        auth_key,
+        auth_secret,
+        f"/agent/{execution_id}/tasks",
+        body,
+        read_response=True,
+    )
+    return resp is not None
 
 
 def _complete_tool_task_nonblocking(
@@ -1053,14 +1035,13 @@ def _complete_tool_task_nonblocking(
 
     def _do_complete():
         try:
-            import requests
-
-            url = f"{server_url}/agent/tasks/{execution_id}/{ref_name}/{status}"
-            headers: Dict[str, str] = {"Content-Type": "application/json"}
-            headers.update(
-                agent_api_auth_headers(server_url, auth_key=auth_key, auth_secret=auth_secret)
+            agent_post(
+                server_url,
+                auth_key,
+                auth_secret,
+                f"/agent/tasks/{execution_id}/{ref_name}/{status}",
+                output_data,
             )
-            requests.post(url, json=output_data, headers=headers, timeout=5)
         except Exception as exc:
             logger.debug(
                 "Complete tool task failed (execution_id=%s, ref=%s): %s",
@@ -1086,14 +1067,13 @@ def _complete_workflow_nonblocking(
 
     def _do_complete():
         try:
-            import requests
-
-            url = f"{server_url}/agent/execution/{workflow_execution_id}/complete"
-            headers: Dict[str, str] = {"Content-Type": "application/json"}
-            headers.update(
-                agent_api_auth_headers(server_url, auth_key=auth_key, auth_secret=auth_secret)
+            agent_post(
+                server_url,
+                auth_key,
+                auth_secret,
+                f"/agent/execution/{workflow_execution_id}/complete",
+                output_data or {},
             )
-            requests.post(url, json=output_data or {}, headers=headers, timeout=5)
         except Exception as exc:
             logger.debug(
                 "Complete workflow failed (execution_id=%s): %s",