asyncua: Pure Python OPC-UA Client and Server Library

1.1.8 · active · verified Mon Apr 13

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

Install

Imports

Quickstart

This quickstart demonstrates how to establish a basic OPC-UA client connection, retrieve the root node, and gracefully disconnect. It uses an environment variable for the server URL, which is a common practice for flexible deployment. Remember that `asyncua` is built on `asyncio`, so all client/server operations must be `await`ed within an `async` function.

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())

view raw JSON →