Slack API 호출 (stepfunction , lambda)
0. 개요
기존에 stepfunction flow 흐름 내부 작업들의 성공/실패 여부를 람다/http로 slack 알림으로 보내고 있었는데
실제 서비스에 적용하다보니 main 메세지로 가게되면 내용파악이 잘안되고 메시지가 너무 많아져서
쓰레드로 상세 내용을 담기 위해 slack api를 활용하기로 했다.
전
후
1. 방법
슬랙 봇을 만든 다음에 람다에 적용해주고 채널에 초대하면 끝이다!
1. Slack에서 준비할 것 (Bot Token과 Channel ID 얻기)
1단계: Slack 앱(App) 생성
1. Slack API (https://api.slack.com/apps) 페이지로 이동하여 Create New App 버튼을 클릭합니다.
2. From scratch 를 선택하고, 앱 이름(예: "My Workflow Bot")을 정한 뒤, 알림을 보낼 Slack
워크스페이스를 선택.
2단계: 앱에 권한(Scopes) 부여하기
1. 생성된 앱의 설정 페이지에서 Features > OAuth & Permissions 메뉴로 이동합니다.
2. Scopes 섹션까지 스크롤을 내린 후, Bot Token Scopes 그룹에서 Add an OAuth Scope 버튼을
클릭합니다.
3. chat:write 권한을 검색하여 추가합니다. 이 권한은 봇이 채널에 메시지를 쓸 수 있게 해줍니다.
3단계: 앱 설치 및 Bot Token 얻기
1. OAuth & Permissions 페이지 상단으로 다시 올라가 Install to Workspace 버튼을 클릭하여 앱을
워크스페이스에 설치하고 허용합니다.
2. 설치가 완료되면 Bot User OAuth Token 이 표시됩니다. 이 토큰은 xoxb-로 시작하며, 이것이 바로
Lambda 함수에 필요한 SLACK_TOKEN 입니다.
4단계: Channel ID 얻기
1. 메시지를 보낼 Slack 채널에서 채널 이름을 마우스 오른쪽 버튼으로 클릭합니다.
2. "링크 복사"를 선택합니다.
3. 복사된 링크는 https://.../archives/C12345678와 같은 형태이며, 여기서 마지막 부분인
`C12345678` 이 바로 `SLACK_CHANNEL_ID` 입니다.
5단계: 채널에 봇 초대하기
1. 메시지를 보낼 채널에서 /invite @<방금 만든 봇 이름>을 입력하여 봇을 채널에 초대해야 합니다.
2. Lambda 함수에 설정할 것 (환경 변수 등록)
위에서 얻은 두 개의 값을 Lambda 함수가 코드 안에서 안전하게 사용할 수 있도록 환경 변수로 등록해야 합니다.
1. AWS Lambda 관리 콘솔에서 해당 함수 페이지로 이동합니다.
2. 구성(Configuration) 탭 > 환경 변수(Environment variables) 메뉴로 갑니다.
3. 편집(Edit)을 누르고 아래와 같이 두 개의 환경 변수를 추가합니다.
* 키: SLACK_TOKEN, 값: 위에서 얻은 xoxb-로 시작하는 토큰
* 키: SLACK_CHANNEL_ID, 값: 위에서 얻은 C로 시작하는 채널 ID
2. slack notification lambda example
- 전체적인 구조만 공유 (messages 내용은 미포함)
- thread_ts 를 통해 main_res_data와 동일한 time stamp 를 넣어주어 쓰레드로 thread_payload 가 전달될 수 있게 함.
def slack_msg(msg, username=''):
slack_uri = os.environ.get("SLACK_WEBHOOK")
headers = {"content-type": "application/json", "Accept-Charset": "UTF-8"}
try:
r = requests.post(slack_uri, json={'text': msg, 'username': username}, headers=headers)
if r.status_code != 200:
print(f"Error : {r.status_code} {r.text}")
except Exception as e:
print(f"Error slack_msg : {e}")
def get_user_to_ping(lambda_name: str, mapping: dict, is_debugging: bool, test_user: str, default_user: str) -> str:
if is_debugging:
return test_user
return mapping.get(lambda_name, default_user)
def lambda_handler(event, context):
messages = [] #담고싶은 메세지를 sf event에서 받아옴
users_to_ping = set()
ping_string = " ".join(users_to_ping) + "\n" if users_to_ping else ""
main_message_text = messages[0] #첫 head 메세지
thread_message_text = ping_string + "\n".join(messages[1:]) #쓰레드에 달리는 메세지
slack_bot_token = os.environ.get("SLACK_TOKEN")
slack_channel_id = os.environ.get("SLACK_CHANNEL_ID")
if slack_bot_token and slack_channel_id:
try:
api_url = "https://slack.com/api/chat.postMessage"
headers = {
"Authorization": f"Bearer {slack_bot_token}",
"Content-Type": "application/json; charset=utf-8"
}
main_payload = {"channel": slack_channel_id, "text": main_message_text}
main_res = requests.post(api_url, headers=headers, json=main_payload)
main_res.raise_for_status()
main_res_data = main_res.json()
if not main_res_data.get("ok"):
raise Exception(f"Error : {main_res_data.get('error')}")
if thread_message_text.strip():
message_ts = main_res_data.get("ts")
if message_ts:
thread_payload = {
"channel": slack_channel_id,
"text": thread_message_text,
"thread_ts": message_ts
}
thread_res = requests.post(api_url, headers=headers, json=thread_payload)
thread_res.raise_for_status()
if not thread_res.json().get("ok"):
print(f"Failed to post thread message: {thread_res.json().get('error')}")
return {"statusCode": 200, "body": json.dumps(main_res_data)}
except Exception as e:
print(error_detail)
final_text = f"{main_message_text}\n{thread_message_text}"
slack_msg(f"Slack API 호출 실패. Webhook으로 전체 메시지 전송.\n\n{final_text}")
return {"statusCode": 500, "body": {e}}
else:
print("Webhook으로 전송 됨")
final_text = f"{main_message_text}\n{thread_message_text}"
slack_msg(final_text)
return {"statusCode": 200, "body": "Sent via webhook."}