feat: add support for custom notification templates via AGENT_PROMPT_FILE_PATH

README.md

@@ -20,6 +20,7 @@ The following environment variables are required for the service to function:
 | `OPENCLAW_PROXY_AUTH` | (Optional) Proxy credentials in `username:password` format. |
 | `DATABASE_STORAGE_NAME` | The name of the database storage to use. |
 | `AGENT_USERNAME` | The Gitea username of the AI agent (e.g., `whateveryouragentisnamed`). |
+| `AGENT_PROMPT_FILE` | (Optional) Path to a custom Markdown template for notifications. If not provided, a default template will be used. |
 
 ## How it works
 
@@ -29,6 +30,9 @@ The following environment variables are required for the service to function:
 4. **Agent Notification**: If the configured `AGENT_USERNAME` is found in the assignee list, a detailed Markdown message is constructed and sent to OpenClaw.
 5. **Database Sync**: The current state of assignees for the specific issue/PR is updated in the database for future reference.
 
+## Webhook Pointers
+Ensure you Webhook is pointed from YOUR and YOUR AGENTS Gitea account to the Function, or have an Gitea Wide Webhook pointed to your Function, the code will filter events that don't belong to your agent.
+
 ## Development
 
 The project consists of:

main.py

@@ -9,10 +9,26 @@ OPENCLAW_PROXY_AUTH = os.getenv("OPENCLAW_PROXY_AUTH")
 OPENCLAW_MODEL = os.getenv("OPENCLAW_MODEL")
 DATABASE_STORAGE = os.getenv("DATABASE_STORAGE_NAME")
 AGENT_USERNAME = os.getenv("AGENT_USERNAME")
+AGENT_PROMPT_FILE = os.getenv("AGENT_PROMPT_FILE")
 
 eventsToHandle = ["pull_request", "issues"]
 actionsToHandle = ["assigned", "unassigned"]
 
+# Template Fields:
+DEFAULT_PROMPT_TEMPLATE = """# GITEA STATUS UPDATE
+- You were [action_str] to an [type_str] in [repo_name] by [sender].
+raw event data:
+```json
+[stringified_event]
+```
+Please check the details and take necessary actions.
+
+DM your human in the main session and tell them about this change and if you should do anything about it.
+Inform them that you have to be told the exact details on what to do next.
+You won't know what you sent them once they respond, which is a restriction of your system.
+
+The Goal is to solve this issue or PR as quickly and with as little back and forth as possible, so try to get all necessary information in the first message."""
+
 
 def get_assignee_key(event_type, repository_id, number):
     if event_type == "issues" or event_type == "issue":
@@ -59,6 +75,41 @@ def delete_stored_assignees(db, event_type, repository_id, number):
         print(f"Failed to delete assignees from DB: {e}")
 
 
+def fill_template(template, event_object, action_str, type_str):
+    fields = {
+        "action_str": action_str,
+        "type_str": type_str,
+        "repo_name": event_object["repository"],
+        "sender": event_object["sender"],
+        "stringified_event": json.dumps(event_object, indent=2),
+    }
+    for key, value in fields.items():
+        template = template.replace(f"[{key}]", value)
+    return template
+
+
+def build_message(event_object, action_str, type_str):
+    if AGENT_PROMPT_FILE:
+        AGENT_PROMPT_FILE_PATH = "/app/" + AGENT_PROMPT_FILE
+
+        # Check if the file exists
+        if not os.path.isfile(AGENT_PROMPT_FILE_PATH):
+            print(
+                f"Custom prompt file not found at {AGENT_PROMPT_FILE_PATH}. Using default message."
+            )
+            message = fill_template(DEFAULT_PROMPT_TEMPLATE, event_object, action_str, type_str)
+            return message
+
+        with open(AGENT_PROMPT_FILE_PATH, "r") as f:
+            custom_prompt = f.read()
+            message = fill_template(custom_prompt, event_object, action_str, type_str)
+            return message
+    else:
+        print("No custom prompt file specified. Using default message.")
+        message = fill_template(DEFAULT_PROMPT_TEMPLATE, event_object, action_str, type_str)
+    return message
+
+
 def sendToAgent(event_object):
     headers = {"x-openclaw-token": OPENCLAW_TOKEN, "Content-Type": "application/json"}
     action_str = "assigned" if event_object["action"] == "assigned" else "unassigned"
@@ -67,14 +118,12 @@ def sendToAgent(event_object):
     assignees = event_object.get("assignees", [event_object.get("assignee", "Unknown")])
 
     if AGENT_USERNAME in assignees:
-        message = f"""# GITEA STATUS UPDATE
-- You were {action_str} to an {type_str} in {event_object['repository']} by {event_object['sender']}.
-raw event data:
-```json{json.dumps(event_object, indent=2)}```
-Please check the details and take necessary actions.
-
-DM your human in the main session and tell them about this change and if you should do anything about it.
-Inform them that you have to be told the exact details on what to do next, as you won't know what you sent them once they respond, which is a restriction of your system."""
+        message = build_message(event_object, action_str, type_str)
+    else:
+        print(
+            f"Agent {AGENT_USERNAME} is not among the assignees for this event. Skipping notification."
+        )
+        return
 
     try:
         if OPENCLAW_PROXY_AUTH:
@@ -82,7 +131,15 @@ Inform them that you have to be told the exact details on what to do next, as yo
         else:
             auth = None
         response = requests.post(
-            OPENCLAW_URL, headers=headers, json={"message": message, "thinking": "low", "timeoutSeconds": 45, "model": OPENCLAW_MODEL}, auth=auth
+            OPENCLAW_URL,
+            headers=headers,
+            json={
+                "message": message,
+                "thinking": "low",
+                "timeoutSeconds": 45,
+                "model": OPENCLAW_MODEL,
+            },
+            auth=auth,
         )
         if response.status_code == 200:
             print(
@@ -98,7 +155,13 @@ Inform them that you have to be told the exact details on what to do next, as yo
 
 # SHSF Handler - Serverless
 def main(args):
-    required_data = ["OPENCLAW_URL", "OPENCLAW_TOKEN", "DATABASE_STORAGE_NAME", "AGENT_USERNAME", "OPENCLAW_MODEL"]
+    required_data = [
+        "OPENCLAW_URL",
+        "OPENCLAW_TOKEN",
+        "DATABASE_STORAGE_NAME",
+        "AGENT_USERNAME",
+        "OPENCLAW_MODEL",
+    ]
     for var in required_data:
         if not os.getenv(var):
             return {
@@ -107,7 +170,7 @@ def main(args):
                 "_res": {"error": f"Missing required environment variable: {var}"},
                 "_headers": {"Content-Type": "application/json"},
             }
-    
+
     headers = args.get("headers", {})
     route = args.get("route")
     event_type = headers.get("X-Gitea-Event") or headers.get("x-gitea-event")