asyncua: Pure Python OPC-UA Client and Server Library
asyncua is a comprehensive, pure Python library providing both OPC-UA client and server capabilities, designed for asynchronous operations. It enables Python applications to communicate with industrial automation systems using the OPC-UA protocol. Currently at version 1.1.8, the library maintains an active development pace, with frequent bug fixes and feature enhancements, including recent beta releases for upcoming major versions.
Warnings
- breaking Starting with version 1.2.0 (currently in beta), asyncua removes support for Python versions older than 3.10. Users on Python 3.9 or earlier must upgrade their Python environment before migrating to asyncua 1.2.x or later.
- gotcha asyncua is an asyncio-based library. All network operations (connect, read, write, subscribe, disconnect, etc.) are coroutines and must be properly `await`ed within an `async` function. Forgetting `await` will result in coroutine objects not being run, leading to unexpected behavior or resource leaks.
- gotcha Always ensure to call `await client.disconnect()` or use `async with Client(...)` for client connections to gracefully close the connection and release resources. Failing to do so can leave dangling connections and consume system resources unnecessarily.
- gotcha When interacting with OPC-UA values, ensure correct handling of `asyncua.ua.Variant` and `asyncua.ua.VariantType`. Directly assigning Python types might work for simple cases, but explicit `ua.Variant` usage is often required for specific OPC-UA data types, especially when writing values.
Install
-
pip install asyncua -
pip install --pre asyncua
Imports
- Client
from asyncua import Client
- Server
from asyncua import Server
- ua
from asyncua import ua
- Node
from asyncua.common.node import Node
Quickstart
import asyncio
from asyncua import Client, ua
import os
async def main():
# Replace with your OPC-UA server endpoint
server_url = os.environ.get('OPCUA_SERVER_URL', 'opc.tcp://localhost:4840/freeopcua/server/')
# Client connection example
client = Client(url=server_url)
try:
await client.connect()
print(f"Connected to OPC-UA server at {server_url}")
# Get the root node
root = client.get_root_node()
print(f"Root node: {root}")
# Example: Read a variable (replace with actual node ID)
# For example, to read a variable under Objects/MyObject/MyVariable
# node_id_str = "ns=2;i=1001" # Example NodeId
# my_variable = await client.get_node(node_id_str)
# value = await my_variable.read_value()
# print(f"Value of {node_id_str}: {value}")
# You can browse nodes, subscribe to data changes, write values, etc.
except Exception as e:
print(f"Error connecting or interacting with server: {e}")
finally:
print("Disconnecting from OPC-UA server...")
await client.disconnect()
if __name__ == '__main__':
asyncio.run(main())