OPC UA を Python で扱う完全ガイド — asyncua で工場の標準プロトコルに対応

1. はじめに — なぜ今 OPC UA なのか

製造業のスマートファクトリ化が進むにつれ、PLC・SCADA・MES・ERP・クラウドの各層を横断するデータ流通が必要になりました。レガシーな Modbus では情報モデルや型情報を持てず、メーカー固有プロトコルでは横断できません。そこで採用が進んでいるのが OPC UA（仕様は OPC Foundation 公式仕様）です。

2026 年時点で主要 PLC メーカー（シーメンス、ロックウェル、三菱、オムロン、キーエンス）はすべて OPC UA サーバー機能を標準あるいはオプションで提供しています（各社の対応状況は OPC Foundation の加盟企業・製品リストで確認可能）。「Modbus の次に学ぶべき産業プロトコル」を 1 つ挙げるなら、間違いなく OPC UA です。

本記事は asyncua（旧 opcua-asyncio）を使って、自宅 PC のみで OPC UA を体験する構成です。

2. OPC UA の基本概念

2.1 アドレス空間とノード

OPC UA は階層的な「アドレス空間（Address Space）」上に「ノード（Node）」を配置します。各ノードには NodeId（識別子）、BrowseName（表示名）、NodeClass（種別）が付き、変数ノードであれば値と型情報を持ちます。

2.2 主要なノード種別

• Variable: 値を持つノード（温度、圧力、稼働率など）
• Object: ノードをまとめるコンテナ（「設備A」「ライン1」など）
• Method: サーバー側で実行できる関数（「再起動」「リセット」など）
• ObjectType / VariableType: 型定義（インスタンス生成のテンプレート）

2.3 通信パターン

• Read / Write: 同期的な値の読み書き
• Subscribe: 値の変化をサーバーから自動通知させる
• Method Call: サーバー側のメソッドを呼ぶ

「定周期で読みに行く」のではなく、「変化があったときだけ通知してもらう」サブスクリプションは、Modbus にはない大きな利点です。

3. 環境構築

python -m pip install asyncua

社内プロキシや閉域網で pip install が通らない場合は、別記事の Modbus 編・PyInstaller 編で触れた「--proxy オプション」「別 PC で pip download → 持ち込み」の手順がそのまま使えます。

asyncua は asyncio ベースの実装で、サーバー / クライアント両方を提供します。古い opcua（同期版）はメンテナンスがほぼ止まっているため、新規は asyncua 一択です。ライセンスは LGPLv3+ で、動的リンクで利用する限り無償で商用利用できます（ただし PyInstaller の --onefile での配布形態など、LGPL の解釈は W18「PyQt6/PySide6」記事で触れたのと同じ論点があります。実務では自社の法務・顧問弁護士に確認してください）。Python 対応バージョンは 3.8 以上を想定しています。

本記事のサンプルは async def / await / asyncio.run() が頻出します。asyncio（非同期処理）に馴染みがない場合は、いまは「関数の頭に async、通信の行の前に await を付け、いちばん外側を asyncio.run(main()) で回す」というお約束と捉えて構いません。OPC UA は 1 本の接続で読み書き・通知を同時にさばくため、非同期モデルと相性が良く、asyncua もこの形を前提にしています。仕組みの詳細は公式の asyncio ドキュメントを参照してください。

4. シミュレータサーバーを立ち上げる

まずは検証用のサーバーを 1 つ立ち上げます。「設備A」というオブジェクトの下に温度・圧力・稼働状態の変数を作る、シンプルな構成です。

# server.py
import asyncio
import random
from asyncua import Server, ua


async def main() -> None:
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/genbapy/server/")
    server.set_server_name("GenbaPy OPC UA Server")

    # 名前空間を登録
    uri = "https://genbapy.dev/opcua/sim"
    idx = await server.register_namespace(uri)

    # 「設備A」オブジェクトを作成
    objects = server.nodes.objects
    eq_a = await objects.add_object(idx, "EquipmentA")

    # 変数を 3 つ追加
    temperature = await eq_a.add_variable(idx, "Temperature", 25.0)
    pressure = await eq_a.add_variable(idx, "Pressure", 1.0)
    running = await eq_a.add_variable(idx, "Running", True)

    # クライアントから書き込めるようにする
    await temperature.set_writable()
    await pressure.set_writable()
    await running.set_writable()

    print(f"OPC UA server started at {server.endpoint}")
    print(f"namespace index: {idx}")

    async with server:
        while True:
            # サーバー側でランダムに値を更新
            await temperature.write_value(25.0 + random.uniform(-2.0, 2.0))
            await pressure.write_value(1.0 + random.uniform(-0.05, 0.05))
            await asyncio.sleep(1.0)


if __name__ == "__main__":
    asyncio.run(main())

このスクリプトを実行すると、opc.tcp://localhost:4840 で OPC UA サーバーが起動します。1 秒ごとに温度・圧力がランダムに更新されます。

python server.py

このサーバーは起動しっぱなしにします。ターミナルを 1 つ占有するので、次章以降のクライアントは別のターミナルをもう 1 枚開いて実行してください（VS Code なら「＋」でターミナル分割、コマンドプロンプトならもう 1 窓）。Ctrl+C で停止できます。

なお set_endpoint("opc.tcp://0.0.0.0:4840/...") の 0.0.0.0 は「この PC のすべてのネットワークカードで待ち受ける」指定で、同じ LAN の別 PC からも接続できてしまいます。自分の PC の中だけで試すうちは 127.0.0.1（localhost 限定）にしておくと安全です。社内 LAN に出すときは、待ち受けアドレス・ファイアウォール・後述のセキュリティ設定をセットで見直してください。

5. クライアントから接続して読み書きする

サーバーを起動したまま、別ターミナルで次を実行します。

# client_basic.py
import asyncio
from asyncua import Client


async def main() -> None:
    url = "opc.tcp://localhost:4840/genbapy/server/"
    async with Client(url=url) as client:
        # 名前空間 URI からインデックスを取得
        idx = await client.get_namespace_index("https://genbapy.dev/opcua/sim")

        # アドレス指定でノードを取得
        temperature = await client.nodes.root.get_child(
            ["0:Objects", f"{idx}:EquipmentA", f"{idx}:Temperature"]
        )

        # 読み取り
        value = await temperature.read_value()
        print(f"temperature: {value}")

        # 書き込み
        await temperature.write_value(30.0)
        print(f"after write: {await temperature.read_value()}")


if __name__ == "__main__":
    asyncio.run(main())

python client_basic.py

OPC UA は「ブラウズパス」でノードを指定する方式です。XML パスのような書き方になりますが、最初は分かりにくいので「アドレス空間ブラウザ」を使って構造を確認するのがおすすめです（後述）。get_child に渡す各要素の 0: や {idx}: は名前空間インデックスで、標準ノード（Objects など）は 0、自分で登録したノードは register_namespace() が返した番号（このサーバーでは通常 2）になります。

6. サブスクリプションで変化通知を受け取る

OPC UA の真骨頂は「サーバー側の値変化をクライアントに自動で通知させる」サブスクリプション機能です。

# client_subscribe.py
import asyncio
from asyncua import Client, ua


class SubHandler:
    """変化通知を受け取るハンドラ。

    asyncua が datachange_notification を呼んでくれる。
    """

    def datachange_notification(self, node, val, data) -> None:
        print(f"data change: {node} = {val}")

    def event_notification(self, event) -> None:
        print(f"event: {event}")


async def main() -> None:
    url = "opc.tcp://localhost:4840/genbapy/server/"
    async with Client(url=url) as client:
        idx = await client.get_namespace_index("https://genbapy.dev/opcua/sim")
        temperature = await client.nodes.root.get_child(
            ["0:Objects", f"{idx}:EquipmentA", f"{idx}:Temperature"]
        )
        pressure = await client.nodes.root.get_child(
            ["0:Objects", f"{idx}:EquipmentA", f"{idx}:Pressure"]
        )

        handler = SubHandler()
        # 100ms ごとに変化チェック、変化があれば通知
        sub = await client.create_subscription(100, handler)
        await sub.subscribe_data_change([temperature, pressure])

        # 1 分間サブスクリプション継続
        await asyncio.sleep(60)


if __name__ == "__main__":
    asyncio.run(main())

python client_subscribe.py

定周期ポーリングと違い、変化がなければ通信は発生しません。多数のノードを監視するときに、ネットワークと CPU の負荷を大きく下げられます。server.py が 1 秒ごとに温度・圧力を更新しているので、実行すると data change: ... = 24.7... のような通知が流れ続けます。Ctrl+C で止めてください。

7. アドレス空間をブラウズする

サーバーのアドレス空間がどう構成されているか分からないときは、ブラウズして表示する一行スクリプトが便利です。

async def browse(node, depth: int = 0) -> None:
    name = await node.read_browse_name()
    print(f"{'  ' * depth}- {name}")
    for child in await node.get_children():
        await browse(child, depth + 1)


# 使い方
async with Client(url=url) as client:
    await browse(client.nodes.objects)

UI 付きのアドレス空間ブラウザが必要なら、Unified Automation 社の UaExpert が OPC UA クライアントの定番です。無償でダウンロードできますが、公式の Software License Agreement（SLA） は「UaExpert は評価用途に限る（evaluation purpose only）」と規定しており、目安としては評価期間 3 ヶ月・期間経過後は削除必須と読める内容です。業務・商用での恒久使用は原則不可、有償版の購入が必要になります。動作検証用に短期間だけ使うのは問題ありませんが、本番運用のクライアントとして常設する場合は必ず商用ライセンスを購入してください。純粋にコードだけで済ませたいなら、上記の browse() スクリプトでも構造は十分たどれます。

なお上記の browse() スクリプトは断片で、単体では動きません。使うときは §5 の client_basic.py の async with Client(url=url) as client: ブロックの中で await browse(client.nodes.objects) と呼び出してください。

8. セキュリティの設定

OPC UA は既定で「暗号化なし・認証なし」モード（None）でも動きます。まずは前章までの None モードで動作を掴み、慣れてから署名・暗号化を足すのがおすすめです。本番環境では署名＋暗号化（SignAndEncrypt）の有効化が推奨されます。

8.1 証明書と秘密鍵を生成する（一度だけ）

暗号化には証明書（.der）と秘密鍵（.pem）が要ります。asyncua の暗号機能は追加パッケージなので、先に crypto エクストラを入れます。

python -m pip install "asyncua[crypto]"

次のスクリプトで、サーバー用とクライアント用の自己署名証明書を生成します。asyncua 同梱の setup_self_signed_certificate を使うので、OpenSSL コマンドは不要です。

# gen_certs.py — 検証用の自己署名証明書を生成（一度だけ実行）
import asyncio
import socket
from pathlib import Path

from cryptography.x509.oid import ExtendedKeyUsageOID
from asyncua.crypto.cert_gen import setup_self_signed_certificate


async def main() -> None:
    host = socket.gethostname()
    subject = {
        "countryName": "JP",
        "stateOrProvinceName": "Tokyo",
        "localityName": "Tokyo",
        "organizationName": "GenbaPy",
    }

    # サーバー証明書（SERVER_AUTH）
    await setup_self_signed_certificate(
        Path("server-key.pem"),
        Path("server-cert.der"),
        f"urn:{host}:genbapy:server",  # app_uri（後で一致必須）
        host,
        [ExtendedKeyUsageOID.SERVER_AUTH],
        subject,
    )
    # クライアント証明書（CLIENT_AUTH）
    await setup_self_signed_certificate(
        Path("client-key.pem"),
        Path("client-cert.der"),
        f"urn:{host}:genbapy:client",
        host,
        [ExtendedKeyUsageOID.CLIENT_AUTH],
        subject,
    )
    print("generated: server-cert.der / server-key.pem / client-cert.der / client-key.pem")


if __name__ == "__main__":
    asyncio.run(main())

python gen_certs.py

第 3 引数の app_uri（urn:...）は証明書に埋め込まれる識別子で、後述のクライアント設定 application_uri と一字一句一致させる必要があります。ズレると接続時に BadCertificateUriInvalid で弾かれます。秘密鍵（*-key.pem）が既にあると再生成しない作りなので、作り直したいときは .pem と .der を消してから再実行してください。秘密鍵はパスワードと同じ扱いで、Git にコミットしないよう .gitignore に *.pem を入れておきましょう。

8.2 サーバー側の設定

§4 の server.py の await server.init() の直後に、セキュリティポリシーと証明書の読み込みを足します。

server = Server()
await server.init()
server.set_endpoint("opc.tcp://0.0.0.0:4840/genbapy/server/")

# 暗号化を有効化（None も残すと平文接続も受け付ける）
server.set_security_policy([
    ua.SecurityPolicyType.NoSecurity,
    ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt,
])
await server.load_certificate("server-cert.der")
await server.load_private_key("server-key.pem")

平文接続を一切許さないなら、NoSecurity をリストから外します。

8.3 クライアント側の設定

セキュリティ設定は接続（async with に入る）の 前 に呼びます。接続後に呼んでも、そのセッションには適用されません。ここが OPC UA でつまずきやすい点です。

# client_secure.py
import asyncio
import socket
from asyncua import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256


async def main() -> None:
    host = socket.gethostname()
    url = "opc.tcp://localhost:4840/genbapy/server/"

    client = Client(url=url)
    # 証明書生成時の app_uri と一致させる
    client.application_uri = f"urn:{host}:genbapy:client"
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate="client-cert.der",
        private_key="client-key.pem",
        server_certificate="server-cert.der",
    )

    # ここで初めて接続する
    async with client:
        idx = await client.get_namespace_index("https://genbapy.dev/opcua/sim")
        temperature = await client.nodes.root.get_child(
            ["0:Objects", f"{idx}:EquipmentA", f"{idx}:Temperature"]
        )
        print(await temperature.read_value())


if __name__ == "__main__":
    asyncio.run(main())

server_certificate にはサーバーの公開証明書（server-cert.der）を渡します。クライアントがサーバー本体を検証するためです。逆に、サーバー側もクライアント証明書を「信頼済み」として受け入れる必要があります。自己署名同士だと既定の証明書検証で拒否されることがあり、その場合はお互いの証明書を信頼フォルダに置くか、検証条件を明示する設定が要ります。ここは実装が長くなるので、公式リポジトリの server-with-encryption.py / client-with-encryption.py（CertificateValidator と TrustStore の使い方）を出発点にするのが確実です。

本番運用では「証明書の発行・配布・更新」のフローまで設計に含めてください。自己署名で運用するか、社内 CA を立てるかは、規模と監査要件次第です。

9. 認証と権限管理

OPC UA は匿名 / ユーザー名 + パスワード / 証明書認証の 3 種を組み合わせられます。読み取り専用ユーザーと書き込み可能ユーザーを分けるのが基本パターンです。asyncua では UserManager を継承したクラスを作り、サーバー生成時に渡します。

# サーバー側のユーザー管理
from asyncua import Server
from asyncua.crypto.permission_rules import User, UserRole
from asyncua.server.user_managers import UserManager


class GenbaPyUserManager(UserManager):
    """ユーザー名 + パスワードで認証する簡易マネージャ。

    get_user はサーバー内部から呼ばれ、認証OKなら User を、
    NG なら None を返す（None を返すと接続が拒否される）。
    """

    def get_user(self, iserver, username=None, password=None, certificate=None):
        # ⚠️ 実運用ではパスワード直書き禁止。環境変数から読み込み、passlib 等でハッシュ照合する。
        if username == "viewer" and password == "CHANGE_ME_VIEWER":
            return User(role=UserRole.User, name=username)
        if username == "admin" and password == "CHANGE_ME_ADMIN":
            return User(role=UserRole.Admin, name=username)
        return None


# UserManager は Server 生成時に渡す（init より前）
server = Server(user_manager=GenbaPyUserManager())
await server.init()

User / UserRole の import 元は asyncua.crypto.permission_rules です（バージョンで移動した経緯があり、古い記事の from asyncua.server.users import ... や server.user_manager.set_user_manager(...) は現行版では動きません）。UserRole には Admin / User / Anonymous があります。

クライアントからは、接続前にユーザー名とパスワードを設定します（set_security と同じく async with の前）。

client = Client(url=url)
client.set_user("viewer")
client.set_password("CHANGE_ME_VIEWER")  # 実運用は環境変数から読み込む
async with client:
    ...

⚠️ ID／パスワード認証は必ず暗号化（Sign 以上）と併用してください。§8 の暗号化なし（None）モードだと、ユーザー名とパスワードがネットワーク上を平文で流れます。工場 LAN でも盗聴・なりすましのリスクがあるため、認証を使うなら Basic256Sha256_SignAndEncrypt をセットにするのが原則です。

書き込みを伴う操作（しきい値変更、レシピ書き込みなど）は UserRole.Admin 経由でしか通さない設計にすると、現場での誤操作リスクが下がります。ただし「User ロールの書き込みを実際に拒否する」には、ロールごとの権限ルール（asyncua.crypto.permission_rules）の設定も必要です。まずは「認証で入口を絞る」ところから始め、権限の作り込みは要件が固まってからで十分です。なお、上のコードはパスワードを直書きしていますが検証用です。実運用では環境変数やハッシュ照合にし、平文比較は避けてください。

10. 実装パターン

パターン A: PLC データを定周期で取り、CSV に書き出すブリッジ

シーメンス S7-1500 などの OPC UA 対応 PLC から、必要な変数だけをサブスクリプションで監視し、変化時に CSV へ追記するパターン。

パターン B: 複数 PLC のデータを 1 つの OPC UA サーバーに集約する

古い PLC（Modbus / FINS / S7Comm）からデータを読み取って、自前の OPC UA サーバーで標準化する「ゲートウェイ」パターン。MES からは OPC UA だけ見ればよくなる。

パターン C: クラウドへのアップロードゲートウェイ

OPC UA でサブスクリプションして、AWS IoT / Azure IoT Hub に MQTT でアップロードする「現場ゲートウェイ」。Raspberry Pi クラスのエッジデバイスでも十分動く負荷です。

11. 24 時間運用のチェックリスト

• 再接続: asyncua のクライアントは切断時に例外。try/except + バックオフ再接続を最外で実装
• サブスクリプション再生成: 接続が切れたらサブスクリプションも消えるため、再接続後に作り直す
• セッション維持: session_timeout を仕様に合わせる。短すぎるとアイドル中に切れる
• セキュリティ証明書の有効期限: 自己署名でも有効期限はある。期限切れアラートを別系統で持つ
• ログ: asyncua の logging 出力を WARNING 以上で残し、INFO レベルは別ファイルに分ける
• ノード変更への追従: PLC 側の構成変更でノードが消えると例外。起動時にブラウズして実存確認

12. Modbus vs OPC UA — 選定判断のポイント

OPC UA の学習が済んだところで、実務でよく問われる「Modbus と OPC UA、どちらを採用すべきか」の判断軸を整理します。

判断フロー: ①既存 PLC が OPC UA を喋れないなら Modbus 一択（あるいは メーカー固有プロトコル）→ ②複数ベンダ・複数設備を統合する必要があるなら OPC UA → ③データ配布を MES・クラウドまで伸ばすなら OPC UA + MQTT のブリッジ。「軽い」ものから「重い」ものへ、必要が出た段階で追加していくのが現実的です。PLC と Python で通信する方法 と本記事は、単一設備の書き込み（Modbus / メーカー固有）から始めて統合層に OPC UA を置く、という段階設計を前提に住み分けています。

13. おわりに

OPC UA は仕様が大きく初学者にはハードルがありますが、「アドレス空間・読み書き・サブスクリプション」さえ押さえれば、実用レベルの実装は一気に書けます。本記事のサンプルは自宅 PC のみで再現できる構成にしているため、実機がない段階でも先行して経験を積めます。筆者環境（Python 3.11 / asyncua 1.1 系 / Windows 11）で通しで動作確認しました。

Modbus が古典的な「軽量・シンプル」、OPC UA は「重厚・標準・拡張可能」。現場の要件に応じて両方を扱える Python エンジニアは、製造業 IT 領域で希少な存在になれます。

⚠️ 実機接続時の注意: 本記事のサンプルはローカル検証用（自己署名証明書・ダミーデータ）です。実際の PLC・産業機器の OPC UA サーバに接続する場合は、必ず読み出し専用モードで検証し、書き込み（メソッド呼び出し・変数書き込み）は安全インターロック・計画停止のうえで実施してください。24 時間稼働機器の停止・書き込みは、自社の安全基準と保全側の承認を経ることが前提です。本記事のコードによる損害・事故について筆者・GenbaPy は責任を負いません。証明書・秘密鍵の管理、ユーザー認証情報の保管も、社内セキュリティポリシーに従ってください。