Mqtt Mcp

MQTT MCP サーバー

LLM エージェントを MQTT デバイスに安全かつ標準化された方法で接続する、軽量な Model Context Protocol (MCP) サーバーです。AI 主導のワークフローをビルディングオートメーション（BAS）、産業用制御システム（ICS）、スマートホームシステムとシームレスに統合し、エージェントによるリアルタイムのセンサーデータの監視、デバイスの操作、複雑な自動化タスクのオーケストレーションを可能にします。

はじめに

uv を使って、プロジェクトの依存関係として MQTT MCP サーバーを追加・管理するか、uv pip install または pip install で直接インストールしてください。詳しいインストール手順や詳細については、ドキュメントの インストール セクションをご覧ください。

uv add mqtt-mcp

サーバーはアプリケーションに組み込んで、アプリケーションから直接実行できます。デフォルトでは、http://127.0.0.1:8000/mcp/ に Streamable HTTP エンドポイントを公開します。

# app.py
from mqtt_mcp import MQTTMCP

mcp = MQTTMCP()

if __name__ == "__main__":
    mcp.run(transport="http")

ソースコードを変更せずに、付属の CLI を使ってコマンドラインから起動することもできます。

mqtt-mcp

または、uvx を使って、一時的な隔離環境で起動することもできます。詳しくは、ツールの使用 ガイドをご覧ください。

uvx mqtt-mcp

設定

ほとんどの操作で特定の MQTT ブローカーを対象とする場合、実行時に環境変数で接続設定（host と port）を指定できます。これにより、接続パラメーターを明示的に指定しないすべてのプロンプトは、このブローカーにルーティングされます。

export MQTT_MCP_MQTT__HOST=10.0.0.1
export MQTT_MCP_MQTT__PORT=1883

これらの設定は、作業ディレクトリ内の .env ファイルで指定することもできます。

# .env
mqtt_mcp_mqtt__host=10.0.0.1
mqtt_mcp_mqtt__port=1883

MCP Inspector

サーバーが起動していることを確認し、利用可能なリソースとツールを調べるには、MCP Inspector を実行し、MQTT MCP サーバーの http://127.0.0.1:8000/mcp/ に接続してください。トランスポートを Streamable HTTP に設定してください。

npx @modelcontextprotocol/inspector

主な概念

MQTT MCP サーバーは、FastMCP 2.0 の中核的な構成要素であるリソーステンプレート、ツール、プロンプトを活用し、ボイラープレートを最小限に抑えた、簡潔で Python らしいインターフェースを通じて、MQTT の受信および公開操作を効率化します。

メッセージの受信

デバイス上の各トピックはリソースにマッピングされ（ツールとして公開されます）、リソーステンプレートを使って接続の詳細（host、port）と受信パラメーター（topic、timeout）を指定します。

@mcp.resource("mqtt://{host}:{port}/{topic*}")
@mcp.tool(
    annotations={
        "title": "Receive Message",
        "readOnlyHint": True,
        "openWorldHint": True,
    }
)
async def receive_message(
    topic: str,
    host: str = settings.mqtt.host,
    port: int = settings.mqtt.port,
    timeout: int = 60,
) -> str:
    """Receives a message published to the specified topic, if any."""
    ...

メッセージの公開

公開操作はツールとして公開され、同じ接続の詳細（host、port）を受け取り、単一のアトミックな呼び出しで特定のトピックにメッセージを公開できます。

@mcp.tool(
    annotations={
        "title": "Publish Message",
        "readOnlyHint": False,
        "openWorldHint": True,
    }
)
async def publish_message(
    topic: str,
    message: str,
    host: str = settings.mqtt.host,
    port: int = settings.mqtt.port,
) -> str:
    """Publishes a message to the specified topic."""
    ...

認証

Streamable HTTP トランスポートで組み込みの AuthKit プロバイダーを使った認証を有効にするには、.env ファイルに AuthKit のドメインとリダイレクト URL を指定してください。詳しくは、AuthKit プロバイダー セクションをご覧ください。

インタラクティブなプロンプト

構造化された応答メッセージは、対話をガイドし、不足しているパラメーターを明確にし、エラーを適切に処理するためのプロンプトを使って実装されています。

@mcp.prompt(name="mqtt_help", tags={"mqtt", "help"})
def mqtt_help() -> list[Message]:
    """Provides examples of how to use the MQTT MCP server."""
    ...

以下は、サーバーとの対話に使用できるテキスト入力の例です。

Publish {"foo":"bar"} to topic "devices/foo" on 127.0.0.1:1883.
Receive a message from topic "devices/bar", waiting up to 30 seconds.

例

examples フォルダーには、さまざまなクライアント API を使用して LLM にツールとコンテキストを提供する、MQTT MCP サーバーとの統合方法を示すサンプルプロジェクトが含まれています。

• openai-agents - OpenAI Agents SDK を使用して MQTT MCP サーバーに接続する方法を示します。
• openai - OpenAI Python library のリモート MCP サーバーサポートを活用した最小限のアプリです。
• pydantic-ai - PydanticAI Agent Framework を使用して MQTT MCP サーバーに接続する方法を示します。

Docker

MQTT MCP サーバーは、次のように Docker コンテナーとしてデプロイできます。

docker run -dit \
  --name mqtt-mcp \
  --restart=always \
  -p 8080:8000 \
  --env-file .env \
  ghcr.io/ezhuk/mqtt-mcp:latest

ホストのポート 8080 をコンテナー内の MCP サーバーのポート 8000 に割り当て、存在する場合は .env ファイルから設定を読み込みます。

ライセンス

このサーバーはMIT Licenseの下でライセンスされています。