← Back to DevBytes
Python July 09, 2026 13 min read

Fix 'ConnectionRefusedError' in Python: Complete Troubleshooting Guide

Understanding ConnectionRefusedError in Python

When your Python application tries to establish a network connection and the remote server actively rejects it, you'll encounter a ConnectionRefusedError. This exception is Python's wrapper around the operating system error ECONNREFUSED (error number 111 on Linux/Unix systems). Unlike a timeout where no response arrives, this error means the target machine did respond—but with a firm "no."

The error typically manifests as:

ConnectionRefusedError: [Errno 111] Connection refused

Or when using higher-level libraries like requests or urllib:

requests.exceptions.ConnectionError: ('Connection aborted.', ConnectionRefusedError(111, 'Connection refused'))

What Triggers This Error?

At the TCP/IP level, a connection refusal happens when the client sends a SYN packet to initiate a connection, and the server responds with a RST (reset) packet instead of the expected SYN-ACK. This can occur for several distinct reasons:

Diagnosing the Root Cause

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

Before you can fix the error, you need to determine exactly where the connection attempt is failing. Work through this systematic checklist.

Step 1: Verify the Target Is Reachable

Use basic network tools to confirm the remote host is alive and the port is open:

# Check if the host is reachable at all
ping -c 4 192.168.1.100

# On Linux/macOS: test if a TCP port is open
nc -zv 192.168.1.100 5432

# Alternative: use telnet
telnet 192.168.1.100 5432

# On Windows: use Test-NetConnection in PowerShell
Test-NetConnection -ComputerName 192.168.1.100 -Port 5432

If nc or telnet also reports "Connection refused," the problem is on the server side — no process is listening there. If they hang or timeout, a firewall is likely silently dropping packets, which is a different problem from connection refusal.

Step 2: Check What's Actually Listening on the Target

On the server machine, run these commands to see which ports have active listeners:

# Linux/macOS: list all listening ports
sudo lsof -i -P -n | grep LISTEN
# Or more concisely:
ss -tlnp

# Windows:
netstat -an | find "LISTENING"

Look for your expected port number. If it's absent, you've found the problem. If it's present but bound to 127.0.0.1 (localhost) only, remote connections will be refused — the server needs to bind to 0.0.0.0 or its public IP.

Step 3: Write a Minimal Reproduction Script

Isolate the connection logic in a small script to eliminate interference from other application code:

import socket
import sys

HOST = sys.argv[1] if len(sys.argv) > 1 else "localhost"
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8000

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)

try:
    sock.connect((HOST, PORT))
    print(f"✓ Successfully connected to {HOST}:{PORT}")
    sock.close()
except ConnectionRefusedError:
    print(f"✗ Connection refused on {HOST}:{PORT}")
    print("  → No process is listening, or firewall is rejecting traffic")
except socket.timeout:
    print(f"✗ Connection timed out on {HOST}:{PORT}")
    print("  → Firewall may be dropping packets (different from refusal)")
except Exception as e:
    print(f"✗ Unexpected error: {type(e).__name__}: {e}")

Run this script with the exact host and port your application uses. The output will tell you definitively whether you're dealing with a refusal, a timeout, or something else entirely.

Common Scenarios and Their Fixes

Scenario A: Local Development Server Not Running

You're building a web app and your frontend tries to reach http://localhost:5000, but you forgot to start the backend. The fix is straightforward — start the server:

# Example: starting a Flask development server
flask run --host=0.0.0.0 --port=5000

# Or with uvicorn for FastAPI
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Always verify with your reproduction script afterward to confirm the port is now accepting connections.

Scenario B: Wrong Port Number

Your PostgreSQL connection string specifies port 5432 but the database was configured to listen on 5433. Check your configuration files, environment variables, or .env files:

# Example .env file — verify the port
DB_HOST=db.example.com
DB_PORT=5433  # ← Is this correct? Check server config
DB_NAME=myapp

# In Python, always pull from configuration, never hardcode:
import os
import psycopg2

try:
    conn = psycopg2.connect(
        host=os.environ["DB_HOST"],
        port=int(os.environ["DB_PORT"]),
        dbname=os.environ["DB_NAME"],
        connect_timeout=10
    )
except psycopg2.OperationalError as e:
    if "Connection refused" in str(e):
        print(f"Cannot reach database at {os.environ['DB_HOST']}:{os.environ['DB_PORT']}")
        print("Verify the port is correct and the database is running")
    raise

Scenario C: Server Bound to Localhost Only

This is extremely common in development and security-conscious deployments. A service listens on 127.0.0.1 and rejects all external traffic. You can detect this with ss -tlnp — if the address column shows 127.0.0.1:5432 instead of 0.0.0.0:5432 or *:5432, remote connections will be refused.

Fix it by updating the server configuration to bind to the appropriate interface:

# Flask example: bind to all interfaces
app.run(host="0.0.0.0", port=5000)

# PostgreSQL: edit postgresql.conf
# Change: listen_addresses = 'localhost'
# To:     listen_addresses = '*'
# Then restart: sudo systemctl restart postgresql

# Nginx: ensure the listen directive isn't scoped to localhost
# Good: listen 80;
# Bad:  listen 127.0.0.1:80;

Scenario D: Docker Port Mapping Issues

When running services in Docker, connection refusal often means the container's port isn't exposed to the host. You might have started the container without the -p flag, or the port mapping doesn't match:

# Wrong: container port 5432 is not exposed to the host
docker run -d --name my-postgres postgres:15

# Correct: map host port 5432 to container port 5432
docker run -d --name my-postgres -p 5432:5432 postgres:15

# Verify port mapping
docker port my-postgres

# If connecting from another container, use Docker network, not localhost
docker run -d --name app --network my-network my-app
# Inside the app container, connect to: my-postgres:5432

Scenario E: Firewall REJECT Rules

A firewall configured with REJECT policy (as opposed to DROP) will actively send RST packets, producing a ConnectionRefusedError rather than a timeout. On Linux with iptables:

# Check current firewall rules
sudo iptables -L -n

# Look for REJECT rules targeting your port
# Example problematic rule:
# REJECT tcp -- 0.0.0.0/0  0.0.0.0/0  tcp dpt:5432 reject-with icmp-port-unreachable

# Temporarily allow the port for testing
sudo iptables -I INPUT -p tcp --dport 5432 -j ACCEPT

# On cloud platforms, check security groups / firewall rules in the console
# AWS: check EC2 security group inbound rules
# GCP: check VPC firewall rules
# Azure: check NSG rules

Implementing Robust Retry Logic

Even after fixing the root cause, transient failures can occur — a service may be restarting, or a load balancer may temporarily reject connections during a health check failure. Implementing exponential backoff with jitter prevents overwhelming a recovering service while giving it time to stabilize.

import time
import random
import socket
from typing import Callable

def retry_with_backoff(
    func: Callable,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    backoff_factor: float = 2.0,
    jitter: bool = True
):
    """
    Execute a function with exponential backoff retry logic.
    
    Args:
        func: Callable that raises ConnectionRefusedError on failure
        max_retries: Maximum number of retry attempts
        base_delay: Initial delay in seconds before first retry
        max_delay: Maximum delay cap in seconds
        backoff_factor: Multiplier for successive delays
        jitter: Add random jitter to avoid thundering herd
    
    Returns:
        The return value of func on success
    
    Raises:
        ConnectionRefusedError if all retries exhausted
    """
    last_exception = None
    
    for attempt in range(max_retries + 1):
        try:
            return func()
        except ConnectionRefusedError as e:
            last_exception = e
            if attempt == max_retries:
                break
            
            delay = min(base_delay * (backoff_factor ** attempt), max_delay)
            if jitter:
                delay = delay * (0.5 + random.random())
            
            print(f"Attempt {attempt + 1} failed, retrying in {delay:.1f}s...")
            time.sleep(delay)
    
    raise ConnectionRefusedError(
        f"All {max_retries + 1} connection attempts failed. "
        f"Last error: {last_exception}"
    ) from last_exception


# Usage example with a database connection
def connect_to_database():
    import psycopg2
    conn = psycopg2.connect(
        host="db.example.com",
        port=5432,
        dbname="myapp",
        connect_timeout=5
    )
    # Store conn somewhere and return
    return conn

try:
    db_connection = retry_with_backoff(
        connect_to_database,
        max_retries=5,
        base_delay=0.5,
        max_delay=10.0
    )
    print("Database connection established successfully")
except ConnectionRefusedError as e:
    print(f"Fatal: Could not connect to database after multiple attempts: {e}")
    # Trigger alerting, fall back to read-only mode, etc.
    raise

Connection Pooling with Resilience

For database connections, use a connection pool that handles reconnections automatically. Here's an example with psycopg2's pool and a custom wrapper:

from psycopg2 import pool
from psycopg2 import OperationalError
import threading
import time

class ResilientConnectionPool:
    """A connection pool that retries on ConnectionRefusedError."""
    
    def __init__(self, minconn=2, maxconn=10, **db_kwargs):
        self.db_kwargs = db_kwargs
        self._pool = None
        self._minconn = minconn
        self._maxconn = maxconn
        self._lock = threading.Lock()
        self._initialize_pool()
    
    def _initialize_pool(self):
        """Create the pool with retry logic."""
        for attempt in range(10):
            try:
                self._pool = pool.ThreadedConnectionPool(
                    self._minconn,
                    self._maxconn,
                    **self.db_kwargs
                )
                # Validate by getting and returning a test connection
                test_conn = self._pool.getconn()
                self._pool.putconn(test_conn)
                print("Connection pool initialized successfully")
                return
            except OperationalError as e:
                if attempt == 9:
                    raise
                wait = min(0.5 * (2 ** attempt), 15)
                print(f"Pool init failed (attempt {attempt + 1}), "
                      f"retrying in {wait:.1f}s...")
                time.sleep(wait)
    
    def getconn(self):
        with self._lock:
            for attempt in range(3):
                try:
                    return self._pool.getconn()
                except pool.PoolError:
                    # Pool exhausted or connection failed
                    time.sleep(0.2 * (attempt + 1))
            raise RuntimeError("Could not acquire connection from pool")
    
    def putconn(self, conn, close=False):
        with self._lock:
            self._pool.putconn(conn, close=close)

# Usage
pool = ResilientConnectionPool(
    minconn=2,
    maxconn=10,
    host="db.example.com",
    port=5432,
    dbname="myapp",
    connect_timeout=5
)
conn = pool.getconn()
# ... use conn ...
pool.putconn(conn)

Preventing ConnectionRefusedError with Health Checks

Proactively checking service availability before making critical requests can prevent cascading failures. Implement a health-check pattern:

import socket
import time
import threading
from collections import namedtuple

ServiceStatus = namedtuple('ServiceStatus', ['available', 'last_check', 'latency'])

class ServiceHealthChecker:
    """Periodically checks service availability to avoid connection attempts
    to known-down services."""
    
    def __init__(self, host: str, port: int, check_interval: float = 5.0):
        self.host = host
        self.port = port
        self.check_interval = check_interval
        self._status = ServiceStatus(False, 0, None)
        self._running = False
        self._thread = None
    
    def start(self):
        self._running = True
        self._thread = threading.Thread(target=self._check_loop, daemon=True)
        self._thread.start()
    
    def stop(self):
        self._running = False
        if self._thread:
            self._thread.join(timeout=2)
    
    def _check_loop(self):
        while self._running:
            available, latency = self._ping()
            self._status = ServiceStatus(available, time.time(), latency)
            time.sleep(self.check_interval)
    
    def _ping(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        start = time.monotonic()
        try:
            sock.connect((self.host, self.port))
            sock.close()
            return True, (time.monotonic() - start) * 1000  # ms
        except (ConnectionRefusedError, socket.timeout, OSError):
            return False, None
    
    def is_available(self) -> bool:
        return self._status.available
    
    def status(self) -> ServiceStatus:
        return self._status


# Usage in a web application
checker = ServiceHealthChecker("payment-service.internal", 8443, check_interval=2.0)
checker.start()

def process_payment(amount: float):
    if not checker.is_available():
        raise RuntimeError("Payment service is currently unavailable")
    # Proceed with the actual payment API call
    # ...

Handling the Error in HTTP Clients

When using requests, httpx, or aiohttp, ConnectionRefusedError is wrapped inside a library-specific exception. You need to catch the right type and extract the underlying cause:

import requests
from requests.exceptions import ConnectionError as RequestsConnectionError

def fetch_with_resilience(url: str, max_retries: int = 3):
    """Fetch a URL with retry logic that handles connection refusals."""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=(5, 30))
            response.raise_for_status()
            return response.json()
        except RequestsConnectionError as e:
            # Check if the underlying cause is ConnectionRefusedError
            if isinstance(e.args[0].args[1], ConnectionRefusedError):
                print(f"Server refused connection on attempt {attempt + 1}")
                if attempt == max_retries - 1:
                    raise
                import time
                time.sleep(2 ** attempt)
            else:
                # Different connection problem (DNS failure, etc.)
                raise
        except requests.exceptions.Timeout:
            print(f"Request timed out on attempt {attempt + 1}")
            if attempt == max_retries - 1:
                raise
            import time
            time.sleep(2 ** attempt)

# Using httpx (modern alternative)
import httpx
from httpx import ConnectError

async def async_fetch(url: str):
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(url)
            return response.json()
    except ConnectError as e:
        # httpx wraps low-level connection errors
        print(f"Connection failed: {e}")
        if "Connection refused" in str(e):
            print("The target service is not accepting connections")
        raise

Best Practices Summary

Complete Troubleshooting Script

Here's a production-ready diagnostic tool you can deploy alongside your application to quickly identify connection problems:

#!/usr/bin/env python3
"""
Connection Diagnostics Tool
Tests connectivity to a list of services and reports detailed status.
"""

import socket
import time
import json
import sys
from dataclasses import dataclass, asdict
from typing import List, Optional

@dataclass
class ConnectionResult:
    host: str
    port: int
    service_name: str
    status: str  # 'ok', 'refused', 'timeout', 'dns_error', 'other'
    latency_ms: Optional[float]
    error_detail: Optional[str]

def test_connection(host: str, port: int, timeout: float = 5.0) -> ConnectionResult:
    """Test a single TCP connection and return detailed result."""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    
    start = time.monotonic()
    
    try:
        # DNS resolution happens implicitly in connect()
        sock.connect((host, port))
        latency = (time.monotonic() - start) * 1000
        sock.close()
        return ConnectionResult(
            host=host, port=port, service_name="",
            status="ok", latency_ms=round(latency, 2), error_detail=None
        )
    except ConnectionRefusedError:
        latency = (time.monotonic() - start) * 1000
        sock.close()
        return ConnectionResult(
            host=host, port=port, service_name="",
            status="refused",
            latency_ms=round(latency, 2),
            error_detail="Server actively refused connection — no listener on port"
        )
    except socket.timeout:
        sock.close()
        return ConnectionResult(
            host=host, port=port, service_name="",
            status="timeout",
            latency_ms=None,
            error_detail="No response within timeout — firewall DROP or network issue"
        )
    except socket.gaierror as e:
        return ConnectionResult(
            host=host, port=port, service_name="",
            status="dns_error",
            latency_ms=None,
            error_detail=f"DNS resolution failed: {e}"
        )
    except Exception as e:
        sock.close()
        return ConnectionResult(
            host=host, port=port, service_name="",
            status="other",
            latency_ms=None,
            error_detail=f"{type(e).__name__}: {e}"
        )

def diagnose_services(services: List[dict]) -> List[ConnectionResult]:
    """Run diagnostics on a list of service definitions."""
    results = []
    for svc in services:
        result = test_connection(svc["host"], svc["port"])
        result.service_name = svc.get("name", f"{svc['host']}:{svc['port']}")
        results.append(result)
    return results

def print_report(results: List[ConnectionResult]):
    """Print a human-readable diagnostic report."""
    print("\n" + "=" * 60)
    print("  CONNECTION DIAGNOSTICS REPORT")
    print("=" * 60)
    
    status_icons = {
        "ok": "✓",
        "refused": "✗ REFUSED",
        "timeout": "⏱ TIMEOUT",
        "dns_error": "✗ DNS",
        "other": "✗ ERROR"
    }
    
    for r in results:
        icon = status_icons.get(r.status, "?")
        latency = f"{r.latency_ms:.1f}ms" if r.latency_ms else "N/A"
        print(f"\n  {icon}  {r.service_name} ({r.host}:{r.port})")
        print(f"       Latency: {latency}")
        if r.error_detail:
            print(f"       Detail:  {r.error_detail}")
    
    # Summary
    ok_count = sum(1 for r in results if r.status == "ok")
    total = len(results)
    print(f"\n  Summary: {ok_count}/{total} services reachable")
    
    if ok_count < total:
        print("  ⚠  ACTION REQUIRED: Some services are unavailable.")
        print("     1. Verify service processes are running on target hosts")
        print("     2. Check firewall rules (REJECT vs DROP)")
        print("     3. Confirm port numbers match configuration")
        print("     4. Ensure services bind to 0.0.0.0 if remote access is needed")
    print("=" * 60 + "\n")

# Example usage
if __name__ == "__main__":
    services_to_check = [
        {"name": "Primary Database", "host": "db-primary.internal", "port": 5432},
        {"name": "Redis Cache", "host": "cache.internal", "port": 6379},
        {"name": "Auth Service", "host": "auth.internal", "port": 8443},
        {"name": "Local API", "host": "localhost", "port": 8000},
    ]
    
    # Allow override from command line: python diagnose.py host port
    if len(sys.argv) >= 3:
        services_to_check = [
            {"name": "Custom Check", "host": sys.argv[1], "port": int(sys.argv[2])}
        ]
    
    results = diagnose_services(services_to_check)
    print_report(results)
    
    # Also output machine-readable JSON
    print(json.dumps([asdict(r) for r in results], indent=2))

Conclusion

ConnectionRefusedError is one of the most straightforward network errors to diagnose once you understand the TCP handshake lifecycle. It always means the target machine received your connection request and explicitly rejected it — not silence, not a crash midway through, but a clear refusal at the TCP level. The fix almost always falls into one of the categories covered above: start the missing service, correct the port number, bind to the right interface, expose a Docker port, or adjust firewall rules from REJECT to ACCEPT. By combining systematic diagnosis with robust retry logic, health checks, and proper configuration management, you can turn this error from a production outage into a minor blip that your application handles gracefully without human intervention.

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles