What causes "Connection closed by server." in redis/redis-py?

Ensure the asyncio event loop is running for the whole lifetime of the async Redis client (create it inside an async context / the running loop).

redis/redis-py · error · ConnectionError

Connection closed by server.

Error message

Connection closed by server.

What it means

Raised in _AsyncRESPBase.on_connect() (redis/_parsers/base.py:528) when connection._reader (the asyncio StreamReader) is None at the moment the parser's connect hook runs. This means the underlying asyncio transport was torn down or never created before the parser tried to use it, so there is no stream to read responses from. It surfaces as a ConnectionError carrying the message 'Connection closed by server.'

Solutions

Ensure the asyncio event loop is running for the whole lifetime of the async Redis client (create it inside an async context / the running loop).
Do not issue commands after calling disconnect()/aclose(); create a fresh redis.asyncio.Redis instance instead of reusing a torn-down one.
Guard against concurrent disconnect during connect (e.g., avoid closing the client from another task while a command is in flight).
If using a connection pool, let it manage reconnection rather than manually disconnecting individual connections.

Example fix

# before
r = redis.asyncio.Redis(...)
await r.disconnect()
await r.get("x")  # reader already gone -> ConnectionError

# after
r = redis.asyncio.Redis(...)
await r.get("x")  # keep client alive for its lifetime

Defensive patterns

Strategy: validation

Validate before calling

# Ensure the event loop is running and the client is not torn down before use
import asyncio, redis.asyncio as redis

async def safe_get(client, key):
    if asyncio.get_event_loop().is_closed():
        raise RuntimeError("event loop closed; create a new client in a running loop")
    # reconnect transparently if the pool permits it
    return await client.get(key)

Try / catch

try:
    await r.get("k")
except redis.ConnectionError:
    # reader was gone at connect; build a fresh client in a running loop
    r = redis.Redis(...)
    await r.get("k")

Prevention

Create redis.asyncio.Redis inside the running event loop, not at import time in a different loop.
Never issue commands after await client.aclose()/disconnect(); instantiate a new client instead.
Keep the event loop alive for the entire lifetime of the async client.

When it happens

Trigger: Calling an async Redis command on a connection whose asyncio transport/reader was never attached or was concurrently disconnected. Happens when the event loop shuts down mid-handshake, when an explicit disconnect()/close() races with connect, or when a failover/sentinel path nulls the reader before on_connect fires.

Common situations: Event loop closed while connections are still being established; reusing an async Redis client after awaiting client.disconnect(); reconnect logic that hands a half-torn-down connection back to the pool; 'Task was destroyed but it is pending!' scenarios leaving readers unattached.

Understand the failure class

Connection failures: ECONNREFUSED, ECONNRESET, and friends — why connections get refused, reset, or dropped.

Related errors

AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10). Data as JSON: /api/errors/5a0704eed488ab67. Report an issue: GitHub.

Appendix: source

Thrown at redis/_parsers/base.py:528

    __slots__ = AsyncBaseParser.__slots__ + ("encoder", "_buffer", "_pos", "_chunks")

    def __init__(self, socket_read_size: int):
        super().__init__(socket_read_size)
        self.encoder: Optional[Encoder] = None
        self._buffer = b""
        self._chunks = []
        self._pos = 0

    def _clear(self):
        self._buffer = b""
        self._chunks.clear()

    def on_connect(self, connection):
        """Called when the stream connects"""
        self._stream = connection._reader
        if self._stream is None:
            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
        self.encoder = connection.encoder
        self._clear()
        self._connected = True

    def on_disconnect(self):
        """Called when the stream disconnects"""
        self._connected = False

    @deprecated_function(
        version="8.0.0",
        reason="Use can_read() instead",
        name="can_read_destructive",
    )
    async def can_read_destructive(self) -> bool:
        return await self.can_read()

    async def can_read(self) -> bool:
        # TODO: Rename this API; it detects pending data or dirty/closed

View on GitHub (pinned to 6a6b581b48)