Claude Media
Claude CodeでOPC UAのasyncuaサーバーを立て設備値を読む検証手順

Claude CodeでOPC UAのasyncuaサーバーを立て設備値を読む検証手順

実機の設備につながずに、Claude CodeでOPC UAのクライアントを書かせる手順です。asyncuaで模擬サーバーを立て、値の読み取りを終了コードで合否判定します。

OPC UAのクライアントをClaude Codeに書かせるとき、いちばん困るのは試す相手が実機になりやすいことです。結論から言うと、PythonライブラリのasyncuaでPC上に模擬の設備サーバーを立て、読み取りの確認を「終了コードで合否が返るスクリプト」にします。Claude Codeは実機に一度もつながないまま、書いて、動かして、直す反復を回せます。

この記事の例は、asyncua 2.1.0とPython 3.12.2で手元で動かしたものです。模擬サーバーには温度を持つポンプ1台を置き、読むのは値だけにします。書き込みを伴う操作は扱いません。

構成は2ファイルと1本の規約だけ

用意するのは、模擬サーバー、検証クライアント、Claude Codeへの規約の3つです。サーバーとクライアントは別プロセスで動かします。実機を相手にするときと同じ形で確かめるためです。

手順

検証ループの流れ

  1. 1

    模擬サーバーを起動する

    sim/server.pyが127.0.0.1でOPC UAサーバーを待ち受けます。

  2. 2

    検証クライアントを実行する

    sim/check_client.pyが値を読み、範囲外なら終了コード1で終わります。

  3. 3

    結果をClaudeに読ませる

    標準出力と終了コードを見て、Claudeがクライアント側のコードを直します。

asyncuaはasyncioベースの非同期なOPC UAのクライアントとサーバーのライブラリです。READMEには、高水準のクラスで数行でサーバーやクライアントが書けると説明があります。同期コード向けのラッパーも用意されています。

インストールはuvまたはpipで行い、PyPIの記載ではPython 3.10以上が必要です。

python3 -m venv .venv
.venv/bin/pip install asyncua

Claudeに守らせる規約をCLAUDE.mdに置く

通信コードの検証で事故になるのは、接続先の取り違えと意図しない書き込みです。この2点は、口頭の指示ではなくCLAUDE.mdの規約に落としておきます。CLAUDE.mdは公式のドキュメントでも、具体的で簡潔なほど守られやすいと説明されています。目安は1ファイル200行未満です。

## OPC UAの検証ルール
- 接続先は opc.tcp://127.0.0.1:4840/plant/ のみ。実機のURLを書かない
- クライアントは読み取り専用。write_value / call_method は使わない
- 検証は `.venv/bin/python sim/check_client.py` を実行し、終了コードで判断する
- サーバーは `sim/server.py` を別プロセスで起動してから検証する

ただし、CLAUDE.mdはあくまで文脈です。公式のドキュメントは、メモリは強制される設定ではなく、行動を止めたいときはPreToolUseフックを使うよう案内しています。強制したい部分は次の権限設定で担保します。

許可するコマンドを絞る

権限のルールは、拒否、確認、許可の順に評価されます。最初に一致したルールが結果を決めるため、広い拒否ルールに対して狭い許可ルールで例外を作ることはできません。この性質を使い、実行してよいコマンドだけを許可します。

{
  "permissions": {
    "allow": [
      "Bash(.venv/bin/python sim/check_client.py)",
      "Bash(.venv/bin/uals *)",
      "Bash(.venv/bin/uaread *)"
    ],
    "deny": [
      "Read(./.env)",
      "Read(./secrets/**)"
    ]
  }
}

サーバーの起動は、あえて許可に入れていません。常駐するプロセスは、初回は人が確認して許可する運用にすると、意図しないポートを開いていないか目で確かめられます。権限ルールの書式は、公式のドキュメントにあるBash(npm run *)のようなワイルドカードの形に従っています。

ualsとuareadは、asyncuaに含まれる読み取り系のコマンドです。書き込みを行うuawriteと、メソッドを呼ぶuacallは許可に入れません。

模擬の設備サーバーを書く

サーバーは、公式のminimalサンプルと同じ流れで書きます。サーバーを初期化し、エンドポイントを決め、名前空間を登録して、オブジェクトと変数を追加します。そのうえで、値を0.5秒ごとに更新し続けます。

import asyncio
import math
 
from asyncua import Server
 
ENDPOINT = "opc.tcp://127.0.0.1:4840/plant/"
NAMESPACE = "urn:example:plant-sim"
 
 
async def main():
    server = Server()
    await server.init()
    server.set_endpoint(ENDPOINT)
    idx = await server.register_namespace(NAMESPACE)
 
    pump = await server.nodes.objects.add_object(idx, "Pump1")
    temperature = await pump.add_variable(idx, "Temperature", 40.0)
    running = await pump.add_variable(idx, "Running", True)
    setpoint = await pump.add_variable(idx, "Setpoint", 45.0)
    await setpoint.set_writable()
 
    async with server:
        t = 0
        while True:
            await asyncio.sleep(0.5)
            t += 1
            await temperature.write_value(40.0 + 5.0 * math.sin(t / 10))
            await running.write_value(True)
 
 
if __name__ == "__main__":
    asyncio.run(main())

温度は40度を中心に、±5度の範囲で正弦波として動きます。実機の挙動を再現するものではなく、読めることを確かめるための値です。

公式のサンプルは、待ち受けアドレスに0.0.0.0を使っています。この記事では127.0.0.1に絞りました。模擬サーバーを同じネットワークの他の機器へ見せないためです。

値を読んで合否を返す検証クライアント

クライアントは、async withで接続と切断を任せます。READMEにも、クライアントはコンテキストマネージャーとして使えると書かれています。名前空間のインデックスは決め打ちにせず、URIから引きます。

import asyncio
import sys
 
from asyncua import Client
 
URL = "opc.tcp://127.0.0.1:4840/plant/"
NAMESPACE = "urn:example:plant-sim"
 
 
async def main() -> int:
    async with Client(url=URL) as client:
        idx = await client.get_namespace_index(NAMESPACE)
        pump = await client.nodes.objects.get_child([f"{idx}:Pump1"])
        temp = await pump.get_child([f"{idx}:Temperature"])
        running = await pump.get_child([f"{idx}:Running"])
        value = await temp.read_value()
        is_running = await running.read_value()
        print(f"namespace index={idx} Temperature={value:.2f} Running={is_running}")
        if not (30.0 <= value <= 60.0):
            print("FAIL: Temperature out of range")
            return 1
        if is_running is not True:
            print("FAIL: Running is not True")
            return 1
        print("OK")
        return 0
 
 
if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

読み取りに使うのはread_value()だけで、write_value()は呼びません。CLAUDE.mdの「読み取り専用」の規約と揃えています。

実行すると何が返るか

サーバーを別のターミナルで起動してから、クライアントを実行します。

.venv/bin/python sim/server.py &
.venv/bin/python sim/check_client.py

手元では、次の出力でした。1行目はasyncuaが出す警告で、値の読み取りには影響しません。

WARNING:asyncua.client.client:Requested session timeout to be 3600000ms, got 600000ms instead
namespace index=2 Temperature=41.95 Running=True
OK

温度は起動からの時間で変わるため、実行のたびに値が違います。名前空間のインデックスが2になったのも、このサーバーで自前の名前空間を1つ登録した場合の結果です。そのため、番号をコードに書かず、URIから引く作りにしています。

範囲の上限を38度に下げて同じクライアントを動かすと、次のように失敗し、終了コードは1になりました。

namespace index=2 Temperature=42.40 Running=True
FAIL: Temperature out of range

この「失敗が終了コードに出る」性質が、Claude Codeに反復させる土台です。公式のベストプラクティスも、テストやビルドの終了コードのように、Claudeが読める合否の信号を与えるよう勧めています。

ualsとuareadでサーバー側を切り分ける

クライアントのコードが失敗したとき、原因がコードにあるのか、サーバー側にあるのかを切り分けたい場面が出ます。asyncuaには、そのためのコマンドが付属しています。READMEの一覧では、ualsが子ノードの一覧、uareadがノードの属性の読み取り、uasubscribeがデータ変更の購読です。

サーバーのObjectsはi=85です。READMEのクライアントの例にも、このノードIDが登場します。ここを起点にパスをたどります。

.venv/bin/uaread -u opc.tcp://127.0.0.1:4840/plant/ \
  -n i=85 -p 2:Pump1,2:Temperature

手元では、警告の行に続いて41.4776010333067のような数値が1行で返りました。パスはカンマ区切りで、起点のノードからの相対で指定します。起点を省くとルートからになるため、-pに2:Pump1,2:Temperatureだけを渡すと一致するノードが見つからず、BadNoMatchが返りました。

つまずきやすい点

落とし穴

サーバーを起動する前にクライアントを動かすと接続拒否になる

サーバーが動いていないと、ConnectionRefusedErrorで終わります。手元でも、サーバーを止めた直後に実行するとConnect call failed ('127.0.0.1', 4840)が出ました。Claudeが原因をクライアントのコードだと誤解しないよう、規約に「先にサーバーを起動する」と書いておくと手戻りが減ります。

ほかにも、次の点を押さえておきます。

  • ポートの取り合い: 4840は、サンプルが使う標準的なポートです。別のプロセスが使っていないか、起動前に確かめます。前回のサーバーが残っていると、新しいサーバーが起動できません。
  • セキュリティは検証用の設定: 手元のサーバーのログには、暗号化の方式が選べない旨の警告が出ました。ここで使うのは暗号化なしの接続で、実機を相手にする設定ではありません。
  • APIの変更: READMEは、1.0より前はAPIが変わりうると警告しています。get_xxからread_xx、set_xxからwrite_xxへの改名も、READMEに記載があります。Claudeが古い記事の名前で書いた場合は、インストール済みのバージョンで試して直させます。
  • サポート範囲: READMEの「Not implemented yet」には、クライアント側の自動再接続の可能性や、サーバー側のアラームが挙がっています。実機の機能を網羅する検証には、この模擬サーバーでは足りません。

実機に向ける前に決めておくこと

模擬サーバーで読めたからといって、実機でも同じように読めるとは限りません。READMEには、動作を確認した相手としてKepwareやBeckhoff、WinCC、B&Rなどが挙がっています。ただし、手元の設備の個別の挙動までは保証されません。

実機に向ける段階では、次の順で進めると安全です。

  1. 接続先のURLだけを設定ファイルに分け、CLAUDE.mdの規約にある許可リストを実機のURLに更新する
  2. 最初はualsとuareadで、読み取りだけを人が実行して確かめる
  3. 書き込みやメソッド呼び出しは、設備の管理者と範囲を決めてから、別の手順として扱う

設備の稼働中に動かす操作は、Claude Codeの権限だけに頼らず、人の承認を挟んでください。

他のプロトコルとの使い分け

同じ考え方は、ほかの産業用プロトコルでも使えます。Modbusなら、書き込みを既定で拒否するラッパーをかませる方法がpymodbusの模擬機器の記事にあります。MQTTなら、ブローカーを立てて送受信を機械的に確かめる手順がmosquittoの検証の記事にあります。

OPC UAは、値だけでなくノードの階層と名前空間を持つ点が、他の二つと違います。そのため、検証クライアントでもパスのたどり方と名前空間の引き方を確かめる必要があります。

検証コマンドをMakefileにまとめると、make checkの1語で回せます。やり方はMakefileの検証ループの記事で解説しています。

まとめ

OPC UAのコードをClaude Codeに任せるときの要は、実機の代わりに立てた模擬サーバーと、合否を終了コードで返すクライアントの2つです。書き込みを避ける規約はCLAUDE.mdに、実行を許すコマンドは権限設定に置くと、役割が分かれます。

ここで確かめられるのは、接続して値を読み、期待の範囲かを判定するところまでです。暗号化や認証、実機固有のノード構成は別の検証になるため、実機に向ける前に人が範囲を決めてください。

この記事を共有:XはてブLinkedIn