send email with ai (#91)

* send email with ai

* schedule workflow

* fix openai

* workflow failure simulate

* restack ai update

Author: justrdk

email_sender/.env.example

@@ -0,0 +1,4 @@
+FROM_EMAIL=[email protected]
+SENDGRID_API_KEY=your-sendgrid-api-key
+OPENAI_API_KEY=your-openai-api-key
+TO_EMAIL=[email protected]

email_sender/README.md

@@ -0,0 +1,26 @@
+[tool.poetry]
+name = "email_sender"
+version = "0.0.1"
+description = "Send emails with sendgrid"
+authors = [
+    "Restack Team <[email protected]>",
+]
+readme = "README.md"
+packages = [{include = "src"}]
+
+[tool.poetry.dependencies]
+python = "3.12.7"
+pydantic = "2.9.2"
+python-dotenv = "1.0.1"
+sendgrid = "6.11.0"
+openai = "1.56.2"
+restack-ai = "0.0.42"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.scripts]
+services = "src.services:run_services"
+schedule = "schedule_workflow:run_schedule_workflow"
+schedule_failure = "schedule_workflow_failure:run_schedule_workflow_failure"

email_sender/poetry.lock

@@ -0,0 +1,45 @@
+import asyncio
+import time
+from restack_ai import Restack
+from dataclasses import dataclass
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+@dataclass
+class InputParams:
+    email_context: str
+    subject: str
+    to: str
+
+async def main():
+    client = Restack()
+
+    workflow_id = f"{int(time.time() * 1000)}-SendEmailWorkflow"
+    to_email = os.getenv("TO_EMAIL")
+    if not to_email:
+        raise Exception("TO_EMAIL environment variable is not set")
+
+    run_id = await client.schedule_workflow(
+        workflow_name="SendEmailWorkflow",
+        workflow_id=workflow_id,
+        input={
+            "email_context": "This email should contain a greeting. And telling user we have launched a new AI feature with Restack workflows. Workflows now offer logging and automatic retries when one of its steps fails. Name of user is not provided. You can set as goodbye message on the email just say 'Best regards' or something like that. No need to mention name of user or name of person sending the email.",
+            "subject": "Hello from Restack",
+            "to": to_email
+        }
+    )
+
+    await client.get_workflow_result(
+        workflow_id=workflow_id,
+        run_id=run_id
+    )
+
+    exit(0)
+
+def run_schedule_workflow():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    run_schedule_workflow()

email_sender/pyproject.toml

@@ -0,0 +1,46 @@
+import asyncio
+import time
+from restack_ai import Restack
+from dataclasses import dataclass
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+@dataclass
+class InputParams:
+    email_context: str
+    subject: str
+    to: str
+
+async def main():
+    client = Restack()
+
+    workflow_id = f"{int(time.time() * 1000)}-SendEmailWorkflow"
+    to_email = os.getenv("TO_EMAIL")
+    if not to_email:
+        raise Exception("TO_EMAIL environment variable is not set")
+
+    run_id = await client.schedule_workflow(
+        workflow_name="SendEmailWorkflow",
+        workflow_id=workflow_id,
+        input={
+            "email_context": "This email should contain a greeting. And telling user we have launched a new AI feature with Restack workflows. Workflows now offer logging and automatic retries when one of its steps fails. Name of user is not provided. You can set as goodbye message on the email just say 'Best regards' or something like that. No need to mention name of user or name of person sending the email.",
+            "subject": "Hello from Restack",
+            "to": to_email,
+            "simulate_failure": True
+        }
+    )
+
+    await client.get_workflow_result(
+        workflow_id=workflow_id,
+        run_id=run_id
+    )
+
+    exit(0)
+
+def run_schedule_workflow_failure():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    run_schedule_workflow_failure()

email_sender/schedule_workflow.py

@@ -0,0 +1,19 @@
+import os
+from restack_ai import Restack
+from restack_ai.restack import CloudConnectionOptions
+from dotenv import load_dotenv  
+
+# Load environment variables from a .env file
+load_dotenv()
+
+
+engine_id = os.getenv("RESTACK_ENGINE_ID")
+address = os.getenv("RESTACK_ENGINE_ADDRESS")
+api_key = os.getenv("RESTACK_ENGINE_API_KEY")
+
+connection_options = CloudConnectionOptions(
+    engine_id=engine_id,
+    address=address,
+    api_key=api_key
+)
+client = Restack(connection_options)

email_sender/schedule_workflow_failure.py

@@ -0,0 +1,45 @@
+from restack_ai.function import function, log, FunctionFailure
+from dataclasses import dataclass
+from openai import OpenAI
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+tries = 0
+
+@dataclass
+class GenerateEmailInput:
+    email_context: str
+    simulate_failure: bool = False
+
+@function.defn()
+async def generate_email_content(input: GenerateEmailInput):
+    global tries
+
+    if input.simulate_failure and tries == 0:
+        tries += 1
+        raise FunctionFailure("Simulated failure", non_retryable=False)
+    
+    if (os.environ.get("OPENAI_API_KEY") is None):
+        raise FunctionFailure("OPENAI_API_KEY is not set", non_retryable=True)
+    
+    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
+
+    response = client.chat.completions.create(
+        model="gpt-4",
+        messages=[
+            {
+                "role": "system",
+                "content": "You are a helpful assistant that generates short emails based on the provided context."
+            },
+            {
+                "role": "user",
+                "content": f"Generate a short email based on the following context: {input.email_context}"
+            }
+        ],
+        max_tokens=150
+    )
+    
+    return response.choices[0].message.content
+

email_sender/src/__init__.py

@@ -0,0 +1,40 @@
+import sendgrid
+from sendgrid.helpers.mail import Mail
+import os
+from restack_ai.function import function, FunctionFailure
+from dataclasses import dataclass
+from dotenv import load_dotenv
+
+load_dotenv()
+
+@dataclass
+class SendEmailInput:
+    text: str
+    subject: str
+    to: str
+
+@function.defn()
+async def send_email(input: SendEmailInput) -> None:
+    from_email = os.getenv('FROM_EMAIL')
+    
+    if not from_email:
+        raise FunctionFailure('FROM_EMAIL is not set', non_retryable=True)
+    
+    sendgrid_api_key = os.getenv('SENDGRID_API_KEY')
+
+    if not sendgrid_api_key:
+        raise FunctionFailure('SENDGRID_API_KEY is not set', non_retryable=True)
+    
+    
+    message = Mail(
+        from_email=from_email,
+        to_emails=input.to,
+        subject=input.subject,
+        plain_text_content=input.text
+    )
+    
+    try:
+        sg = sendgrid.SendGridAPIClient(sendgrid_api_key)
+        sg.send(message)
+    except Exception:
+        raise FunctionFailure('Failed to send email', non_retryable=False)

email_sender/src/client.py

@@ -0,0 +1,19 @@
+import asyncio
+from src.client import client
+from src.workflows.send_email import SendEmailWorkflow
+from src.functions.send_email import send_email
+from src.functions.generate_email_content import generate_email_content
+
+async def main():
+    await asyncio.gather(
+        client.start_service(
+            workflows=[SendEmailWorkflow],
+            functions=[generate_email_content, send_email]
+        )
+    )
+
+def run_services():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    run_services()

email_sender/src/functions/__init__.py

@@ -0,0 +1,44 @@
+from restack_ai.workflow import workflow, import_functions, log, RetryPolicy
+from dataclasses import dataclass
+from datetime import timedelta
+
+with import_functions():
+    from src.functions.send_email import send_email, SendEmailInput
+    from src.functions.generate_email_content import generate_email_content, GenerateEmailInput
+
+@dataclass
+class WorkflowInputParams:
+    email_context: str
+    subject: str
+    to: str
+    simulate_failure: bool = False
+
+@workflow.defn()
+class SendEmailWorkflow:
+    @workflow.run
+    async def run(self, input: WorkflowInputParams):
+        log.info("SendEmailWorkflow started", input=input)
+
+        text = await workflow.step(
+            generate_email_content,
+            GenerateEmailInput(
+                email_context=input.email_context,
+                simulate_failure=input.simulate_failure,
+            ),
+            retry_policy=RetryPolicy(
+                initial_interval=timedelta(seconds=10),
+                backoff_coefficient=1,
+            ),
+        )
+
+        await workflow.step(
+            send_email,
+            SendEmailInput(
+                text=text,
+                subject=input.subject,
+                to=input.to,
+            ),
+            start_to_close_timeout=timedelta(seconds=120)
+        )
+
+        return 'Email sent successfully'