df4cd98a9e
Incoming-webhook / Power Automate Workflows, 3 commands: send message, send Adaptive Card (title/text/facts/action button). Webhook-URL auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
import json, os, sys, urllib.request, urllib.error
|
|
|
|
|
|
def _cfg():
|
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
|
|
|
|
def _inputs():
|
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
|
|
|
|
def post(cfg, payload):
|
|
url = str(cfg.get("webhook_url", ""))
|
|
if not url:
|
|
raise Exception("webhook_url is not configured")
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers = {"Content-Type": "application/json"}
|
|
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
r.read()
|
|
return r.status
|
|
|
|
|
|
def adaptive_card(body_blocks, actions=None):
|
|
card = {
|
|
"type": "AdaptiveCard",
|
|
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
|
|
"version": "1.4",
|
|
"body": body_blocks,
|
|
}
|
|
if actions:
|
|
card["actions"] = actions
|
|
return {
|
|
"type": "message",
|
|
"attachments": [
|
|
{"contentType": "application/vnd.microsoft.card.adaptive", "content": card}
|
|
],
|
|
}
|
|
|
|
|
|
def _run(fn):
|
|
try:
|
|
print(json.dumps(fn(_cfg(), _inputs())))
|
|
except urllib.error.HTTPError as e:
|
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(json.dumps({"error": str(e)}))
|
|
sys.exit(1)
|
|
|
|
|
|
def main(cfg, inputs):
|
|
text = str(inputs.get("text", "")).strip()
|
|
if not text:
|
|
raise Exception("text is required")
|
|
|
|
payload = adaptive_card([{"type": "TextBlock", "text": text, "wrap": True}])
|
|
status = post(cfg, payload)
|
|
return {"ok": True, "status": status}
|
|
|
|
|
|
_run(main)
|