Deep-dive · Smart Home + Agent Protocols¶

import json
import aiohttp

HA_URL = "http://homeassistant.local:8123"
HA_TOKEN = "YOUR_LONG_LIVED_TOKEN"
HEADERS = {"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"}


async def turn_on_light(entity_id: str, brightness: int = 255) -> dict:
    async with aiohttp.ClientSession() as s:
        async with s.post(
            f"{HA_URL}/api/services/light/turn_on",
            headers=HEADERS,
            json={"entity_id": entity_id, "brightness": brightness},
        ) as r:
            return await r.json()


async def stream_states():
    """WebSocket subscribe state_changed."""
    async with aiohttp.ClientSession() as s:
        async with s.ws_connect(f"ws://homeassistant.local:8123/api/websocket") as ws:
            assert (await ws.receive_json())["type"] == "auth_required"
            await ws.send_json({"type": "auth", "access_token": HA_TOKEN})
            assert (await ws.receive_json())["type"] == "auth_ok"
            await ws.send_json({
                "id": 1, "type": "subscribe_events", "event_type": "state_changed"
            })
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    data = json.loads(msg.data)
                    if data.get("type") == "event":
                        evt = data["event"]["data"]
                        print(f"[HA] {evt.get('entity_id')} → {evt.get('new_state', {}).get('state')}")

import aiohttp


async def commission_device(code: str) -> dict:
    async with aiohttp.ClientSession() as s:
        async with s.ws_connect("ws://localhost:5580/ws") as ws:
            await ws.send_json({
                "message_id": "1",
                "command": "commission_with_code",
                "args": {"code": code, "network_only": False},
            })
            return await ws.receive_json()


async def discover_nodes() -> list:
    async with aiohttp.ClientSession() as s:
        async with s.ws_connect("ws://localhost:5580/ws") as ws:
            await ws.send_json({"message_id": "2", "command": "get_nodes", "args": {}})
            return (await ws.receive_json()).get("result", [])

import json
import aiomqtt


async def publish_command(device_id: str, capability: str, payload: dict):
    async with aiomqtt.Client(
        hostname="mosquitto", port=8883,
        tls_params=aiomqtt.TLSParameters(
            ca_certs="/certs/ca.crt",
            certfile="/certs/jarvis-agent.crt",
            keyfile="/certs/jarvis-agent.key",
        ),
    ) as c:
        await c.publish(f"jarvis/{device_id}/{capability}/set",
                        payload=json.dumps(payload), qos=1)

"""
MCP server per Jarvis home control.
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource, CallToolResult
import aiohttp

app = Server("jarvis-home-mcp")
HA_URL = "http://homeassistant.local:8123"
HA_HEADERS = {"Authorization": f"Bearer YOUR_TOKEN"}


@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="home_turn_on_light",
            description="Accende una luce o gruppo di luci",
            inputSchema={
                "type": "object",
                "properties": {
                    "entity_id": {"type": "string"},
                    "brightness_pct": {"type": "integer", "minimum": 1, "maximum": 100},
                },
                "required": ["entity_id"],
            },
        ),
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    async with aiohttp.ClientSession(headers=HA_HEADERS) as s:
        if name == "home_turn_on_light":
            payload = {"entity_id": arguments["entity_id"]}
            if "brightness_pct" in arguments:
                payload["brightness_pct"] = arguments["brightness_pct"]
            async with s.post(f"{HA_URL}/api/services/light/turn_on", json=payload) as r:
                result = await r.json()
                return CallToolResult(content=[TextContent(type="text", text=str(result))])

import httpx
from a2a.client import A2AClient
from a2a.types import SendMessageRequest, MessageSendParams

HOME_AGENT_URL = "http://jarvis-home-agent:8001"


async def delegate_to_home_agent(task: str) -> str:
    async with httpx.AsyncClient() as http:
        client = A2AClient(http_client=http, url=HOME_AGENT_URL)
        request = SendMessageRequest(
            params=MessageSendParams(
                message={
                    "role": "user",
                    "parts": [{"kind": "text", "text": task}],
                    "messageId": "msg-001",
                }
            )
        )
        response = await client.send_message(request)
        return response.result.parts[0].text if response.result else ""

# Notifiche via DBus
import dbus


def send_notification(device_id: str, title: str, body: str):
    bus = dbus.SessionBus()
    device = bus.get_object(
        "org.kde.kdeconnect.daemon",
        f"/modules/kdeconnect/devices/{device_id}/notifications"
    )
    iface = dbus.Interface(device, "org.kde.kdeconnect.device.notifications")
    iface.sendNotification(title, body, "jarvis", 0)

import jwt as pyjwt
from functools import lru_cache

AUTHENTIK_URL = "https://auth.jarvis.local"
JWKS_URL = f"{AUTHENTIK_URL}/application/o/jarvis/.well-known/jwks.json"


@lru_cache(maxsize=1)
def jwks_client() -> pyjwt.PyJWKClient:
    return pyjwt.PyJWKClient(JWKS_URL)


def verify_jwt(token: str) -> dict:
    signing_key = jwks_client().get_signing_key_from_jwt(token)
    return pyjwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256", "ES256"],
        audience="jarvis-agent",
        issuer=f"{AUTHENTIK_URL}/application/o/jarvis/",
        options={"require": ["exp", "iat", "sub", "aud"]},
    )