feat(autogpt/cli): Check if port is available before running server (#6996)

autogpts/autogpt/autogpt/app/agent_protocol_server.py

@@ -31,6 +31,7 @@ from autogpt.agent_factory.configurators import configure_agent_with_state
 from autogpt.agent_factory.generators import generate_agent_for_task
 from autogpt.agent_manager import AgentManager
 from autogpt.agents.utils.exceptions import AgentFinished
+from autogpt.app.utils import is_port_free
 from autogpt.commands.system import finish
 from autogpt.commands.user_interaction import ask_user
 from autogpt.config import Config
@@ -64,6 +65,14 @@ class AgentProtocolServer:
     async def start(self, port: int = 8000, router: APIRouter = base_router):
         """Start the agent server."""
         logger.debug("Starting the agent server...")
+        if not is_port_free(port):
+            logger.error(f"Port {port} is already in use.")
+            logger.info(
+                "You can specify a port by either setting the AP_SERVER_PORT "
+                "environment variable or defining AP_SERVER_PORT in the .env file."
+            )
+            return
+
         config = HypercornConfig()
         config.bind = [f"localhost:{port}"]
         app = FastAPI(

autogpts/autogpt/autogpt/app/utils.py

@@ -2,6 +2,7 @@ import contextlib
 import logging
 import os
 import re
+import socket
 import sys
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -272,3 +273,12 @@ def set_env_config_value(key: str, value: str) -> None:
             file.write(f"{key}={value}\n")
 
         file.truncate()
+
+
+def is_port_free(port: int, host: str = "127.0.0.1"):
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        try:
+            s.bind((host, port))  # Try to bind to the port
+            return True  # If successful, the port is free
+        except OSError:
+            return False  # If failed, the port is likely in use