Building a TCP Server from Scratch with Python’s socket Module

In the world of network programming, understanding how to build servers from the ground up is an essential skill for any Python developer. While modern frameworks abstract away much of the complexity, knowing how to work directly with sockets gives you a deeper understanding of network communication and allows you to build custom, lightweight solutions when frameworks are overkill.

The Transmission Control Protocol (TCP) is one of the core protocols of the Internet Protocol Suite. It provides reliable, ordered, and error-checked delivery of data between applications running on networked computers. Python’s built-in socket module provides a low-level interface for network communication, allowing us to create both servers and clients without external dependencies.

In this guide, we’ll explore how to build a TCP server from scratch, covering from basic concepts to server-ready implementations. Whether you’re building a custom API, a chat server, or any network application, this knowledge will serve as your foundation.

Understanding TCP and Sockets

What is TCP?

TCP (Transmission Control Protocol) is a connection-oriented protocol that establishes a reliable communication channel between two endpoints. Unlike UDP, which is connectionless and doesn’t guarantee delivery, TCP ensures that:

• Data arrives in the correct order
• No data is lost or duplicated
• The connection is maintained until explicitly closed
• Flow control prevents overwhelming the receiver

What is a Socket?

A socket is an endpoint for sending or receiving data across a network. Think of it as a virtual “plug” that you can connect to another socket. In Python, the socket module provides an interface to the BSD socket API, which is the standard for network programming.

There are two main types of sockets:

• Stream sockets (SOCK_STREAM): Use TCP for reliable communication
• Datagram sockets (SOCK_DGRAM): Use UDP for faster, connectionless communication

For this article, we’ll focus on TCP stream sockets.

Creating a Basic TCP Server

The Core Components

Every TCP server needs to perform these fundamental steps:

1. Create a socket — Initialize the socket object
2. Bind to an address — Associate the socket with a specific IP and port
3. Listen for connections — Put the socket into listening mode
4. Accept connections — Accept incoming client connections
5. Receive and send data — Communicate with the client
6. Close the connection — Clean up resources

Example 1: The Simplest TCP Server

Let’s start with the most basic TCP server possible:

import socket

# Create a TCP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Bind to localhost on port 8080
server_address = ('localhost', 8080)
server_socket.bind(server_address)
# Listen for incoming connections (backlog of 5)
server_socket.listen(5)
print(f"Server listening on {server_address[0]}:{server_address[1]}")
# Accept a connection
client_socket, client_address = server_socket.accept()
print(f"Connection from {client_address}")
# Receive data
data = client_socket.recv(1024)
print(f"Received: {data.decode('utf-8')}")
# Send a response
response = "Hello from server!"
client_socket.send(response.encode('utf-8'))
# Close connections
client_socket.close()
server_socket.close()

When you run this server, it will:

1. Create a socket using IPv4 (AF_INET) and TCP (SOCK_STREAM)
2. Bind to localhost:8080 and start listening
3. Block at accept() until a client connects
4. Once a client connects, receive up to 1024 bytes of data
5. Send a response back to the client
6. Close both the client and server sockets

Output:

Server listening on localhost:8080
Connection from ('127.0.0.1', 54321)
Received: Hello server!

The client address shows the IP and a random port number assigned by the operating system.

Building a Simple TCP Client

To test our server, we need a client. Here’s a basic TCP client:

Example 2: Basic TCP Client

import socket

# Create a TCP socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Connect to the server
server_address = ('localhost', 8080)
client_socket.connect(server_address)
print(f"Connected to {server_address[0]}:{server_address[1]}")
# Send data
message = "Hello server!"
client_socket.send(message.encode('utf-8'))
# Receive response
response = client_socket.recv(1024)
print(f"Server response: {response.decode('utf-8')}")
# Close the connection
client_socket.close()

When you run this client (while the server is running):

1. It creates a socket and connects to the server at localhost:8080
2. Sends the message “Hello server!”
3. Waits for and receives the server’s response
4. Closes the connection

Output:

Connected to localhost:8080
Server response: Hello from server!

Creating a Multi-Client TCP Server

The basic server above has a critical limitation: it only handles one client and then exits. Real-world servers need to handle multiple clients. Let’s improve it.

Example 3: Server with Continuous Loop

import socket

def start_server(host='localhost', port=8080):
    # Create socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # Allow port reuse (prevents "Address already in use" errors)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # Bind and listen
    server_socket.bind((host, port))
    server_socket.listen(5)
    print(f"Server started on {host}:{port}")
    
    try:
        while True:
            # Accept connection
            client_socket, client_address = server_socket.accept()
            print(f"n[NEW CONNECTION] {client_address} connected")
            
            # Receive data
            data = client_socket.recv(1024).decode('utf-8')
            print(f"[{client_address}] Received: {data}")
            
            # Send response
            response = f"Echo: {data}"
            client_socket.send(response.encode('utf-8'))
            
            # Close client connection
            client_socket.close()
            print(f"[{client_address}] Connection closed")
            
    except KeyboardInterrupt:
        print("n[SERVER] Shutting down...")
    finally:
        server_socket.close()
if __name__ == "__main__":
    start_server()

This improved server:

• Uses SO_REUSEADDR to allow immediate port reuse after the program closes
• Runs in an infinite loop, handling multiple clients sequentially
• Uses try-except to handle graceful shutdown with Ctrl+C
• Implements an echo server that sends back what it receives

Output (after multiple client connections):

Server started on localhost:8080

[NEW CONNECTION] ('127.0.0.1', 54322) connected
[('127.0.0.1', 54322)] Received: First client
[('127.0.0.1', 54322)] Connection closed
[NEW CONNECTION] ('127.0.0.1', 54323) connected
[('127.0.0.1', 54323)] Received: Second client
[('127.0.0.1', 54323)] Connection closed
[SERVER] Shutting down...

Handling Multiple Clients Simultaneously with Threading

The previous example handles clients sequentially — one at a time. For a real server, we need concurrent client handling. Let’s use threading.

Example 4: Multi-Threaded TCP Server

import socket
import threading

def handle_client(client_socket, client_address):
    """Handle communication with a single client"""
    print(f"[NEW THREAD] Handling {client_address}")
    
    try:
        while True:
            # Receive data
            data = client_socket.recv(1024).decode('utf-8')
            
            # If no data, client disconnected
            if not data:
                break
            
            print(f"[{client_address}] {data}")
            
            # Send response
            response = f"Server received: {data}"
            client_socket.send(response.encode('utf-8'))
            
    except Exception as e:
        print(f"[ERROR] {client_address}: {e}")
    finally:
        client_socket.close()
        print(f"[DISCONNECTED] {client_address}")
def start_threaded_server(host='localhost', port=8080):
    """Start a multi-threaded TCP server"""
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind((host, port))
    server_socket.listen(5)
    
    print(f"[LISTENING] Server is listening on {host}:{port}")
    
    try:
        while True:
            client_socket, client_address = server_socket.accept()
            
            # Create a new thread for each client
            thread = threading.Thread(
                target=handle_client,
                args=(client_socket, client_address)
            )
            thread.daemon = True  # Thread will die when main program exits
            thread.start()
            
            print(f"[ACTIVE CONNECTIONS] {threading.active_count() - 1}")
            
    except KeyboardInterrupt:
        print("n[SHUTDOWN] Server is shutting down...")
    finally:
        server_socket.close()
if __name__ == "__main__":
    start_threaded_server()

This advanced server:

• Creates a new thread for each client connection
• Handles multiple clients simultaneously (concurrently)
• Each client gets its own handle_client() function running in a separate thread
• The main thread continues accepting new connections while other threads handle existing clients
• Uses daemon threads so they automatically terminate when the main program exits

Output (with multiple simultaneous clients):

[LISTENING] Server is listening on localhost:8080
[NEW THREAD] Handling ('127.0.0.1', 54324)
[ACTIVE CONNECTIONS] 1
[('127.0.0.1', 54324)] Hello from client 1
[NEW THREAD] Handling ('127.0.0.1', 54325)
[ACTIVE CONNECTIONS] 2
[('127.0.0.1', 54325)] Hello from client 2
[('127.0.0.1', 54324)] Another message from client 1
[DISCONNECTED] ('127.0.0.1', 54325)
[ACTIVE CONNECTIONS] 1

Building a Persistent Connection Client

For the multi-threaded server, we need a client that can send multiple messages:

Example 5: Interactive TCP Client

import socket

def start_client(host='localhost', port=8080):
    """Start an interactive TCP client"""
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        client_socket.connect((host, port))
        print(f"Connected to {host}:{port}")
        print("Type 'quit' to exitn")
        
        while True:
            # Get user input
            message = input("You: ")
            
            if message.lower() == 'quit':
                break
            
            # Send message
            client_socket.send(message.encode('utf-8'))
            
            # Receive response
            response = client_socket.recv(1024).decode('utf-8')
            print(f"Server: {response}n")
            
    except Exception as e:
        print(f"Error: {e}")
    finally:
        client_socket.close()
        print("Connection closed")
if __name__ == "__main__":
    start_client()

This client allows interactive communication:

• Maintains a persistent connection to the server
• Sends multiple messages in a loop
• Waits for server response after each message
• Cleanly closes when user types ‘quit’

Output:

Connected to localhost:8080
Type 'quit' to exit

You: Hello server
Server: Server received: Hello server
You: How are you?
Server: Server received: How are you?
You: quit
Connection closed

Error Handling and Best Practices

Example 6: Server with Error Handling

import socket
import threading
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
class TCPServer:
    def __init__(self, host='0.0.0.0', port=8080, max_connections=5):
        self.host = host
        self.port = port
        self.max_connections = max_connections
        self.server_socket = None
        self.is_running = False
        
    def handle_client(self, client_socket, client_address):
        """Handle individual client connection"""
        logging.info(f"New connection from {client_address}")
        
        try:
            # Set timeout for recv operations
            client_socket.settimeout(300)  # 5 minutes
            
            while self.is_running:
                try:
                    data = client_socket.recv(1024)
                    
                    if not data:
                        logging.info(f"Client {client_address} disconnected")
                        break
                    
                    message = data.decode('utf-8')
                    logging.info(f"[{client_address}] Received: {message}")
                    
                    # Process and respond
                    response = self.process_message(message)
                    client_socket.send(response.encode('utf-8'))
                    
                except socket.timeout:
                    logging.warning(f"Timeout for {client_address}")
                    break
                except UnicodeDecodeError:
                    logging.error(f"Invalid UTF-8 from {client_address}")
                    client_socket.send(b"ERROR: Invalid encoding")
                    
        except Exception as e:
            logging.error(f"Error handling {client_address}: {e}")
        finally:
            client_socket.close()
            logging.info(f"Connection closed for {client_address}")
    
    def process_message(self, message):
        """Process incoming message and generate response"""
        # Add your business logic here
        return f"Processed: {message}"
    
    def start(self):
        """Start the TCP server"""
        try:
            # Create socket
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            
            # Bind to address
            self.server_socket.bind((self.host, self.port))
            self.server_socket.listen(self.max_connections)
            
            self.is_running = True
            logging.info(f"Server started on {self.host}:{self.port}")
            
            while self.is_running:
                try:
                    # Accept connection
                    client_socket, client_address = self.server_socket.accept()
                    
                    # Create thread for client
                    client_thread = threading.Thread(
                        target=self.handle_client,
                        args=(client_socket, client_address),
                        daemon=True
                    )
                    client_thread.start()
                    
                except OSError as e:
                    if self.is_running:
                        logging.error(f"Socket error: {e}")
                    break
                    
        except Exception as e:
            logging.error(f"Server error: {e}")
        finally:
            self.stop()
    
    def stop(self):
        """Stop the server gracefully"""
        logging.info("Shutting down server...")
        self.is_running = False
        
        if self.server_socket:
            self.server_socket.close()
        
        logging.info("Server stopped")
if __name__ == "__main__":
    server = TCPServer(host='localhost', port=8080)
    
    try:
        server.start()
    except KeyboardInterrupt:
        logging.info("Keyboard interrupt received")
        server.stop()

This server implements:

• Class-based design for better organization and reusability
• Logging instead of print statements for production environments
• Timeout handling to prevent indefinite blocking
• Error handling for encoding errors, socket errors, and unexpected exceptions
• Graceful shutdown that properly closes all resources
• Configurable parameters (host, port, max connections)

Output:

2025-10-11 10:30:15 - INFO - Server started on localhost:8080
2025-10-11 10:30:20 - INFO - New connection from ('127.0.0.1', 54326)
2025-10-11 10:30:21 - INFO - [('127.0.0.1', 54326)] Received: Test message
2025-10-11 10:30:25 - INFO - Client ('127.0.0.1', 54326) disconnected
2025-10-11 10:30:25 - INFO - Connection closed for ('127.0.0.1', 54326)
2025-10-11 10:30:30 - INFO - Keyboard interrupt received
2025-10-11 10:30:30 - INFO - Shutting down server...
2025-10-11 10:30:30 - INFO - Server stopped

Working with Binary Data

Sometimes you need to send binary data (images, files, etc.) instead of text:

Example 7: Sending Binary Data

import socket
import struct
def send_file_server():
    """Server that receives a file"""
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(('localhost', 8080))
    server_socket.listen(1)
    
    print("File server listening...")
    
    client_socket, address = server_socket.accept()
    print(f"Connection from {address}")
    
    # Receive file size (4 bytes)
    file_size_data = client_socket.recv(4)
    file_size = struct.unpack('>I', file_size_data)[0]
    print(f"Expecting {file_size} bytes")
    
    # Receive file data
    received_data = b''
    while len(received_data) < file_size:
        chunk = client_socket.recv(4096)
        if not chunk:
            break
        received_data += chunk
    
    # Save to file
    with open('received_file.txt', 'wb') as f:
        f.write(received_data)
    
    print(f"Received {len(received_data)} bytes and saved to file")
    
    client_socket.close()
    server_socket.close()
def send_file_client(filename):
    """Client that sends a file"""
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect(('localhost', 8080))
    
    # Read file
    with open(filename, 'rb') as f:
        file_data = f.read()
    
    # Send file size first (4 bytes, big-endian unsigned int)
    file_size = len(file_data)
    client_socket.send(struct.pack('>I', file_size))
    
    # Send file data
    client_socket.sendall(file_data)
    print(f"Sent {file_size} bytes")
    
    client_socket.close()
# Usage:
# First run: send_file_server()
# Then run: send_file_client('test.txt')

This example demonstrates:

• Binary protocol design: Sending file size first, then data
• struct.pack/unpack: Converting integers to/from bytes
• sendall(): Ensures all data is sent (handles partial sends)
• Chunked receiving: Reading data in chunks until complete

Output (Server):

File server listening...
Connection from ('127.0.0.1', 54327)
Expecting 1024 bytes
Received 1024 bytes and saved to file

Output (Client):

Sent 1024 bytes

Context Managers for Resource Management

Context managers provide:

• Automatic cleanup: Sockets are always closed, even if exceptions occur
• Cleaner code: No need for try-finally blocks everywhere
• Resource safety: Prevents socket leaks

Example 8: Using Context Managers

import socket
from contextlib import contextmanager

@contextmanager
def create_server_socket(host, port):
    """Context manager for server socket"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        sock.bind((host, port))
        sock.listen(5)
        yield sock
    finally:
        sock.close()
        print("Server socket closed")
@contextmanager
def accept_client(server_socket):
    """Context manager for client connection"""
    client_sock, address = server_socket.accept()
    try:
        yield client_sock, address
    finally:
        client_sock.close()
        print(f"Client connection closed: {address}")
# Usage example
def run_server():
    with create_server_socket('localhost', 8080) as server:
        print("Server running...")
        
        with accept_client(server) as (client, address):
            print(f"Connected: {address}")
            data = client.recv(1024)
            client.send(b"Received: " + data)
if __name__ == "__main__":
    run_server()

Output:

Server running...
Connected: ('127.0.0.1', 54328)
Client connection closed: ('127.0.0.1', 54328)
Server socket closed

Buffer Sizes

The recv() buffer size affects performance. Common values:

• 1024 bytes: Good for simple text protocols
• 4096 bytes: Better for general use
• 8192–65536 bytes: Optimal for high-throughput applications

Socket Options

# Enable TCP_NODELAY to disable Nagle's algorithm (reduces latency)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

# Set send/receive buffer sizes
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)
# Set timeout
sock.settimeout(10.0)  # 10 seconds

Security Considerations

When building production TCP servers, consider:

1. Input Validation: Always validate and sanitize received data
2. Rate Limiting: Prevent abuse by limiting connections per IP
3. Timeouts: Set appropriate timeouts to prevent resource exhaustion
4. Buffer Overflow Protection: Use fixed-size buffers
5. TLS/SSL: For sensitive data, use ssl.wrap_socket() for encryption
6. Authentication: Implement proper authentication mechanisms

Example 9: Basic SSL/TLS Server

import socket
import ssl

def create_ssl_server():
    # Create regular socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(('localhost', 8443))
    server_socket.listen(5)
    
    # Wrap with SSL
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain(certfile='server.crt', keyfile='server.key')
    
    print("SSL Server listening on port 8443...")
    
    while True:
        client_socket, address = server_socket.accept()
        
        # Wrap client socket with SSL
        ssl_client = context.wrap_socket(client_socket, server_side=True)
        
        print(f"Secure connection from {address}")
        data = ssl_client.recv(1024)
        ssl_client.send(b"Secure response")
        ssl_client.close()
# Note: Requires SSL certificate and key files

Building a TCP server from scratch with Python’s socket module provides invaluable insights into network programming fundamentals. Throughout this article, we’ve progressed from simple single-client servers to production-ready, multi-threaded implementations capable of handling concurrent connections.

---

Building a TCP Server from Scratch with Python’s socket Module was originally published in ScriptSerpent on Medium, where people are continuing the conversation by highlighting and responding to this story.