21 lines
1.3 KiB
Plaintext
21 lines
1.3 KiB
Plaintext
import os, requests
|
|
from dotenv import load_dotenv
|
|
from mcp.server.fastmcp import FastMCP
|
|
load_dotenv(); mcp=FastMCP("Alfred Home Assistant")
|
|
BASE=os.getenv("HA_URL","").rstrip("/"); TOKEN=os.getenv("HA_TOKEN","")
|
|
def headers(): return {"Authorization":f"Bearer {TOKEN}","Content-Type":"application/json"}
|
|
@mcp.tool()
|
|
def get_state(entity_id:str):
|
|
"""Read one Home Assistant entity state exposed for Alfred."""
|
|
r=requests.get(f"{BASE}/api/states/{entity_id}",headers=headers(),timeout=15); r.raise_for_status(); return r.json()
|
|
@mcp.tool()
|
|
def call_service(domain:str, service:str, entity_id:str, data:dict|None=None):
|
|
"""Call an allowed HA service. Prefer entities explicitly exposed to Assist."""
|
|
allowed=set(x.strip() for x in os.getenv("HA_ALLOWED_DOMAINS","light,switch,cover,climate,media_player,calendar").split(','))
|
|
if domain not in allowed: raise ValueError("Domaine non autorisé")
|
|
payload={"entity_id":entity_id, **(data or {})}; r=requests.post(f"{BASE}/api/services/{domain}/{service}",headers=headers(),json=payload,timeout=20); r.raise_for_status(); return r.json()
|
|
@mcp.tool()
|
|
def list_states(domain:str=""):
|
|
r=requests.get(f"{BASE}/api/states",headers=headers(),timeout=20);r.raise_for_status();items=r.json();return [x for x in items if not domain or x['entity_id'].startswith(domain+'.')]
|
|
if __name__=='__main__': mcp.run()
|