はじめに
UiPath Orchestratorで発生するイベント（例：ジョブの開始・終了、キューアイテムの追加など）の通知を受け取りたいケースってよくあると思います。今回はTeamsで通知を受け取る方法にチャレンジしてみます。
※今回はRobotの処理中で発生したイベント（業務エラーなど）の通知ではありません。その場合は、ワークフローにメール送信機能などを組み込むケースが多いと思います。
使うもの
UiPath Orchestrator（Clould） ※2021/12時点
AWS API Gateway
AWS Lambda
Microsoft Teams
仕組み
大まかな仕組みは、こんな感じです。
Orchestratorで発生したイベントはWebhookで外部アプリケーション（AWS）へ送信します。
Webhookで発信されたイベントはAPI Gatewayで受け取り、Lambdaで解析され、Teamsに送られます。
TeamsではIncoming Webhookを使い、チャネルにポストします。
作ってみる
その１ Teamsの設定
Teamsでは通知先としたいチャネルからコネクタ「Incoming Webhook」を有効化させます。
その際、URLが表示されるのでメモしておきます。
その２ Lambdaの設定
Orchestratorから受け取ったWebhookを解析し、TeamsのAPIを使ってチャネルにポストするプログラムを実装します。
言語は何でもいいのですが、今回はPython 3.8とします。
以下は、ジョブの開始・終了・失敗のイベントを処理するソースコードです。
Pythonはかじった程度なので、いいコードではないかもしれません。また最低限しか実装していませんので、参考程度としてください。
import json
import datetime
import dateutil.parser
import urllib.request
import traceback
import sys
def lambda_handler(event, context):
# TeamsのIncoming WebhookのURL
teams_url = "https://..."
data = {}
err_flg = 0
try:
# イベント別にメッセージとテーマカラーを設定する
if event["Type"] == "job.started":
msg_title = "ジョブが開始されました"
theme_color = "17a2b8"
elif event["Type"] == "job.faulted":
msg_title = "ジョブが異常終了しました"
theme_color = "dc3545"
elif event["Type"] == "job.completed":
msg_title = "ジョブが正常終了しました"
theme_color = "28a745"
else:
# 指定以外のイベント発生時
err_flg = 9
if err_flg == 0:
# メッセージの内容
teams_message = [
{"name" : "発生日時", "value" : (dateutil.parser.parse(event["Timestamp"]) + datetime.timedelta(hours=9)).strftime("%Y/%m/%d %H:%M:%S")},
{"name" : "ジョブID", "value" : str(event["Job"]["Id"])},
{"name" : "プロセス名", "value" : event["Job"]["Release"]["ProcessKey"]},
{"name" : "マシン名", "value" : event["Job"]["Robot"]["MachineName"]}
]
# ジョブ異常終了の場合はエラー情報を付加
if event["Type"] == "job.faulted":
teams_message.append({"name" : "情報", "value" : event["Job"]["Info"]})
data = {
"themeColor" : theme_color,
"summary" : msg_title,
"sections" : [{
"activityTitle" : msg_title,
"facts" : teams_message
}],
}
except Exception as e:
# Lambdaエラー時
err_flg = 1
traceback.print_exc()
data = {
"themeColor" : "dc3545",
"summary" : "AWS Lambda Error",
"sections" : [{
"activityTitle" : "AWS Lambda Error",
"facts" : [
{"name" : "発生日時", "value" : datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).strftime("%Y/%m/%d %H:%M:%S")},
{"name" : "情報", "value" : str(e)}
]
}],
}
if err_flg != 9:
# メッセージをTeamsに送る(Incoming Webhook)
req = urllib.request.Request(teams_url, json.dumps(data).encode())
urllib.request.urlopen(req)
if err_flg == 0:
print("[Success] " + event["Type"] + ": " + event["Job"]["Release"]["ProcessKey"] + "(" + str(event["Job"]["Id"]) + ")")
elif err_flg == 1:
print("[Error] AWS Lambda Error")
else:
print("[Warn] Other OC Event(Don't notify Teams)");
return 200
Orchestrator Webhookのメッセージ（JSON）フォーマットやTeamsメッセージの装飾などは下記参考をご覧ください。
その３ API Gatewayを設定する
REST APIを作成し、OrchestratorのWebhookの受け口を作成します。
今回はリソース「jobs」、メソッドは「POST」とします。
メソッドを作成の際、上記作成したLambda関数を指定します。
設定が終わったら、任意のステージにデプロイします。その際、URLが表示されるのでメモしておきます。
その４ Orchestratorを設定する
最後にOrchestratorのWebhookを有効化します。
URLにはAPI Gatewayの設定でメモしたURLにリソースを追加したものを指定します。
イベントは通知したいものを選択します。全選択もできますが、その分、Lambdaのプログラムが複雑になるのでお勧めしません。必要最小限に絞った方が無難です。
動かしてみる
Orchestratorでジョブを動かしてみると、無事Teamsに通知できました。
