executing-plans
executing-plans
Use when executing an implementation plan in the current session as the implementer yourself — your human partner chose inline execution, or no subagent tool is available
aws-snsImplements Amazon Simple Notification Service (SNS) for building and managing message-driven applications through an event-driven architecture.
Independent GitHub snapshot · Automated classification · Original documentation governs
name: aws-sns description: Implements Amazon Simple Notification Service (SNS) for building and managing message-driven applications through an event-driven architecture. license: MIT compatibility: opencode metadata: version: "1.1.1" domain: cloud triggers: aws sns, event-driven architecture, notification service, publish-subscribe model, scalable messaging archetypes: [implementation, event management] anti_triggers: [hardcoded notification systems] response_profile: verbosity: medium directive_strength: high abstraction_level: operational role: implementation scope: infrastructure output-format: code
Amazon Simple Notification Service (SNS) is a fully managed service for enabling message-driven application architectures with numerous advantages, including:
import boto3
# Create an SNS client
client = boto3.client('sns')
# Create a new SNS topic
response = client.create_topic(Name='MyTopic')
# Publish a message to the topic
response = client.publish(
TopicArn=response['TopicArn'],
Message='This is my first SNS message!',
)
print(response['MessageId'])
By implementing AWS SNS strategies, businesses can efficiently manage messaging needs, enhance application decoupling, and ensure a secure and scalable infrastructure.
import boto3
from typing import Optional
def create_topic_and_subscribe(client, topic_name: str, protocol: str, endpoint: str) -> dict:
"""Create an SNS topic with a subscription."""
topic_resp = client.create_topic(Name=topic_name)
topic_arn = topic_resp["TopicArn"]
sub_resp = client.subscribe(
TopicArn=topic_arn,
Protocol=protocol,
Endpoint=endpoint,
)
print(f"Created subscription: {sub_resp['SubscriptionArn']} for topic {topic_arn}")
return {"topic_arn": topic_arn, "subscription_arn": sub_resp["SubscriptionArn"]}
def publish_with_attributes(client, topic_arn: str, message: str,
event_type: str = "general") -> dict:
"""Publish a message with attributes for subscriber filtering."""
response = client.publish(
TopicArn=topic_arn,
Message=message,
MessageAttributes={
"event_type": {"DataType": "String", "StringValue": event_type},
"priority": {"DataType": "String", "StringValue": "high"},
},
)
print(f"Published message with attributes to {topic_arn}: {response['MessageId']}")
return response
def get_topic_attributes(client, topic_arn: str) -> dict:
"""Get all attributes for an SNS topic."""
return client.get_topic_attributes(TopicArn=topic_arn)
Authoritative documentation links for this domain. The model follows markdown links at load time to resolve external references and inline content.