Skip to content

Socket Programming

socket is Python’s standard library module for low-level network communication. All higher-level networking libraries (http, asyncio, aiohttp, etc.) are built on top of socket.

Socket Basic Functions

FunctionDescription
socket.socket(family, type)Create a socket. AF_INET=IPv4, SOCK_STREAM=TCP, SOCK_DGRAM=UDP
s.bind((host, port))Bind an address and port (server side)
s.listen(backlog)Start listening; backlog is the length of the pending connection queue
s.accept()Block and wait for a client connection; returns (conn, addr)
s.connect((host, port))Connect to the server (client side)
s.send(data)Send byte data
s.recv(bufsize)Receive up to bufsize bytes
s.close()Close the socket
s.setsockopt(level, option, value)Set socket options
s.settimeout(seconds)Set a timeout

TCP Communication Example

TCP is a reliable, connection-oriented protocol. A three-way handshake must establish a connection before communication begins.

TCP Server

import socket

def run_server(host: str = "127.0.0.1", port: int = 9000) -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    # SO_REUSEADDR: allow reuse of the port to avoid "address already in use" errors from TIME_WAIT state
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    server.bind((host, port))
    server.listen(5)
    print(f"Server listening on {host}:{port}")

    while True:
        conn, addr = server.accept()   # Block and wait for a new connection
        print(f"Client connected: {addr}")

        try:
            while True:
                data = conn.recv(1024)
                if not data:
                    break   # Client disconnected

                message = data.decode("utf-8")
                print(f"Received: {message}")

                reply = f"Server received: {message}"
                conn.sendall(reply.encode("utf-8"))
        finally:
            conn.close()
            print(f"Client {addr} disconnected")

if __name__ == "__main__":
    run_server()

TCP Client

import socket

def run_client(host: str = "127.0.0.1", port: int = 9000) -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
        client.connect((host, port))
        print(f"Connected to {host}:{port}")

        messages = ["Hello!", "Python socket", "Goodbye"]
        for msg in messages:
            client.sendall(msg.encode("utf-8"))
            data = client.recv(1024)
            print(f"Server reply: {data.decode('utf-8')}")

if __name__ == "__main__":
    run_client()

Handling Multiple Clients (Multi-threading)

A single-process server can only handle one client at a time. Using multiple threads allows concurrent handling of multiple connections:

import socket
import threading

def handle_client(conn: socket.socket, addr: tuple) -> None:
    print(f"New connection: {addr}")
    try:
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data)   # Echo server: send data back as-is
    finally:
        conn.close()

def run_echo_server(host: str = "0.0.0.0", port: int = 9001) -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((host, port))
        server.listen(10)
        print(f"Echo server started on {host}:{port}")

        while True:
            conn, addr = server.accept()
            t = threading.Thread(target=handle_client, args=(conn, addr), daemon=True)
            t.start()

if __name__ == "__main__":
    run_echo_server()

UDP Communication Example

UDP is connectionless and does not guarantee delivery, but it is fast and suitable for real-time scenarios (video streaming, gaming, etc.).

UDP Server

import socket

def run_udp_server(host: str = "127.0.0.1", port: int = 9002) -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
        server.bind((host, port))
        print(f"UDP server listening on {host}:{port}")

        while True:
            data, addr = server.recvfrom(1024)   # Returns both data and the sender's address
            print(f"From {addr}: {data.decode()}")

            reply = f"Received: {data.decode()}"
            server.sendto(reply.encode(), addr)

if __name__ == "__main__":
    run_udp_server()

UDP Client

import socket

def run_udp_client(host: str = "127.0.0.1", port: int = 9002) -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
        client.settimeout(5)   # 5-second timeout

        for msg in ["Hello", "World"]:
            client.sendto(msg.encode(), (host, port))
            try:
                reply, server_addr = client.recvfrom(1024)
                print(f"Server reply: {reply.decode()}")
            except socket.timeout:
                print("Timed out waiting for reply")

if __name__ == "__main__":
    run_udp_client()

Timeouts and Non-blocking

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("127.0.0.1", 9003))
server.listen(5)

# Set a timeout for accept
server.settimeout(10)

try:
    conn, addr = server.accept()
    print(f"Connection: {addr}")
except socket.timeout:
    print("No client connected within 10 seconds, exiting")
finally:
    server.close()

Getting Local Host Information

import socket

hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
print(f"Hostname: {hostname}")
print(f"Local IP: {local_ip}")

# Look up a remote host's IP
remote_ip = socket.gethostbyname("www.python.org")
print(f"python.org IP: {remote_ip}")
Last updated on