Simple TCP Listener — Minimal Example in PythonA TCP listener is the foundation of many networked applications: it waits for incoming TCP connections, accepts them, and communicates with connected clients. This article shows a minimal, clear Python example of a TCP listener, explains the code line-by-line, covers common extensions (concurrency, error handling, basic protocols), and gives tips for testing and debugging. The goal is practical: you’ll be able to run the example, understand how it works, and adapt it for small tools, prototypes, or learning purposes.
Why a minimal TCP listener?
- Simple: Focuses on the essential concepts without framework overhead.
- Educational: Demonstrates socket basics — binding, listening, accepting, sending/receiving.
- Practical: Easy to extend for real use (concurrency, TLS, authentication, simple protocols).
Minimal synchronous TCP listener (Python 3 — complete example)
#!/usr/bin/env python3 import socket HOST = '127.0.0.1' # localhost; use '0.0.0.0' to listen on all interfaces PORT = 65432 # non-privileged port > 1023 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((HOST, PORT)) s.listen() print(f"Listening on {HOST}:{PORT} ...") conn, addr = s.accept() with conn: print('Connected by', addr) while True: data = conn.recv(1024) if not data: break print('Received:', data) conn.sendall(data) # echo back This script implements a minimal TCP echo server: it accepts one connection, echoes back any received data, and exits when the client closes the connection.
Line-by-line explanation
-
import socket
Loads the standard library socket module, which exposes low-level networking primitives. -
HOST = ‘127.0.0.1’ / PORT = 65432
Choose an address and port. Use loopback for local testing. Ports under 1024 usually require elevated privileges. -
socket.socket(socket.AF_INET, socket.SOCK_STREAM)
Creates a TCP/IPv4 socket. AF_INET means IPv4; SOCK_STREAM selects the TCP protocol. -
s.bind((HOST, PORT))
Associates the socket with the chosen address and port. If the port is in use, bind raises an error. -
s.listen()
Puts the socket into listening mode, ready to accept connections. You can pass a backlog value, e.g., s.listen(5), to limit queued connections. -
conn, addr = s.accept()
Blocks until an incoming connection arrives. Returns a new socket object (conn) for that client and the client’s address tuple (addr). -
with conn:
Ensures the client socket is closed automatically when the block exits. -
data = conn.recv(1024)
Reads up to 1024 bytes from the client. recv returns b” when the client cleanly closes the connection. -
conn.sendall(data)
Sends data back to the client, ensuring all bytes are transmitted.
Running and testing the listener
-
Save the script as simple_listener.py and run: python3 simple_listener.py
-
From another terminal, connect with netcat (nc) or telnet: nc 127.0.0.1 65432 Type text and hit Enter — the server will echo it back.
-
To stop the server: close the client connection (Ctrl+C in the client or close netcat). The server in this minimal example will exit after the connection ends. For persistent servers, wrap accept() in a loop.
Making it persistent (handle multiple sequential connections)
Wrap accept in a loop so the server continues accepting new clients:
while True: conn, addr = s.accept() with conn: print('Connected by', addr) while True: data = conn.recv(1024) if not data: break conn.sendall(data) This accepts clients one at a time, serving each to completion before accepting the next.
Adding concurrency (serve multiple clients simultaneously)
For simple concurrency, use threads:
import threading def handle_client(conn, addr): with conn: print('Connected by', addr) while True: data = conn.recv(1024) if not data: break conn.sendall(data) with socket.socket(...) as s: s.bind((HOST, PORT)) s.listen() while True: conn, addr = s.accept() thread = threading.Thread(target=handle_client, args=(conn, addr), daemon=True) thread.start() Threads let each client be handled independently. For high concurrency or async-friendly designs, consider asyncio or a process pool.
Asyncio example (modern, scalable)
import asyncio async def handle(reader, writer): addr = writer.get_extra_info('peername') print('Connected by', addr) while True: data = await reader.read(1024) if not data: break writer.write(data) await writer.drain() writer.close() await writer.wait_closed() async def main(): server = await asyncio.start_server(handle, '127.0.0.1', 65432) async with server: await server.serve_forever() asyncio.run(main()) asyncio scales better for many concurrent connections with low per-connection overhead.
Basic robustness and best practices
- Use try/except to handle socket.error and clean up resources.
- Set socket options like SO_REUSEADDR to avoid “address already in use” during quick restarts: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- Validate and limit client input to avoid memory exhaustion.
- Use timeouts (conn.settimeout(seconds)) to prevent stuck connections.
- For production, prefer TLS (wrap socket with ssl) to encrypt traffic.
- Run network services with least privilege and consider chroot/jails or containers for isolation.
Simple protocol ideas
- Plain text command/response (commands terminated by newline).
- Line-based HTTP-like responses for quick testing.
- Length-prefixed binary frames for reliable message boundaries.
Example: line-based echo — read until newline, respond, continue.
Debugging tips
- Use tcpdump/wireshark or ss/netstat to inspect connections and ports.
- Test with telnet/nc for text protocols and curl for HTTP.
- Log client addresses and errors; include timestamps for incident analysis.
When to use this minimal approach
- Learning socket programming fundamentals.
- Prototyping simple tools or local services.
- Embedded scripts or utilities where full frameworks are unnecessary.
When to use more advanced solutions
- High throughput, many concurrent clients: use asyncio or an event-driven framework.
- Production-grade services: add TLS, structured logging, metrics, robust error handling, and a tested protocol design.
Full minimal example recap
#!/usr/bin/env python3 import socket HOST = '127.0.0.1' PORT = 65432 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((HOST, PORT)) s.listen() print(f"Listening on {HOST}:{PORT} ...") while True: conn, addr = s.accept() with conn: print('Connected by', addr) while True: data = conn.recv(1024) if not data: break print('Received:', data) conn.sendall(data) This version adds SO_REUSEADDR and runs continuously, echoing every client until they disconnect.
If you want, I can:
- Provide a TLS-enabled minimal listener.
- Convert the example to Windows-compatible idioms or containerize it with Docker.
- Add a simple line-based protocol parser.
Leave a Reply