Python Zeroconf (mDNS)
Zero-configuration networking for service discovery and advertisement on local networks using multicast DNS.
Python Zeroconf (mDNS)
Zero-configuration networking for service discovery and advertisement on local networks using multicast DNS.
Overview
Zeroconf (Zero Configuration Networking) enables automatic discovery of devices and services on a local network without manual configuration. The Python zeroconf library implements mDNS (multicast DNS) and DNS-SD (DNS Service Discovery) protocols, commonly used for IoT devices, network printers, media servers, and local development environments.
flowchart LR
subgraph Local Network
A[Service Provider] -->|Advertise| B[mDNS/DNS-SD]
C[Service Consumer] -->|Browse/Query| B
B -->|Resolve| C
end
A -->|Register _http._tcp.local.| B
C -->|Find web servers| B
Installation
uv add zeroconf # or: pip install zeroconf
The async API (zeroconf.asyncio) ships with the base package — there is no
separate extra to install.
Service Discovery Basics
How mDNS Works
mDNS operates on UDP port 5353 using multicast address 224.0.0.251 (IPv4) or ff02::fb (IPv6). Services announce themselves and respond to queries without a central DNS server.
sequenceDiagram
participant S as Service
participant N as Network (mDNS)
participant C as Client
S->>N: Register service
S->>N: Announce (multicast)
C->>N: Query for services
N-->>C: Response with service info
C->>S: Direct connection
Core Components
from zeroconf import Zeroconf, ServiceInfo, ServiceBrowser
# Main Zeroconf instance - manages all mDNS operations
zc = Zeroconf()
# ServiceInfo - describes a service
# ServiceBrowser - discovers services
# ServiceListener - handles discovery events
Service Types and Naming Conventions
Service types follow the format: _service._protocol.local.
| Service Type | Description |
|---|---|
_http._tcp.local. |
Web servers |
_https._tcp.local. |
Secure web servers |
_ssh._tcp.local. |
SSH servers |
_mqtt._tcp.local. |
MQTT brokers |
_hap._tcp.local. |
HomeKit accessories |
_printer._tcp.local. |
Network printers |
_airplay._tcp.local. |
AirPlay devices |
_googlecast._tcp.local. |
Chromecast devices |
_ipp._tcp.local. |
Internet Printing Protocol |
Service names format: Instance Name._service._protocol.local.
# Example: "My Web Server._http._tcp.local."
service_type = "_http._tcp.local."
service_name = f"My Web Server.{service_type}"
Browsing and Resolving Services
Basic Service Browser
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener
class MyListener(ServiceListener):
def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
info = zc.get_service_info(type_, name)
if info:
print(f"Service added: {name}")
print(f" Address: {info.parsed_addresses()[0]}:{info.port}")
print(f" Properties: {info.properties}")
def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None:
print(f"Service removed: {name}")
def update_service(self, zc: Zeroconf, type_: str, name: str) -> None:
print(f"Service updated: {name}")
# Start browsing
zc = Zeroconf()
listener = MyListener()
browser = ServiceBrowser(zc, "_http._tcp.local.", listener)
try:
input("Press enter to exit...\n")
finally:
zc.close()
Browse Multiple Service Types
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener
class MultiServiceListener(ServiceListener):
def __init__(self):
self.services = {}
def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
info = zc.get_service_info(type_, name)
if info:
self.services[name] = {
'type': type_,
'addresses': info.parsed_addresses(),
'port': info.port,
'properties': info.properties
}
zc = Zeroconf()
listener = MultiServiceListener()
# Browse multiple service types
service_types = [
"_http._tcp.local.",
"_ssh._tcp.local.",
"_mqtt._tcp.local."
]
browsers = [
ServiceBrowser(zc, stype, listener)
for stype in service_types
]
# Let discovery run
import time
time.sleep(5)
# Access discovered services
for name, info in listener.services.items():
print(f"{name}: {info}")
zc.close()
Resolving Service Details
from zeroconf import Zeroconf
def resolve_service(service_type: str, service_name: str) -> dict:
"""Resolve a specific service by name."""
zc = Zeroconf()
try:
full_name = f"{service_name}.{service_type}"
info = zc.get_service_info(service_type, full_name, timeout=3000)
if info:
return {
'name': service_name,
'addresses': info.parsed_addresses(),
'port': info.port,
'server': info.server,
'properties': {
k.decode(): v.decode() if isinstance(v, bytes) else v
for k, v in info.properties.items()
}
}
return None
finally:
zc.close()
# Usage
result = resolve_service("_http._tcp.local.", "My Web Server")
if result:
print(f"Connect to: {result['addresses'][0]}:{result['port']}")
Registering and Advertising Services
Basic Service Registration
from zeroconf import Zeroconf, ServiceInfo
import socket
def register_service():
"""Register a service on the local network."""
zc = Zeroconf()
# Get local IP address
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
# Create service info
service_info = ServiceInfo(
type_="_http._tcp.local.",
name="My API Server._http._tcp.local.",
port=8080,
properties={
'version': '1.0',
'path': '/api/v1',
'secure': 'false'
},
server=f"{hostname}.local.",
addresses=[socket.inet_aton(local_ip)]
)
# Register the service
zc.register_service(service_info)
print(f"Service registered: {service_info.name}")
try:
input("Press enter to unregister...\n")
finally:
zc.unregister_service(service_info)
zc.close()
register_service()
Service with Multiple Addresses
from zeroconf import Zeroconf, ServiceInfo, IPVersion
import socket
def get_all_ip_addresses():
"""Get all local IP addresses."""
addresses = []
for info in socket.getaddrinfo(socket.gethostname(), None):
addr = info[4][0]
if not addr.startswith('127.'):
addresses.append(socket.inet_aton(addr))
return addresses
service_info = ServiceInfo(
type_="_http._tcp.local.",
name="Multi-Address Server._http._tcp.local.",
port=8080,
properties={'version': '2.0'},
server=f"{socket.gethostname()}.local.",
addresses=get_all_ip_addresses()
)
# Register with specific IP version preference
zc = Zeroconf(ip_version=IPVersion.All)
zc.register_service(service_info)
Updating Service Properties
from zeroconf import Zeroconf, ServiceInfo
import socket
zc = Zeroconf()
hostname = socket.gethostname()
addresses = [socket.inet_aton(socket.gethostbyname(hostname))]
service_info = ServiceInfo(
type_="_http._tcp.local.",
name="Dynamic Service._http._tcp.local.",
port=8080,
properties={'status': 'starting'},
server=f"{hostname}.local.",
addresses=addresses
)
zc.register_service(service_info)
# Update properties later. ServiceInfo.properties is read-only on modern
# zeroconf; build a fresh ServiceInfo (same name) and pass it to update_service.
service_info = ServiceInfo(
type_="_http._tcp.local.",
name="Dynamic Service._http._tcp.local.",
port=8080,
properties={'status': 'ready', 'connections': '0'},
server=f"{hostname}.local.",
addresses=addresses
)
zc.update_service(service_info)
# Clean up
zc.unregister_service(service_info)
zc.close()
Async Patterns with asyncio
Async Service Browser
import asyncio
from zeroconf import IPVersion
from zeroconf.asyncio import AsyncZeroconf, AsyncServiceBrowser, AsyncServiceInfo
class AsyncListener:
def __init__(self):
self.found_services = []
def add_service(self, zc, type_: str, name: str) -> None:
self.found_services.append((type_, name))
def remove_service(self, zc, type_: str, name: str) -> None:
pass
def update_service(self, zc, type_: str, name: str) -> None:
pass
async def discover_services(service_type: str, duration: float = 5.0):
"""Async service discovery."""
aiozc = AsyncZeroconf(ip_version=IPVersion.V4Only)
listener = AsyncListener()
browser = AsyncServiceBrowser(
aiozc.zeroconf,
service_type,
listener
)
await asyncio.sleep(duration)
# Resolve all found services
services = []
for type_, name in listener.found_services:
info = AsyncServiceInfo(type_, name)
await info.async_request(aiozc.zeroconf, 3000)
if info.addresses:
services.append({
'name': name,
'addresses': info.parsed_addresses(),
'port': info.port
})
await aiozc.async_close()
return services
# Run async discovery
async def main():
services = await discover_services("_http._tcp.local.")
for svc in services:
print(f"Found: {svc['name']} at {svc['addresses'][0]}:{svc['port']}")
asyncio.run(main())
Async Service Registration
import asyncio
from zeroconf.asyncio import AsyncZeroconf, AsyncServiceInfo
import socket
async def register_async_service():
"""Register service using async API."""
aiozc = AsyncZeroconf()
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
service_info = AsyncServiceInfo(
type_="_http._tcp.local.",
name="Async API._http._tcp.local.",
port=8080,
properties={'async': 'true'},
server=f"{hostname}.local.",
addresses=[socket.inet_aton(local_ip)]
)
await aiozc.async_register_service(service_info)
print("Service registered")
# Keep service alive
await asyncio.sleep(60)
await aiozc.async_unregister_service(service_info)
await aiozc.async_close()
asyncio.run(register_async_service())
Integration with FastAPI
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
from zeroconf.asyncio import AsyncZeroconf, AsyncServiceInfo
import socket
aiozc = None
service_info = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global aiozc, service_info
# Startup: register service
aiozc = AsyncZeroconf()
hostname = socket.gethostname()
service_info = AsyncServiceInfo(
type_="_http._tcp.local.",
name=f"FastAPI Server._http._tcp.local.",
port=8000,
properties={'framework': 'fastapi', 'version': '1.0'},
server=f"{hostname}.local.",
addresses=[socket.inet_aton(socket.gethostbyname(hostname))]
)
await aiozc.async_register_service(service_info)
yield
# Shutdown: unregister service
await aiozc.async_unregister_service(service_info)
await aiozc.async_close()
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Discoverable API"}
Network Interface Handling
Binding to Specific Interfaces
from zeroconf import Zeroconf, InterfaceChoice, IPVersion
import netifaces
# Use all interfaces (default)
zc = Zeroconf(interfaces=InterfaceChoice.All)
# Use only default interface
zc = Zeroconf(interfaces=InterfaceChoice.Default)
# Specify particular interfaces by IP
zc = Zeroconf(interfaces=['192.168.1.100', '10.0.0.50'])
# IPv4 only
zc = Zeroconf(ip_version=IPVersion.V4Only)
# IPv6 only
zc = Zeroconf(ip_version=IPVersion.V6Only)
# Both IPv4 and IPv6
zc = Zeroconf(ip_version=IPVersion.All)
Discovering Available Interfaces
import netifaces
def get_network_interfaces():
"""Get all available network interfaces with their addresses."""
interfaces = {}
for iface in netifaces.interfaces():
addrs = netifaces.ifaddresses(iface)
# Get IPv4 addresses
if netifaces.AF_INET in addrs:
ipv4 = [addr['addr'] for addr in addrs[netifaces.AF_INET]]
interfaces[iface] = {'ipv4': ipv4}
# Get IPv6 addresses
if netifaces.AF_INET6 in addrs:
ipv6 = [addr['addr'].split('%')[0] for addr in addrs[netifaces.AF_INET6]]
if iface in interfaces:
interfaces[iface]['ipv6'] = ipv6
else:
interfaces[iface] = {'ipv6': ipv6}
return interfaces
# Filter for usable interfaces
def get_bindable_ips():
"""Get IP addresses suitable for zeroconf binding."""
ips = []
for iface, addrs in get_network_interfaces().items():
for ip in addrs.get('ipv4', []):
if not ip.startswith('127.'):
ips.append(ip)
return ips
# Use with Zeroconf
from zeroconf import Zeroconf
zc = Zeroconf(interfaces=get_bindable_ips())
Handling Interface Changes
from zeroconf import Zeroconf, ServiceInfo
import socket
import time
class ServiceManager:
"""Manage service registration with interface monitoring."""
def __init__(self, service_type: str, name: str, port: int):
self.service_type = service_type
self.name = name
self.port = port
self.zc = None
self.service_info = None
def _get_current_ip(self):
"""Get current primary IP address."""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(('8.8.8.8', 80))
return s.getsockname()[0]
finally:
s.close()
def register(self):
"""Register or re-register the service."""
if self.zc:
self.unregister()
self.zc = Zeroconf()
ip = self._get_current_ip()
self.service_info = ServiceInfo(
type_=self.service_type,
name=f"{self.name}.{self.service_type}",
port=self.port,
properties={'ip': ip},
server=f"{socket.gethostname()}.local.",
addresses=[socket.inet_aton(ip)]
)
self.zc.register_service(self.service_info)
return ip
def unregister(self):
"""Unregister the service."""
if self.zc and self.service_info:
self.zc.unregister_service(self.service_info)
self.zc.close()
self.zc = None
self.service_info = None
# Usage with periodic refresh
manager = ServiceManager("_http._tcp.local.", "My Server", 8080)
current_ip = manager.register()
try:
while True:
time.sleep(60)
new_ip = manager._get_current_ip()
if new_ip != current_ip:
print(f"IP changed: {current_ip} -> {new_ip}")
current_ip = manager.register()
finally:
manager.unregister()
Common Use Cases
IoT Device Discovery
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener
import json
class IoTDeviceListener(ServiceListener):
"""Discover IoT devices on the network."""
def __init__(self):
self.devices = {}
def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
info = zc.get_service_info(type_, name)
if info:
device_id = name.split('.')[0]
self.devices[device_id] = {
'type': type_,
'ip': info.parsed_addresses()[0] if info.addresses else None,
'port': info.port,
'properties': {
k.decode(): v.decode() if isinstance(v, bytes) else v
for k, v in info.properties.items()
}
}
print(f"IoT device found: {device_id}")
def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None:
device_id = name.split('.')[0]
if device_id in self.devices:
del self.devices[device_id]
print(f"IoT device removed: {device_id}")
def update_service(self, zc: Zeroconf, type_: str, name: str) -> None:
self.add_service(zc, type_, name)
# Discover common IoT service types
iot_services = [
"_hap._tcp.local.", # HomeKit
"_mqtt._tcp.local.", # MQTT
"_coap._udp.local.", # CoAP
"_http._tcp.local.", # HTTP APIs
]
zc = Zeroconf()
listener = IoTDeviceListener()
browsers = [ServiceBrowser(zc, stype, listener) for stype in iot_services]
import time
time.sleep(10)
print(f"\nDiscovered {len(listener.devices)} devices:")
print(json.dumps(listener.devices, indent=2))
zc.close()
Local Development Service Mesh
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener, ServiceInfo
import socket
from typing import Dict, Optional
class ServiceMesh:
"""Simple service mesh for local development."""
def __init__(self):
self.zc = Zeroconf()
self.services: Dict[str, dict] = {}
self.registered: Dict[str, ServiceInfo] = {}
self._browsers = []
def register(self, name: str, port: int, properties: dict = None):
"""Register a service in the mesh."""
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
info = ServiceInfo(
type_="_devmesh._tcp.local.",
name=f"{name}._devmesh._tcp.local.",
port=port,
properties=properties or {},
server=f"{hostname}.local.",
addresses=[socket.inet_aton(local_ip)]
)
self.zc.register_service(info)
self.registered[name] = info
return info
def discover(self, callback=None):
"""Start discovering services."""
class Listener(ServiceListener):
def __init__(self, mesh, cb):
self.mesh = mesh
self.callback = cb
def add_service(self, zc, type_, name):
info = zc.get_service_info(type_, name)
if info:
svc_name = name.split('.')[0]
self.mesh.services[svc_name] = {
'address': info.parsed_addresses()[0],
'port': info.port,
'properties': dict(info.properties)
}
if self.callback:
self.callback('added', svc_name)
def remove_service(self, zc, type_, name):
svc_name = name.split('.')[0]
if svc_name in self.mesh.services:
del self.mesh.services[svc_name]
if self.callback:
self.callback('removed', svc_name)
def update_service(self, zc, type_, name):
self.add_service(zc, type_, name)
browser = ServiceBrowser(
self.zc,
"_devmesh._tcp.local.",
Listener(self, callback)
)
self._browsers.append(browser)
def get_service(self, name: str) -> Optional[dict]:
"""Get a discovered service by name."""
return self.services.get(name)
def get_url(self, name: str, path: str = "") -> Optional[str]:
"""Get URL for a service."""
svc = self.get_service(name)
if svc:
return f"http://{svc['address']}:{svc['port']}{path}"
return None
def close(self):
"""Clean up resources."""
for info in self.registered.values():
self.zc.unregister_service(info)
self.zc.close()
# Usage
mesh = ServiceMesh()
# Register this service
mesh.register("api-gateway", 8000, {"version": "1.0"})
# Discover other services
mesh.discover(lambda action, name: print(f"{action}: {name}"))
import time
time.sleep(5)
# Connect to discovered service
auth_url = mesh.get_url("auth-service", "/api/auth")
if auth_url:
print(f"Auth service at: {auth_url}")
mesh.close()
Network Printer Discovery
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener
class PrinterDiscovery(ServiceListener):
"""Discover network printers."""
def __init__(self):
self.printers = []
def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
info = zc.get_service_info(type_, name)
if info:
printer = {
'name': name.replace('._ipp._tcp.local.', ''),
'address': info.parsed_addresses()[0] if info.addresses else None,
'port': info.port,
'model': info.properties.get(b'ty', b'').decode(),
'uri': f"ipp://{info.parsed_addresses()[0]}:{info.port}/ipp/print"
}
self.printers.append(printer)
print(f"Found printer: {printer['name']} - {printer['model']}")
def remove_service(self, zc, type_, name):
pass
def update_service(self, zc, type_, name):
pass
zc = Zeroconf()
listener = PrinterDiscovery()
# Browse for IPP printers
browser = ServiceBrowser(zc, "_ipp._tcp.local.", listener)
import time
time.sleep(5)
for printer in listener.printers:
print(f"\nPrinter: {printer['name']}")
print(f" URI: {printer['uri']}")
zc.close()
Performance Considerations
Connection Pooling and Reuse
from zeroconf import Zeroconf
from contextlib import contextmanager
class ZeroconfPool:
"""Reusable Zeroconf instance pool."""
_instance = None
_ref_count = 0
@classmethod
@contextmanager
def get_instance(cls):
"""Get a shared Zeroconf instance."""
if cls._instance is None:
cls._instance = Zeroconf()
cls._ref_count += 1
try:
yield cls._instance
finally:
cls._ref_count -= 1
if cls._ref_count == 0:
cls._instance.close()
cls._instance = None
# Usage - instance is shared and properly cleaned up
with ZeroconfPool.get_instance() as zc:
info = zc.get_service_info("_http._tcp.local.", "MyService._http._tcp.local.")
Optimising Service Resolution
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener
import time
class BatchResolver(ServiceListener):
"""Batch resolve services to reduce network traffic."""
def __init__(self, zc: Zeroconf):
self.zc = zc
self.pending = []
self.resolved = {}
def add_service(self, zc, type_: str, name: str) -> None:
self.pending.append((type_, name))
def remove_service(self, zc, type_, name):
pass
def update_service(self, zc, type_, name):
pass
def resolve_all(self, timeout: int = 3000):
"""Resolve all pending services."""
for type_, name in self.pending:
info = self.zc.get_service_info(type_, name, timeout=timeout)
if info:
self.resolved[name] = {
'addresses': info.parsed_addresses(),
'port': info.port
}
self.pending.clear()
return self.resolved
# Usage
zc = Zeroconf()
resolver = BatchResolver(zc)
browser = ServiceBrowser(zc, "_http._tcp.local.", resolver)
# Wait for discovery
time.sleep(3)
# Batch resolve all found services
services = resolver.resolve_all()
print(f"Resolved {len(services)} services")
zc.close()
Memory Management
from zeroconf import Zeroconf, ServiceBrowser, ServiceListener
import gc
class LightweightListener(ServiceListener):
"""Memory-efficient listener that stores minimal data."""
def __init__(self):
# Store only essential info as tuples (more memory efficient)
self.services = [] # [(name, ip, port), ...]
def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
info = zc.get_service_info(type_, name)
if info and info.addresses:
# Store minimal tuple instead of dict
self.services.append((
name.split('.')[0],
info.parsed_addresses()[0],
info.port
))
def remove_service(self, zc, type_, name):
svc_name = name.split('.')[0]
self.services = [s for s in self.services if s[0] != svc_name]
def update_service(self, zc, type_, name):
pass
# Explicit cleanup
def discover_with_cleanup():
zc = Zeroconf()
listener = LightweightListener()
browser = ServiceBrowser(zc, "_http._tcp.local.", listener)
import time
time.sleep(5)
results = list(listener.services)
# Clean up
zc.close()
del browser, listener, zc
gc.collect()
return results
Timeout Configuration
from zeroconf import Zeroconf
zc = Zeroconf()
# Default timeout is 3000ms
info = zc.get_service_info(
"_http._tcp.local.",
"MyService._http._tcp.local.",
timeout=5000 # 5 second timeout for slow networks
)
# Quick check with short timeout
info = zc.get_service_info(
"_http._tcp.local.",
"MyService._http._tcp.local.",
timeout=500 # 500ms for responsive services
)
Quick Reference
Common Operations
| Operation | Code |
|---|---|
| Create Zeroconf instance | zc = Zeroconf() |
| Browse services | ServiceBrowser(zc, "_type._tcp.local.", listener) |
| Get service info | zc.get_service_info(type_, name, timeout=3000) |
| Register service | zc.register_service(service_info) |
| Update service | zc.update_service(service_info) |
| Unregister service | zc.unregister_service(service_info) |
| Close instance | zc.close() |
ServiceInfo Parameters
ServiceInfo(
type_="_http._tcp.local.", # Service type (required)
name="MyService._http._tcp.local.", # Full service name (required)
port=8080, # Port number (required)
properties={'key': 'value'}, # TXT record properties
server="hostname.local.", # Server hostname
addresses=[socket.inet_aton(ip)], # IP addresses as bytes
host_ttl=4500, # Host record TTL
other_ttl=4500 # Other record TTL
)
ServiceListener Methods
class MyListener(ServiceListener):
def add_service(self, zc, type_, name): # Service discovered
def remove_service(self, zc, type_, name): # Service removed
def update_service(self, zc, type_, name): # Service updated
Common Service Types
| Type | Use Case |
|---|---|
_http._tcp.local. |
Web servers, REST APIs |
_https._tcp.local. |
Secure web servers |
_mqtt._tcp.local. |
IoT message brokers |
_hap._tcp.local. |
HomeKit devices |
_ssh._tcp.local. |
SSH servers |
_smb._tcp.local. |
File sharing |
_ipp._tcp.local. |
Printers |
Common Issues and Solutions
Service Not Discovered
Problem: Services are registered but not being discovered.
# Solution 1: Check firewall allows mDNS traffic
# UDP port 5353 must be open for multicast
# Solution 2: Ensure same network segment
# mDNS doesn't cross routers by default
# Solution 3: Try explicit interface binding
from zeroconf import Zeroconf
import socket
# Get the correct interface IP
local_ip = socket.gethostbyname(socket.gethostname())
zc = Zeroconf(interfaces=[local_ip])
# Solution 4: Increase discovery timeout
import time
time.sleep(10) # Allow more time for mDNS responses
Address Resolution Fails
Problem: parsed_addresses() returns empty list.
from zeroconf import Zeroconf
zc = Zeroconf()
info = zc.get_service_info(type_, name, timeout=3000)
# Check if info exists
if info is None:
print("Service not found")
elif not info.addresses:
print("No addresses resolved - try longer timeout")
info = zc.get_service_info(type_, name, timeout=10000)
else:
print(f"Address: {info.parsed_addresses()[0]}")
Duplicate Service Names
Problem: Registration fails due to name conflict.
import uuid
import socket
# Solution: Add unique identifier to service name
unique_id = uuid.uuid4().hex[:8]
hostname = socket.gethostname()
service_name = f"{hostname}-{unique_id}._http._tcp.local."
Memory Leak with Long-Running Applications
Problem: Memory usage grows over time.
# Solution: Properly close and recreate Zeroconf instances
from zeroconf import Zeroconf, ServiceBrowser
class ManagedDiscovery:
def __init__(self):
self.zc = None
self.browser = None
def start(self, service_type, listener):
self.stop() # Clean up any existing
self.zc = Zeroconf()
self.browser = ServiceBrowser(self.zc, service_type, listener)
def stop(self):
if self.zc:
self.zc.close()
self.zc = None
self.browser = None
def __del__(self):
self.stop()
IPv6 Issues
Problem: IPv6 addresses cause connection problems.
from zeroconf import Zeroconf, IPVersion
# Use IPv4 only to avoid IPv6 issues
zc = Zeroconf(ip_version=IPVersion.V4Only)
# Or filter addresses when resolving
info = zc.get_service_info(type_, name)
if info:
# Get only IPv4 addresses
ipv4_addrs = [
addr for addr in info.parsed_addresses()
if '.' in addr # Simple IPv4 check
]
Docker/Container Networking
Problem: mDNS doesn't work in containers.
# Solution 1: Use host network mode
# docker run --network=host myapp
# Solution 2: Specify container's IP explicitly
from zeroconf import Zeroconf
import os
container_ip = os.environ.get('CONTAINER_IP', '172.17.0.2')
zc = Zeroconf(interfaces=[container_ip])
Service Properties Not Updating
Problem: Property updates not propagating.
from zeroconf import Zeroconf, ServiceInfo
# ServiceInfo.properties is read-only on modern zeroconf. To change properties,
# build a new ServiceInfo (same type/name/port/server/addresses) and pass it to
# update_service — the call is required to propagate the change.
service_info = ServiceInfo(
type_=service_info.type,
name=service_info.name,
port=service_info.port,
properties={'new': 'value'},
server=service_info.server,
addresses=service_info.addresses,
)
zc.update_service(service_info) # Required!
# For frequent updates, consider debouncing
from threading import Timer
class DebouncedUpdater:
def __init__(self, zc, service_info, delay=1.0):
self.zc = zc
self.service_info = service_info
self.delay = delay
self.timer = None
def update(self, properties):
info = self.service_info
self.service_info = ServiceInfo(
type_=info.type, name=info.name, port=info.port,
properties=properties, server=info.server,
addresses=info.addresses,
)
if self.timer:
self.timer.cancel()
self.timer = Timer(self.delay, self._do_update)
self.timer.start()
def _do_update(self):
self.zc.update_service(self.service_info)
Async Event Loop Conflicts
Problem: Async zeroconf conflicts with existing event loop.
import asyncio
from zeroconf.asyncio import AsyncZeroconf
# Solution: Use existing event loop
async def setup_zeroconf():
# Get current event loop
loop = asyncio.get_event_loop()
# AsyncZeroconf uses the current loop automatically
aiozc = AsyncZeroconf()
# ... use aiozc ...
await aiozc.async_close()
# If running in Jupyter or similar
import nest_asyncio
nest_asyncio.apply()
asyncio.run(setup_zeroconf())