Use port constants in core SDK files and fix formatting

- Import DEFAULT_ACP_PORT in deploy_handlers.py and base_acp_server.py
- Fix import sorting and formatting (ruff auto-fix)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: SWE Destroyer

src/agentex/lib/cli/handlers/deploy_handlers.py

@@ -11,9 +11,9 @@
 from rich.console import Console
 
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.constants.ports import DEFAULT_ACP_PORT
 from agentex.lib.cli.utils.exceptions import HelmError, DeploymentError
 from agentex.lib.cli.utils.path_utils import PathResolutionError, calculate_docker_acp_module
-from agentex.lib.constants.ports import DEFAULT_ACP_PORT
 from agentex.lib.environment_variables import EnvVarKeys
 from agentex.lib.cli.utils.kubectl_utils import check_and_switch_cluster_context
 from agentex.lib.sdk.config.agent_config import AgentConfig
@@ -240,7 +240,14 @@ def add_acp_command_to_helm_values(helm_values: dict[str, Any], manifest: AgentM
     try:
         docker_acp_module = calculate_docker_acp_module(manifest, manifest_path)
         # Create the uvicorn command with the correct module path
-        helm_values["command"] = ["uvicorn", f"{docker_acp_module}:acp", "--host", "0.0.0.0", "--port", str(DEFAULT_ACP_PORT)]
+        helm_values["command"] = [
+            "uvicorn",
+            f"{docker_acp_module}:acp",
+            "--host",
+            "0.0.0.0",
+            "--port",
+            str(DEFAULT_ACP_PORT),
+        ]
         logger.info(f"Using dynamic ACP command: uvicorn {docker_acp_module}:acp")
     except (PathResolutionError, Exception) as e:
         # Fallback to default command structure

src/agentex/lib/cli/handlers/run_handlers.py

@@ -15,7 +15,6 @@
     get_file_paths,
     calculate_uvicorn_target_for_local,
 )
-from agentex.lib.constants.ports import DEFAULT_AGENTEX_BASE_URL
 from agentex.lib.environment_variables import EnvVarKeys
 from agentex.lib.sdk.config.agent_manifest import AgentManifest
 from agentex.lib.cli.handlers.cleanup_handlers import cleanup_agent_workflows, should_cleanup_on_restart
@@ -95,16 +94,16 @@ async def start_temporal_worker_with_reload(
         worker_process = await start_temporal_worker(worker_path, env, manifest_dir)
         process_manager.add_process(worker_process)
         return asyncio.create_task(stream_process_output(worker_process, "WORKER"))
-    
+
     async def worker_runner() -> None:
         current_process: asyncio.subprocess.Process | None = None
         output_task: asyncio.Task[None] | None = None
-        
+
         console.print(f"[blue]Starting Temporal worker with auto-reload from {worker_path}...[/blue]")
-        
+
         async def start_worker() -> asyncio.subprocess.Process:
             nonlocal current_process, output_task
-            
+
             # PRE-RESTART CLEANUP - NEW!
             if current_process is not None:
                 # Extract agent name from worker path for cleanup
@@ -113,7 +112,7 @@ async def start_worker() -> asyncio.subprocess.Process:
                 console.print(f"FOUND AGENT_NAME FROM ENV VARS: {agent_name} {agent_name is None}")
                 if agent_name is None:
                     agent_name = worker_path.parent.parent.name
-                
+
                 # Perform cleanup if configured
                 if should_cleanup_on_restart():
                     console.print("[yellow]Cleaning up workflows before worker restart...[/yellow]")
@@ -122,7 +121,7 @@ async def start_worker() -> asyncio.subprocess.Process:
                     except Exception as e:
                         logger.warning(f"Cleanup failed: {e}")
                         console.print(f"[yellow]⚠ Cleanup failed: {str(e)}[/yellow]")
-            
+
             # Clean up previous process
             if current_process and current_process.returncode is None:
                 current_process.terminate()
@@ -131,41 +130,41 @@ async def start_worker() -> asyncio.subprocess.Process:
                 except asyncio.TimeoutError:
                     current_process.kill()
                     await current_process.wait()
-            
+
             # Cancel previous output task
             if output_task:
                 output_task.cancel()
                 try:
                     await output_task
                 except asyncio.CancelledError:
                     pass
-            
+
             current_process = await start_temporal_worker(worker_path, env, manifest_dir)
             process_manager.add_process(current_process)
             console.print("[green]Temporal worker started[/green]")
             return current_process
-        
+
         try:
             # Start initial worker
             current_process = await start_worker()
             if current_process:
                 output_task = asyncio.create_task(stream_process_output(current_process, "WORKER"))
-            
+
             # Watch for file changes
             async for changes in awatch(manifest_dir, recursive=True):
                 # Filter for Python files
-                py_changes = [(change, path) for change, path in changes if str(path).endswith('.py')]
-                
+                py_changes = [(change, path) for change, path in changes if str(path).endswith(".py")]
+
                 if py_changes:
                     changed_files = [str(Path(path).relative_to(worker_path.parent)) for _, path in py_changes]
                     console.print(f"[yellow]File changes detected: {changed_files}[/yellow]")
                     console.print("[yellow]Restarting Temporal worker...[/yellow]")
-                    
+
                     # Restart worker (with cleanup handled in start_worker)
                     await start_worker()
                     if current_process:
                         output_task = asyncio.create_task(stream_process_output(current_process, "WORKER"))
-                    
+
         except asyncio.CancelledError:
             # Clean shutdown
             if output_task:
@@ -174,7 +173,7 @@ async def start_worker() -> asyncio.subprocess.Process:
                     await output_task
                 except asyncio.CancelledError:
                     pass
-            
+
             if current_process and current_process.returncode is None:
                 current_process.terminate()
                 try:
@@ -183,7 +182,7 @@ async def start_worker() -> asyncio.subprocess.Process:
                     current_process.kill()
                     await current_process.wait()
             raise
-    
+
     return asyncio.create_task(worker_runner())
 
 
@@ -193,7 +192,7 @@ async def start_acp_server(
     """Start the ACP server process"""
     # Use file path relative to manifest directory if possible
     uvicorn_target = calculate_uvicorn_target_for_local(acp_path, manifest_dir)
-    
+
     cmd = [
         sys.executable,
         "-m",
@@ -297,11 +296,17 @@ async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = Non
         manifest_dir = Path(manifest_path).parent
         if debug_config and debug_config.should_debug_acp():
             acp_process = await start_acp_server_debug(
-                file_paths["acp"], manifest.local_development.agent.port, agent_env, debug_config  # type: ignore[union-attr]
+                file_paths["acp"],
+                manifest.local_development.agent.port,
+                agent_env,
+                debug_config,  # type: ignore[union-attr]
             )
         else:
             acp_process = await start_acp_server(
-                file_paths["acp"], manifest.local_development.agent.port, agent_env, manifest_dir  # type: ignore[union-attr]
+                file_paths["acp"],
+                manifest.local_development.agent.port,
+                agent_env,
+                manifest_dir,  # type: ignore[union-attr]
             )
         process_manager.add_process(acp_process)
 
@@ -314,14 +319,14 @@ async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = Non
         if is_temporal_agent(manifest) and file_paths["worker"]:
             if debug_config and debug_config.should_debug_worker():
                 # In debug mode, start worker without auto-reload to prevent conflicts
-                worker_process = await start_temporal_worker_debug(
-                    file_paths["worker"], agent_env, debug_config
-                )
+                worker_process = await start_temporal_worker_debug(file_paths["worker"], agent_env, debug_config)
                 process_manager.add_process(worker_process)
                 worker_task = asyncio.create_task(stream_process_output(worker_process, "WORKER"))
             else:
                 # Normal mode with auto-reload
-                worker_task = await start_temporal_worker_with_reload(file_paths["worker"], agent_env, process_manager, manifest_dir)
+                worker_task = await start_temporal_worker_with_reload(
+                    file_paths["worker"], agent_env, process_manager, manifest_dir
+                )
             tasks.append(worker_task)
 
         console.print(
@@ -334,7 +339,7 @@ async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = Non
             await process_manager.wait_for_shutdown()
         except KeyboardInterrupt:
             console.print("\n[yellow]Received shutdown signal...[/yellow]")
-        
+
         # Cancel output streaming tasks
         for task in tasks:
             task.cancel()
@@ -352,9 +357,6 @@ async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = Non
         await process_manager.cleanup_processes()
 
 
-
-
-
 def create_agent_environment(manifest: AgentManifest) -> dict[str, str]:
     """Create environment variables for agent processes without modifying os.environ"""
     # Start with current environment
@@ -378,6 +380,7 @@ def create_agent_environment(manifest: AgentManifest) -> dict[str, str]:
 
     # Add authorization principal if set - for local development, auth is optional
     from agentex.lib.cli.utils.auth_utils import _encode_principal_context
+
     encoded_principal = _encode_principal_context(manifest)
     if encoded_principal:
         env_vars[EnvVarKeys.AUTH_PRINCIPAL_B64] = encoded_principal

src/agentex/lib/environment_variables.py

@@ -1,4 +1,3 @@
-
 from __future__ import annotations
 
 import os
@@ -7,8 +6,8 @@
 
 from dotenv import load_dotenv
 
-from agentex.lib.constants.ports import DEFAULT_ACP_PORT, DEFAULT_AGENTEX_BASE_URL, DEFAULT_HEALTH_CHECK_PORT
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.constants.ports import DEFAULT_ACP_PORT, DEFAULT_AGENTEX_BASE_URL, DEFAULT_HEALTH_CHECK_PORT
 from agentex.lib.utils.model_utils import BaseModel
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -25,11 +25,11 @@
 )
 from agentex.lib.utils.logging import make_logger, ctx_var_request_id
 from agentex.lib.types.json_rpc import JSONRPCError, JSONRPCRequest, JSONRPCResponse
-from agentex.lib.utils.model_utils import BaseModel
-from agentex.lib.utils.registration import register_agent
 
 # from agentex.lib.sdk.fastacp.types import BaseACPConfig
 from agentex.lib.constants.ports import DEFAULT_ACP_PORT
+from agentex.lib.utils.model_utils import BaseModel
+from agentex.lib.utils.registration import register_agent
 from agentex.lib.environment_variables import EnvironmentVariables, refreshed_environment_variables
 from agentex.types.task_message_update import TaskMessageUpdate, StreamTaskMessageFull
 from agentex.types.task_message_content import TaskMessageContent
@@ -140,26 +140,21 @@ async def _handle_jsonrpc(self, request: Request):
                         error=JSONRPCError(code=-32601, message="Unauthorized"),
                     )
 
-
             # Check if method is valid first
             try:
                 method = RPCMethod(rpc_request.method)
             except ValueError:
                 logger.error(f"Method {rpc_request.method} was invalid")
                 return JSONRPCResponse(
                     id=rpc_request.id,
-                    error=JSONRPCError(
-                        code=-32601, message=f"Method {rpc_request.method} not found"
-                    ),
+                    error=JSONRPCError(code=-32601, message=f"Method {rpc_request.method} not found"),
                 )
 
             if method not in self._handlers or self._handlers[method] is None:
                 logger.error(f"Method {method} not found on existing ACP server")
                 return JSONRPCResponse(
                     id=rpc_request.id,
-                    error=JSONRPCError(
-                        code=-32601, message=f"Method {method} not found"
-                    ),
+                    error=JSONRPCError(code=-32601, message=f"Method {method} not found"),
                 )
 
             # Extract application headers using allowlist approach (only x-* headers)
@@ -172,11 +167,11 @@ async def _handle_jsonrpc(self, request: Request):
                 and key.lower() not in FASTACP_HEADER_SKIP_EXACT
                 and not any(key.lower().startswith(p) for p in FASTACP_HEADER_SKIP_PREFIXES)
             }
-            
+
             # Parse params into appropriate model based on method and include headers
             params_model = PARAMS_MODEL_BY_METHOD[method]
             params_data = dict(rpc_request.params) if rpc_request.params else {}
-            
+
             # Add custom headers to the request structure if any headers were provided
             # Gateway sends filtered headers via HTTP, SDK extracts and populates params.request
             if custom_headers:
@@ -192,12 +187,8 @@ async def _handle_jsonrpc(self, request: Request):
                     return None
                 else:
                     # Handle streaming vs non-streaming for MESSAGE_SEND
-                    if method == RPCMethod.MESSAGE_SEND and isinstance(
-                        result, AsyncGenerator
-                    ):
-                        return await self._handle_streaming_response(
-                            rpc_request.id, result
-                        )
+                    if method == RPCMethod.MESSAGE_SEND and isinstance(result, AsyncGenerator):
+                        return await self._handle_streaming_response(rpc_request.id, result)
                     else:
                         if isinstance(result, BaseModel):
                             result = result.model_dump()
@@ -209,14 +200,10 @@ async def _handle_jsonrpc(self, request: Request):
                     return JSONRPCResponse(id=None)
 
                 # For regular requests, start processing in background but return immediately
-                asyncio.create_task(
-                    self._process_request(rpc_request.id, method, params)
-                )
+                asyncio.create_task(self._process_request(rpc_request.id, method, params))
 
                 # Return immediate acknowledgment
-                return JSONRPCResponse(
-                    id=rpc_request.id, result={"status": "processing"}
-                )
+                return JSONRPCResponse(id=rpc_request.id, result={"status": "processing"})
 
         except Exception as e:
             logger.error(f"Error handling JSON-RPC request: {e}", exc_info=True)
@@ -228,9 +215,7 @@ async def _handle_jsonrpc(self, request: Request):
                 error=JSONRPCError(code=-32603, message=str(e)).model_dump(),
             )
 
-    async def _handle_streaming_response(
-        self, request_id: int | str, async_gen: AsyncGenerator
-    ):
+    async def _handle_streaming_response(self, request_id: int | str, async_gen: AsyncGenerator):
         """Handle streaming response by formatting TaskMessageUpdate objects as JSON-RPC stream"""
 
         async def generate_json_rpc_stream():
@@ -240,9 +225,7 @@ async def generate_json_rpc_stream():
                     # Validate using Pydantic's TypeAdapter to ensure it's a proper TaskMessageUpdate
                     try:
                         # This will validate that chunk conforms to the TaskMessageUpdate union type
-                        validated_chunk = task_message_update_adapter.validate_python(
-                            chunk
-                        )
+                        validated_chunk = task_message_update_adapter.validate_python(chunk)
                         # Use mode="json" to properly serialize datetime objects
                         chunk_data = validated_chunk.model_dump(mode="json")
                     except ValidationError as e:
@@ -285,18 +268,14 @@ async def _process_notification(self, method: RPCMethod, params: Any):
         except Exception as e:
             logger.error(f"Error processing notification {method}: {e}", exc_info=True)
 
-    async def _process_request(
-        self, request_id: int | str, method: RPCMethod, params: Any
-    ):
+    async def _process_request(self, request_id: int | str, method: RPCMethod, params: Any):
         """Process a request in the background"""
         try:
             handler = self._handlers[method]
             await handler(params)
             # Note: In a real implementation, you might want to store the result somewhere
             # or notify the client through a different mechanism
-            logger.info(
-                f"Successfully processed request {request_id} for method {method}"
-            )
+            logger.info(f"Successfully processed request {request_id} for method {method}")
         except Exception as e:
             logger.error(
                 f"Error processing request {request_id} for method {method}: {e}",
@@ -369,7 +348,7 @@ async def message_send_wrapper(params: SendMessageParams):
             # Check if the function is an async generator function
 
             # Regardless of whether the Agent developer implemented an Async generator or not, we will always turn the function into an async generator and yield SSE events back tot he Agentex server so there is only one way for it to process the response. Then, based on the client's desire to stream or not, the Agentex server will either yield back the async generator objects directly (if streaming) or aggregate the content into a list of TaskMessageContents and to dispatch to the client. This basically gives the Agentex server the flexibility to handle both cases itself.
-            
+
             if inspect.isasyncgenfunction(fn):
                 # The client wants streaming, an async generator already streams the content, so just return it
                 return fn(params)
@@ -381,7 +360,9 @@ async def message_send_wrapper(params: SendMessageParams):
                     task_message_content_list = []
                 elif isinstance(task_message_content_response, list):
                     # Filter out None values from lists
-                    task_message_content_list = [content for content in task_message_content_response if content is not None]
+                    task_message_content_list = [
+                        content for content in task_message_content_response if content is not None
+                    ]
                 else:
                     task_message_content_list = [task_message_content_response]
 
@@ -405,5 +386,3 @@ async def async_generator(task_message_content_list: list[TaskMessageContent]):
     def run(self, host: str = "0.0.0.0", port: int = DEFAULT_ACP_PORT, **kwargs):
         """Start the Uvicorn server for async handlers."""
         uvicorn.run(self, host=host, port=port, **kwargs)
-
-