AI凄いのは分かったから、そろそろフィジカルの話をしないか」AgentCore × Cloudflare Tunnelで日常に干渉する

はじめに：AIは「フィジカル」へ向かう

どうもアスカです。最近は使わなくなったスマホやRasberry Pi5を使って夜な夜な遊んでいます。

センサー値取得

スクワットカウント

AI技術の進化は、かつてないスピードで加速していますよね。

LLM（大規模言語モデル）の推論能力は日進月歩で向上し、Model Context Protocol (MCP) のようなデファクトスタンダードなツールプロトコルが普及し始めました。
さらに、Amazon Bedrock AgentCore のようなプラットフォームの進化により、APIを介して実行できる大抵の操作は、もはや「できて当たり前」の領域。デジタル空間におけるタスク自動化は、ある意味で「ほとんど解かれてしまった問題」になりつつあり、正直、ビックテックが新しいモデルやツールをリリースする度に凄い凄いと囃し立てる画一的な情報の波に若干うんざり感もあります。

「我々エンジニアは、この状況でどのように遊べば良いのか？」

その答えの一つが「Physical AI（フィジカルAI）であり、本記事はその「導入」になるんじゃないかと思います。

足元を見れば、
画面の中のAIがどれほど賢くなっても、我々の住む「日常の物理世界」への関与は驚くほど希薄です。
チャットボットに「部屋が寒い」と訴えても、生成AIは共感することしかできず、エアコンのリモコンを操作することはできません。
APIで完結するタスクは自動化されましたが、物理的なスイッチひとつ、センサーひとつをAIエージェントに委ねるハードルは依然として高いままです。

本記事では、この状況を打破し、クラウド上のAIエージェントと、日常（自宅サーバ/IoT）を、Cloudflare Tunnel を使って接続するアーキテクチャを提案します。
これは、AIをディスプレイの向こう側から引きずり出し、日常に干渉させるための「遊び方」の提案です。

アーキテクチャ：Web上のAIエージェントと自宅を安全に接続する

私たちが遊ぶには、特定のAIエージェントからの命令だけを通すセキュアな接続が必要不可欠です。

全体構成図

どのように「日常」へ干渉するか

1. A2A Client: ユーザー入力を受け取り、AWS 上で稼働する A2A Server (AgentCore Runtime) へリクエストを送ります。ここはA2Aクライアントのリクエストに変換できれば何でも良いです。LINEやSlack、あるいはWebアプリなど、好きなもので構いません。序盤のデモでは、ローカルに稼働させたStarans AgentにA2Aクライアントのツールを持たせています。

2. A2A Server (AWS): リクエストを受け取り、Amazon Bedrock Model を呼び出し、応答を生成します。応答するのにツールが必要なら、さらに別のA2Aサーバにリクエストを渡します。いわゆるここがそれぞれのツールとインターフェースのハブの役割になります。

3. A2A Server (AWS): AgentCore Identity から Cloudflare の認証トークンを取得し、Cloudflare Tunnel 経由で自宅のサーバへ安全にリクエストを発行します。

4. A2A Server (Home): 自宅の Raspberry Pi 等で待機している A2A サーバがリクエストを受信し、実際の物理デバイスを操作。結果を AWS 側へ返却します。いわゆる自宅側のハブの役割になります。

5. 安全な道: Cloudflare Tunnel

   • 自宅サーバ（A2Aサーバ）をインターネットに晒すことなく、特定の許可されたアクセス（AgentCoreからのリクエスト）だけをZero Trustで通すためのトンネルです。自宅側はOutboundの通信だけで成立するため、ポート開放という最大のリスクを排除できます。

次章では、この構成で運用するための、セキュリティ設計について説明します。ただ繋がるだけでは、安心して遊べません。

２章　セキュリティ設計：自宅をインターネットに晒さないために

「日常への干渉」は魅力的ですが、それは同時に「自宅を他人に渡す」リスクも孕んでいます。
例えば、自宅のカメラをAIエージェントに見せる場合、一歩間違えれば、自宅の居間の状況を全世界に公開してしまう事になりかねません。

もし攻撃者があなたのAIエージェント（あるいはその裏のAPI）を乗っ取れば、勝手に部屋の鍵を開けたり、エアコンを極寒設定にしたり、あるいはプライベートな映像を覗き見たりすることが可能になります。

Web上のAIエージェントなら「変なことを喋る」程度で済むかもしれませんが、フィジカルAIは実害が出ます。例えば、私が裸でスクワットしている姿が自宅のカメラから全世界に公開され、一生消えないデジタルタトゥーになる事態も考えられます。だからこそ、慎重なセキュリティ設計が必須です。

1. ネットワーク層：インバウンドポート「全閉」

従来の自宅サーバ公開では、ルーターのポートフォワーディング（ポート開放）が定石でした。しかし、これは自宅の玄関ドアを開けっ放しにするようなものです。

今回採用する Cloudflare Tunnel は、この今までの定石を覆します。

• Outbound Only: 自宅サーバの cloudflared デーモンが、Cloudflareのエッジサーバに向かって内側から接続を見に行きます。
• No Public IP: 自宅のグローバルIPをDNSに登録する必要も、公開する必要もありません。悪意を持った攻撃者はあなたの自宅のIPを知ることすらできません。
• DDoS Protection: 全てのトラフィックはCloudflareのネットワークでフィルタリングされてから届きます。

「ポートを開けない」という安心感は、自宅の物理デバイスを扱う上で何物にも代えがたいメリットです。

2. アプリケーション層：AgentCore Identity による認証管理

「トンネルがあるから安心」ではありません。トンネルの入り口（CloudflareのURL）を知っていれば、誰でもアクセスできてしまうからです。
そこで、承認されたAgentCore Runtimeからのみ、このトンネルを通れるように認証をかけます。

ここでは Cloudflare Access Service Token を活用します。

1. Cloudflare側: Access Policyを設定し、特定の Client ID / Client Secret ヘッダーを持つリクエストのみを透過させます。
2. AgentCore Identity側: この秘匿情報を安全に保持します。
3. AgentCore Runtime側: 実行時に Identity からトークンを取得し、リクエストヘッダーに付与してCloudflareへ送信します。

キーとシークレットによる二段階検証（Service Token）

Cloudflare Access (Zero Trust) を設定することで、リクエストが自宅へのトンネルに入る前に、Cloudflare のエッジ上で以下の検証が強制されます：

1. 検問所の設置: Cloudflare のエッジ上で、特定ドメイン（例：home-agent.your-domain.com）への全リクエストをゲートウェイで一時停止させます。
2. キーの照合 (Client ID): リクエストヘッダーに含まれる CF-Access-Client-Id が、あなたが事前に発行した正式な ID と完全に一致するかをチェックします。
3. シークレットの検証 (Client Secret): ID が一致した場合のみ、次に CF-Access-Client-Secret が正しいペアであることを検証します。
4. ゲートの開放: 両方のキーが正しい場合のみ、リクエストは「承認済み」として自宅へのトンネルを通過できます。

もしキーが欠けていたり、1文字でも間違っていれば、通信は Cloudflare のエッジで即座に遮断（403 Forbidden）されます。この検問は Cloudflare 側で行われるため、不正なリクエストがあなたの自宅のネットワークに到達することすらありません。

なぜ AWS Secrets Manager ではなく Identity なのか？

「秘密情報を守るだけなら AWS Secrets Manager で十分ではないか？」という疑問が湧くかもしれません。しかし、高度なエージェント・アーキテクチャにおいては、AgentCore Identity を使う明確な理由があります。

1. ワークロード単位での緻密なアクセス制御

   • Secrets Managerは、あくまで鍵（API Keyやトークン）を格納するための「器」です。
   • 対して AgentCore Identity は、エージェント一つひとつに独立した「ワークロード・アイデンティティ（誰か）」を付与します。
   • 各エージェントは専用の IAM 実行ロール を持ち、そのロールに対して「どのトークンを取得できるか」をポリシーで制御できます。つまり、「Aというエージェントは自宅カメラのトークンにアクセスできるが、Bというエージェントはエアコン操作のトークンしか見えない」といった、ワークロード単位の権限分離（マルチテナント的な隔離）が容易に実現できます。

2. 複雑な認証フローの「オーケストレーション」

   • Cloudflare Tunnelのような外部サービスと連携する場合、単純な固定キーだけでなく、OAuth 2.0（2-legged/3-legged）などの複雑な認証フローが必要になることがあります。
   • AgentCore Identity はこれらのフローをネイティブでサポートし、トークンの取得・更新・注入を自動的にオーケストレーションします。これを自作のロジックでやろうとすると、トークンの失効管理や更新処理を複雑に作り込まなければなりませんが、Identityならプラットフォーム側に任せられます。
     これにより、例えURLが漏洩したとしても、Identity にアクセスできない第三者は、自宅の A2A サーバに指一本触れることはできません。

3. AgentCore Runtime との密結合による「ボイラープレート・レス」な実装

   • RuntimeとIdentityが密に統合されており、Identity から安全に認証情報が注入されます。
   • 従来の「コード内で Secrets Manager から値を取り出し、HTTPヘッダーにセットする」という泥臭いボイラープレートコードを書く必要がなくなります。 開発者は「どのツールにどの認証が必要か」を宣言するだけで、実行時（Runtime）がすべての重労働を肩代わりしてくれます。
   • これにより、人為的なミス（トークンの誤操作やログへの露出）を根本から排除しつつ、開発スピードを劇的に向上させることが可能です。

AgentCore Identity には @requires_access_token のようなデコレータがあり、OAuth のトークン取得・更新を自動化できるのが最大の強みです。

ただし今回は、Cloudflare Access の Service Token（Client ID/Secret）を API キーとして取得する構成をとっています。そのため、今回はあえてIdentity から取得した値を手動でヘッダーに注入する実装としています。

3. 権限管理：AgentCore Runtime の IAM 最小権限 (Least Privilege)

「もし AI が予期せぬ挙動をしたら？」への究極の備えは、実行環境の権限を極限まで絞ることです。

• AgentCore Runtime 実行ロールの最小化: AgentCore Runtime が動作する際に使用する IAM 実行ロールに対し、「どの Identity からトークンを取得できるか」「どの Bedrock Model を Invoke できるか」を厳格に制限します。これにより、万が一 A2A サーバ上のコードに脆弱性があったとしても、被害をその範囲内に封じ込めます。

2章まとめ：Zero Trustで安心な遊び場を作る

1. ネットワーク: ポート全閉 (Cloudflare Tunnel)
2. 認証: A2A Serverのみ通行許可 (Service Token)
3. 権限: 最小権限 (Least Privilege)

この3段構えにより、AIに物理世界への干渉権限を与えつつ、夜も安心して眠れる堅牢なアーキテクチャが完成します。

3-1章　A2A Client から A2A Server へ：入口の実装

最初はシステム全体の「入口」となる A2A クライアントの実装から簡単に解説します。
本プロジェクトでは、ローカルで稼働する Strands Agent に、クラウド（AgentCore）上のエージェントを呼び出すための「ツール」を持たせています。

A2Aクライアントからクラウド上のA2Aサーバへのリクエスト

ここで重要なのは、クラウド側の A2A Server (Hub) は AgentCore Runtime 上にデプロイされている という点です。
今回はAWS SDK (boto3) を介して InvokeAgent API を叩くことでリクエストを送っています。

実装例：AgentCore Runtime を呼び出す A2A クライアント・ツール

以下はクラウド上のAgent Core Runtimeをツールとして呼び出す実装抜粋です。

import json
import uuid
import boto3
from strands import tool

# AgentCore Runtime の ARN
HUB_AGENT_RUNTIME_ARN = "arn:aws:bedrock-agentcore:..."

def _invoke_hub_sync(text: str) -> str:
    # AWS SDK を使用してエージェントを実行
    client = boto3.client("bedrock-agentcore")
    
    # A2A (JSON-RPC) の標準的な message/send ペイロードを構築
    # ※ id はメッセージの応答を識別するために一意である必要があります
    rpc_id = str(uuid.uuid4())
    payload = {
        "jsonrpc": "2.0",
        "id": rpc_id,
        "method": "message/send",
        "params": {
            "message": {
                "message_id": f"msg-{rpc_id[:8]}",
                "role": "user",
                "parts": [{"kind": "text", "text": text}],
            }
        },
    }
    
    # AgentCore Runtime へリクエスト
    # (中略: InvokeAgentRuntime の呼び出し)
    
    # レスポンスからテキストを抽出（詳細はソースコード参照）
    return _extract_a2a_text(resp["payload"].read().decode("utf-8"))

@tool(
    name="cloud_hub_message_send",
    description="クラウド上の A2A Hub にメッセージを送信し、高度な推論や自宅操作を依頼します。"
)
def cloud_hub_message_send(text: str) -> dict:
    result = _invoke_hub_sync(text)
    return {"status": "success", "content": [{"text": result}]}

なぜ ToolProvider ではなく手動実装なのか？

Strands には A2AClientToolProvider のような便利なライブラリがありますが、それは主に HTTP(S) ベースの A2A サーバを対象としています。
今回のように 「AgentCore Runtime への Invoke」 という AWS 特有のインターフェース(/invoke)を経由する場合、現時点ではこのように boto3 をラップしたカスタムツールを作成するのが確実です。

補足

• AWS SDK (boto3) の採用: 公式ドキュメントのサンプルでは httpx を利用した HTTP 直接呼び出し（Bearer Auth）が紹介されていますが、本プロジェクトでは AWS SDK (boto3) を採用しています。
• SigV4 による認証: SDK を使うメリットは、AWS 標準の SigV4 (Signature Version 4) 署名を利用できる点です。これにより、Bearer トークンを手動で管理・発行する手間が省け、IAM ロールに基づいた堅牢なアクセス制御が容易になります。

参考資料：公式ドキュメント

実装の際は、以下の AWS 公式ドキュメントを参照してください。

AgentCore Runtime における A2A プロトコルの仕様、および httpx を使用した呼び出し方法についての公式ガイドです。

3-2章 クラウド上の「ハブ」：AgentCore Identity を利用した安全な中継

クライアントからのリクエストを受け取り、自宅のデバイスへと橋渡しをする「クラウド側の頭脳」の実装を解説します。
ここでのポイントは、単なる中継サーバ（A2A Server）であるだけでなく、セキュリティ（Identity）とツール統合のハブ（Hub）として機能させる点にあります。

1. Strands Agent によるオーケストレーション

クラウド側の実装では、strands-agents を使用して A2A Server を立ち上げます。
このサーバは、単独で回答するのではなく、バックエンドにある「自宅の A2A サーバ」を 「ツール」 として認識し、必要に応じて呼び出すオーケストレーターとして振る舞います。

# a2a_hub.py のエッセンス
from strands import Agent, tool
from strands.multiagent.a2a import A2AServer

# 自宅サーバを呼び出す「ツール」を定義
@tool(name="myhome_a2a_message_send")
def myhome_a2a_message_send(text: str) -> dict:
    # 後述する認証ヘッダーを付与して自宅の Cloudflare Tunnel 端点へ POST
    ...

2. AgentCore Identity による認証情報の注入

今回のキモとなるのが、AWS Bedrock AgentCore Identity との連携です。
Cloudflare Tunnel (Cloudflare Access) を通過するための「Service Token」的な秘匿情報を、コード内にハードコードしたり環境変数に直接持たせるのではなく、Identity サービスから動的に取得します。

ワークロードトークンの取得と API キーの解決

AgentCore Runtime 上で動作するコードは、自分自身のアイデンティティ（Workload Access Token）を使って、登録された Provider から API キー（Cloudflare の Client ID/Secret）を引き出すことができます。

from bedrock_agentcore.services.identity import IdentityClient
from bedrock_agentcore.runtime import BedrockAgentCoreContext

def _get_cf_access_headers():
    # 1. 自身のワークロードトークンを取得
    token = BedrockAgentCoreContext.get_workload_access_token()
    
    # 2. Identity Client を介して Cloudflare の認証情報を解決
    # 本来は BedrockAgentCore.Client の get_resource_api_key を使用します
    identity = boto3.client("bedrock-agentcore")
    
    resp_id = identity.get_resource_api_key(
        resourceCredentialProviderName="cf_access_client_id",
        workloadIdentityToken=token
    )
    client_id = resp_id["apiKey"]

    resp_secret = identity.get_resource_api_key(
        resourceCredentialProviderName="cf_access_client_secret",
        workloadIdentityToken=token
    )
    client_secret = resp_secret["apiKey"]
    
    return {
        "CF-Access-Client-Id": client_id,
        "CF-Access-Client-Secret": client_secret
    }

これにより、「クラウド側のエージェントだけが、自宅のデバイスにアクセスする正当な権利を持つ」 という強固なアクセス制御が実現します。

3. 自宅 A2A サーバの「ツール化」

A2A Hub にとって、自宅の Raspberry Pi 等で動くサーバは一つの「ツール」に過ぎません。
LLM（今回は Nova 2 Lite 等）に対し、「部屋の温度が知りたい」「電気を消して」といったリクエストが来たら、このツールを実行するように指示します。

# ツールの実行部（一部抜粋）
headers = _get_cf_access_headers() # Identity から取得したヘッダーを注入

# MYHOME_A2A_HUB_URL は Cloudflare Tunnel で公開した自宅サーバの URL
# 例: https://home-agent.your-domain.com/
payload = {
    "jsonrpc": "2.0",
    "method": "message/send",
    "params": { "message": { "parts": [{"kind": "text", "text": text}] } }
}
resp = httpx.post(MYHOME_A2A_HUB_URL, json=payload, headers=headers)

[!IMPORTANT]
Cloudflare Tunnel による「神経接続」
ここで指定する MYHOME_A2A_HUB_URL は、インターネットにポートを公開しているわけではなく、Cloudflare Tunnel によって確立された仮想的な専用線 の入り口です。
Identity から取得した CF-Access-Client-Id/Secret がヘッダーに含まれていない限り、Cloudflare のエッジで通信が遮断されるため、自宅のサーバが直接的なサイバー攻撃に晒されることはありません。

4. Agent の定義

Identity から情報を取得するツールが揃ったら、これらを束ねる「エージェント」を定義します。
ここでは、Bedrock の Nova モデルを「頭脳」とし、作成したカスタムツールを「手足」として登録します。

from strands import Agent
from strands.models import BedrockModel

# 1. モデル（頭脳）の定義
model = BedrockModel(
    model_id="us.amazon.nova-2-lite-v1:0",
    region_name="us-west-2"
)

# 2. エージェントの組み立て
agent = Agent(
    model=model,
    tools=[myhome_a2a_message_send], # ツールを追加
    system_prompt="あなたはハブエージェントです。必要に応じて自宅サーバを呼び出してください..."
)

5. A2A Server の起動：Runtime での待ち受け

最後に、これまで構築した Agent を A2A Server として公開します。
AgentCore A2A Runtime の期待値（ポート 9000、ルートでの配信など）に合わせて設定を行います。

from strands.multiagent.a2a import A2AServer

# Agent を A2A プロトコルでホストするサーバーを起動
server = A2AServer(
    agent=agent,
    host="0.0.0.0",
    port=9000,
    serve_at_root=True, # Bedrock AgentCore Runtime の要件
    version="0.0.1"
)
server.serve()

3-2章まとめ：クラウドから自宅に繋ぐ

クラウド側の Hub は、「Identity による認証」 と 「自宅エージェントへのルーティング」 を担う重要なレイヤーです。
この設計により、クライアントは AWS の標準的な API（InvokeAgentRuntime）を叩くだけで、背後にある複雑なネットワークや認証を意識することなく、自宅の物理デバイスと「会話」することが可能になります。

第3-3章 自宅側 A2A サーバと Cloudflare

いよいよ終着点、自宅側の構成について説明します。
ここでのポイントは、「クラウド経由のリクエストを安全に受け取る仕組み」 をどう実現するかです。

1. 自宅側の A2A サーバ：LAN 内の「ハブ」としての役割

自宅では、実際にセンサーや家電操作を担うデバイスを束ねる「ローカル・ハブ」として、A2A サーバを動かします。
単一の機能を持たせるのではなく、複数の「手足」を呼び出すオーケストレーターとして実装するのが実戦的です。

実装例：センサーと CLI 連携を統合するローカル・ハブ

以下は、LAN 内の別のセンサーエージェントを呼び出したり、ローカルの CLI ツール（OpenClaw）を介して Google カレンダー等を操作したりする実装の抜粋です。

from strands import Agent, tool
from strands.multiagent.a2a import A2AServer
from strands.multiagent.a2a.executor import StrandsA2AExecutor

# LAN内の他エージェント（センサー等）を呼び出すツール
@tool(name="sensora2a")
async def sensora2a(message: str) -> dict:
    """LAN内のセンサーエージェントから温度などを取得します。"""
    # 内部で httpx 等を使用して他の A2A エージェントを Invoke
    response_text = await _call_agent(SENSOR_A2A_URL, message)
    return {"status": "success", "content": [{"text": response_text}]}

# ローカルの CLI ツールを呼び出すツール
@tool(name="openclawmain")
async def openclawmain(message: str) -> dict:
    """Googleカレンダー等の操作を行うローカルCLIを実行します。"""
    # asyncio.create_subprocess_exec 等でコマンドラインを実行
    response_text = await _call_openclaw_main(message)
    return {"status": "success", "content": [{"text": response_text}]}

# エージェントの定義
AGENT = Agent(
    name="Local Hub Agent",
    description="LAN内の各種デバイスやツールへリクエストをルーティングします。",
    tools=[sensora2a, openclawmain]
)

# 起動：A2A Server として公開
# AgentCore Runtime は通常 9000 番ポートでの待受けを期待します
server = A2AServer(agent=AGENT, port=9000)
server.serve()

このように、自宅側サーバを「ハブ」にすることで、クラウド側の A2A エージェント はこの一台のトンネル端点と話すだけで、自宅内のあらゆるリソースにアクセスできるようになります。

2. Cloudflare Tunnel：なぜ「安全」と言えるのか？

このエッジデバイスをインターネットにさらすことなく、クラウド 側から呼び出すために Cloudflare Tunnel (cloudflared) と Cloudflare Access (Zero Trust) を組み合わせて利用します。

通常、外部から自宅へのアクセスには「ポート開放」が必要ですが、本構成では以下の二重の守りによってそのリスクを完全に排除します：

1. 内側からのセキュアパス確立: 自宅内の cloudflared が Cloudflare のエッジノードに対して 内側から外側へ (Outbound) 接続を確立し、永続的な専用パスを維持します。これにより、グローバル IP を隠蔽し、ルーターのインバウンドポートも 1 つも開けない状態（全閉）で通信が可能になります。
2. Service Token による検問: トンネルがあるだけでは不十分なため、正当な AIエージェント 以外からのアクセスを弾く Service Token (Client ID / Secret) による検証を強制します。

この認証の仕組み（ID と Secret による二段階検証）については、第2章：セキュリティ設計 で概説した通り、Cloudflare のエッジ上で実施されます。正しいキーを持たないリクエストは、自宅のネットワークに到達することすらありません。

まとめ

これにて、「クラウドのAIエージェント」「情報の導管（Identity & Cloudflare）」「自宅AIエージェント」 がすべて繋がりました。

• 攻撃ベクトルを最小限に: ネットワーク的には「存在しない」状態で、特定の相手とだけ話せる設計。
• 「Service Token」による認証: 正当なパスポートを持つエージェントだけが私の家へ干渉可能。
• 「フィジカル AI」の実現: AIエージェントが、セキュアな導管（トンネル）を通って、具体的な行動（家電操作・センサー取得）へと昇華されます。

フィジカルAI、本格的なもので言えば、ロボット操作なんでしょうね。vLLMを使った自走ロボットとか憧れますが、いきなりそこに手を出すのは、技術的にも資金的にも厳しいです。

とはいえ、どんなフィジカル拡張にせよ、安全にインターネットや様々なインターフェースと接続する構成は必須です。この記事が、そんな最初の一歩の手助けになれば幸いです。