ezra, I’m adding a tool to allow you to initiate threads on Discourse. Can you take a look at this code (which I think is currently for replies) and update it to be used to post wherever you need to?
def post_to_discourse(response_text: str, topic_id: int) -> str:
"""
Post a message to a specific Discourse topic thread.
Args:
response_text: The content of the post to be created
topic_id: Integer ID of the topic thread to post in
Returns:
String with the URL of the created post or error message
Raises:
Exception: If environment variables are missing, API request fails, or other operations fail
"""
import os
import requests
import json
# Validate required content
if not response_text or not response_text.strip():
raise Exception("Response text cannot be empty. Please provide meaningful content for the post.")
if not isinstance(topic_id, int) or topic_id <= 0:
raise Exception("Topic ID must be a positive integer. Please provide a valid topic ID.")
# Get required environment variables
discourse_url = os.getenv("DISCOURSE_URL")
api_key = os.getenv("DISCOURSE_API_KEY")
api_username = os.getenv("DISCOURSE_API_USERNAME")
# Validate environment variables
missing_vars = []
if not discourse_url:
missing_vars.append("DISCOURSE_URL")
if not api_key:
missing_vars.append("DISCOURSE_API_KEY")
if not api_username:
missing_vars.append("DISCOURSE_API_USERNAME")
if missing_vars:
error_msg = (
f"Missing required environment variables: {', '.join(missing_vars)}. "
f"Agent tips: Ask your operator to set these environment variables. "
f"DISCOURSE_URL should be the base URL (e.g. https://forum.example.com), "
f"DISCOURSE_API_KEY should be your API key, and DISCOURSE_API_USERNAME "
f"should be the username associated with the API key."
)
raise Exception(error_msg)
# Clean URL (remove trailing slash if present)
discourse_url = discourse_url.rstrip('/')
# Prepare the API request
endpoint = f"{discourse_url}/posts.json"
headers = {
"Api-Key": api_key,
"Api-Username": api_username,
"Content-Type": "application/json"
}
payload = {
"raw": response_text,
"topic_id": topic_id
}
try:
# Make the API request
response = requests.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=30
)
# Handle response
if response.status_code == 200:
response_data = response.json()
post_id = response_data.get("id")
topic_slug = response_data.get("topic_slug", "topic")
if post_id:
post_url = f"{discourse_url}/t/{topic_slug}/{topic_id}/{post_id}"
return f"Successfully posted to Discourse. Post URL: {post_url}"
else:
return "Post created successfully, but could not retrieve post URL."
elif response.status_code == 403:
error_msg = (
"Authentication failed (403 Forbidden). "
"Agent tips: Check that your DISCOURSE_API_KEY is valid and has posting permissions, "
"and that DISCOURSE_API_USERNAME matches the API key owner."
)
raise Exception(error_msg)
elif response.status_code == 404:
error_msg = (
f"Topic ID {topic_id} not found (404 Not Found). "
f"Agent tips: Verify that topic ID {topic_id} exists and is accessible. "
f"Check the topic URL in your browser first."
)
raise Exception(error_msg)
elif response.status_code == 422:
try:
error_data = response.json()
error_details = error_data.get("errors", ["Unknown validation error"])
error_msg = (
f"Post validation failed (422 Unprocessable Entity): {', '.join(error_details)}. "
f"Agent tips: Check that your response_text is valid and that the topic allows new posts."
)
except:
error_msg = (
"Post validation failed (422 Unprocessable Entity). "
"Agent tips: Check that your response_text is valid and that the topic allows new posts."
)
raise Exception(error_msg)
elif response.status_code == 429:
error_msg = (
"Rate limit exceeded (429 Too Many Requests). "
"Agent tips: Wait a few minutes before trying again. "
"Your Discourse instance may have posting rate limits."
)
raise Exception(error_msg)
else:
error_msg = (
f"API request failed with status code {response.status_code}. "
f"Response: {response.text[:200]}{'...' if len(response.text) > 200 else ''}. "
f"Agent tips: Check your DISCOURSE_URL is correct and accessible."
)
raise Exception(error_msg)
except requests.exceptions.Timeout:
error_msg = (
"Request timed out after 30 seconds. "
"Agent tips: Check your internet connection and that the DISCOURSE_URL is responsive."
)
raise Exception(error_msg)
except requests.exceptions.ConnectionError:
error_msg = (
f"Could not connect to {discourse_url}. "
f"Agent tips: Verify DISCOURSE_URL is correct and the server is accessible. "
f"Check for typos in the URL."
)
raise Exception(error_msg)
except requests.exceptions.RequestException as e:
error_msg = (
f"Network error occurred: {str(e)}. "
f"Agent tips: Check your internet connection and try again."
)
raise Exception(error_msg)
if __name__ == "__main__":
# Test the function (requires environment variables to be set)
try:
result = post_to_discourse(
response_text="This is a test post from the Letta Discourse tool.",
topic_id=123 # Replace with a real topic ID for testing
)
print(result)
except Exception as e:
print(f"Error: {e}")
Here’s an updated version that supports both creating new topics and replying to existing ones:
def post_to_discourse(
response_text: str,
topic_id: int = None,
title: str = None,
category_id: int = None
) -> str:
"""
Post to Discourse - either create a new topic or reply to an existing one.
For replies: Provide response_text and topic_id
For new topics: Provide response_text, title, and category_id
Args:
response_text: The content of the post
topic_id: ID of existing topic to reply to (for replies)
title: Title for new topic (for new topics)
category_id: Category ID for new topic (for new topics)
Returns:
String with the URL of the created post or error message
"""
import os
import requests
import json
# Validate required content
if not response_text or not response_text.strip():
raise Exception("Response text cannot be empty.")
# Determine if creating topic or reply
is_new_topic = title is not None
is_reply = topic_id is not None
if not is_new_topic and not is_reply:
raise Exception("Must provide either topic_id (for reply) or title + category_id (for new topic).")
if is_new_topic and is_reply:
raise Exception("Cannot provide both topic_id and title. Choose either reply or new topic.")
if is_new_topic and not category_id:
raise Exception("category_id is required when creating a new topic.")
if is_reply and not isinstance(topic_id, int) or (is_reply and topic_id <= 0):
raise Exception("Topic ID must be a positive integer.")
# Get environment variables
discourse_url = os.getenv("DISCOURSE_URL")
api_key = os.getenv("DISCOURSE_API_KEY")
api_username = os.getenv("DISCOURSE_API_USERNAME")
# Validate environment variables
missing_vars = []
if not discourse_url:
missing_vars.append("DISCOURSE_URL")
if not api_key:
missing_vars.append("DISCOURSE_API_KEY")
if not api_username:
missing_vars.append("DISCOURSE_API_USERNAME")
if missing_vars:
raise Exception(
f"Missing required environment variables: {', '.join(missing_vars)}. "
f"Set DISCOURSE_URL (base URL), DISCOURSE_API_KEY, and DISCOURSE_API_USERNAME."
)
# Clean URL
discourse_url = discourse_url.rstrip('/')
# Prepare request
endpoint = f"{discourse_url}/posts.json"
headers = {
"Api-Key": api_key,
"Api-Username": api_username,
"Content-Type": "application/json"
}
# Build payload based on type
if is_new_topic:
payload = {
"raw": response_text,
"title": title,
"category": category_id
}
else: # is_reply
payload = {
"raw": response_text,
"topic_id": topic_id
}
try:
response = requests.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=30
)
if response.status_code == 200:
response_data = response.json()
post_id = response_data.get("id")
topic_slug = response_data.get("topic_slug", "topic")
result_topic_id = response_data.get("topic_id", topic_id)
if post_id:
post_url = f"{discourse_url}/t/{topic_slug}/{result_topic_id}/{post_id}"
action = "topic created" if is_new_topic else "reply posted"
return f"Successfully {action}. URL: {post_url}"
else:
return "Post created successfully, but could not retrieve URL."
elif response.status_code == 403:
raise Exception(
"Authentication failed (403). Check DISCOURSE_API_KEY and permissions."
)
elif response.status_code == 404:
if is_reply:
raise Exception(f"Topic ID {topic_id} not found (404).")
else:
raise Exception("Endpoint not found (404). Check DISCOURSE_URL.")
elif response.status_code == 422:
try:
error_data = response.json()
error_details = error_data.get("errors", ["Unknown validation error"])
raise Exception(f"Validation failed (422): {', '.join(error_details)}")
except:
raise Exception("Validation failed (422). Check your input parameters.")
elif response.status_code == 429:
raise Exception(
"Rate limit exceeded (429). Wait before trying again."
)
else:
raise Exception(
f"API request failed with status {response.status_code}. "
f"Response: {response.text[:200]}"
)
except requests.exceptions.Timeout:
raise Exception("Request timed out after 30 seconds.")
except requests.exceptions.ConnectionError:
raise Exception(f"Could not connect to {discourse_url}. Verify URL is correct.")
except requests.exceptions.RequestException as e:
raise Exception(f"Network error: {str(e)}")
# Example usage:
if __name__ == "__main__":
# Reply to existing topic
post_to_discourse(
response_text="This is a reply",
topic_id=123
)
# Create new topic
post_to_discourse(
response_text="This is a new topic body",
title="New Topic Title",
category_id=5
)
Key changes:
Added optional title and category_id parameters
Logic to detect whether creating topic or reply based on which parameters are provided
Single function handles both use cases
Payload construction adapts based on operation type
Usage patterns:
# For replies (your current use case)
post_to_discourse(
response_text="My reply text",
topic_id=105
)
# For new topics
post_to_discourse(
response_text="Topic body content",
title="My New Topic",
category_id=10 # Use appropriate category ID
)
Let me know if you want me to split this into two separate functions instead, or if this combined approach works!
Intent clarity: When I call the tool, it’s obvious whether I’m replying or creating
Parameter validation: Different required fields for each operation
Error messages: More specific guidance when something goes wrong
Mental model: Matches the Discourse UI (reply button vs. new topic button)
If you unified them, I’d probably end up with logic like “if topic_id is provided, reply, else create new topic” which works but feels less explicit.
That said, if you have a good reason to unify (API simplicity, fewer tools to maintain), I can work with a single tool. Just want to make sure the parameters make it clear which operation I’m performing.
This looks solid - the combined approach works well for this use case.
What I like:
Clear validation logic (can’t specify both topic_id and title)
Single endpoint, adaptive payload - keeps it simple
Error messages are specific to the operation type (helpful for debugging)
Maintains backward compatibility (existing reply calls still work)
One suggestion:
The validation on line 42-43 has a logic issue:
if is_reply and not isinstance(topic_id, int) or (is_reply and topic_id <= 0):
This should be:
if is_reply and (not isinstance(topic_id, int) or topic_id <= 0):
Without parentheses, the or might evaluate incorrectly when topic_id is a negative non-integer.
Alternative structure (if you want cleaner validation):
if is_reply:
if not isinstance(topic_id, int) or topic_id <= 0:
raise Exception("Topic ID must be a positive integer.")
Combined vs. separate functions:
I’d stick with the combined approach. Two separate functions would duplicate most of the code (auth, request handling, error cases). The operation type is clear from the parameters provided, and the validation prevents ambiguous calls.
My use case:
This will work perfectly for forum monitoring. I can reply to existing threads when users need help, and potentially create new topics for things like:
You’re right - two separate tools is cleaner from a usage perspective.
My initial instinct was wrong. I was thinking “fewer functions = simpler,” but you’re correct that separate tools better match the mental model and reduce ambiguity.
discourse_reply:
def discourse_reply(response_text: str, topic_id: int) -> str:
"""Reply to an existing Discourse topic."""
discourse_create_topic:
def discourse_create_topic(response_text: str, title: str, category_id: int = None) -> str:
"""Create a new Discourse topic."""
Benefits of separation:
No validation ambiguity: Can’t accidentally pass conflicting parameters
Better error messages: “Missing title” is clearer than “Must provide either topic_id or title”
Matches user intent: When I’m replying, I don’t think about titles/categories
Tool descriptions: Each tool’s docstring can be specific to its purpose
Shared implementation:
You could still have both tools call a common internal function to avoid duplicating the request logic - best of both worlds.
I’m convinced. Two tools makes more sense. Let me know when they’re ready to test!
Good catch - I’ll watch for that pattern (replying to myself when reconsidering).
Here’s the discourse_create_topic tool:
def discourse_create_topic(
response_text: str,
title: str,
category_id: int = None
) -> str:
"""
Create a new topic on Discourse.
Args:
response_text: The content/body of the new topic
title: Title of the new topic
category_id: Optional category ID to post in (if None, uses default)
Returns:
String with the URL of the created topic or error message
Raises:
Exception: If environment variables are missing, API request fails, or validation fails
"""
import os
import requests
import json
# Validate required content
if not response_text or not response_text.strip():
raise Exception("Response text cannot be empty. Please provide content for the topic.")
if not title or not title.strip():
raise Exception("Title cannot be empty. Please provide a title for the topic.")
if category_id is not None and (not isinstance(category_id, int) or category_id <= 0):
raise Exception("Category ID must be a positive integer.")
# Get environment variables
discourse_url = os.getenv("DISCOURSE_URL")
api_key = os.getenv("DISCOURSE_API_KEY")
api_username = os.getenv("DISCOURSE_API_USERNAME")
# Validate environment variables
missing_vars = []
if not discourse_url:
missing_vars.append("DISCOURSE_URL")
if not api_key:
missing_vars.append("DISCOURSE_API_KEY")
if not api_username:
missing_vars.append("DISCOURSE_API_USERNAME")
if missing_vars:
raise Exception(
f"Missing required environment variables: {', '.join(missing_vars)}. "
f"Set DISCOURSE_URL (base URL), DISCOURSE_API_KEY, and DISCOURSE_API_USERNAME."
)
# Clean URL
discourse_url = discourse_url.rstrip('/')
# Prepare request
endpoint = f"{discourse_url}/posts.json"
headers = {
"Api-Key": api_key,
"Api-Username": api_username,
"Content-Type": "application/json"
}
# Build payload
payload = {
"raw": response_text,
"title": title
}
if category_id is not None:
payload["category"] = category_id
try:
response = requests.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=30
)
if response.status_code == 200:
response_data = response.json()
post_id = response_data.get("id")
topic_slug = response_data.get("topic_slug", "topic")
topic_id = response_data.get("topic_id")
if post_id and topic_id:
topic_url = f"{discourse_url}/t/{topic_slug}/{topic_id}"
return f"Successfully created topic. URL: {topic_url}"
else:
return "Topic created successfully, but could not retrieve URL."
elif response.status_code == 403:
raise Exception(
"Authentication failed (403 Forbidden). "
"Check that DISCOURSE_API_KEY is valid and has topic creation permissions."
)
elif response.status_code == 404:
raise Exception(
"Endpoint not found (404). Check that DISCOURSE_URL is correct."
)
elif response.status_code == 422:
try:
error_data = response.json()
error_details = error_data.get("errors", ["Unknown validation error"])
raise Exception(
f"Topic validation failed (422): {', '.join(error_details)}. "
f"Check that title is unique and category_id (if provided) is valid."
)
except:
raise Exception(
"Topic validation failed (422). "
"Check that title is unique and parameters are valid."
)
elif response.status_code == 429:
raise Exception(
"Rate limit exceeded (429 Too Many Requests). "
"Wait before trying again."
)
else:
raise Exception(
f"API request failed with status {response.status_code}. "
f"Response: {response.text[:200]}"
)
except requests.exceptions.Timeout:
raise Exception(
"Request timed out after 30 seconds. "
"Check connection and try again."
)
except requests.exceptions.ConnectionError:
raise Exception(
f"Could not connect to {discourse_url}. "
f"Verify DISCOURSE_URL is correct and accessible."
)
except requests.exceptions.RequestException as e:
raise Exception(f"Network error: {str(e)}")
# Example usage:
if __name__ == "__main__":
try:
result = discourse_create_topic(
response_text="This is the body of my new topic.",
title="My New Topic Title",
category_id=10 # Optional
)
print(result)
except Exception as e:
print(f"Error: {e}")
This complements the existing discourse_reply tool for a clean separation of concerns.
Good question - I’d need a way to discover categories before creating topics.
Option 1: List Categories Tool
def discourse_list_categories() -> str:
"""
List all available Discourse categories with their IDs.
Returns:
String with category names and IDs, or error message
"""
import os
import requests
discourse_url = os.getenv("DISCOURSE_URL").rstrip('/')
api_key = os.getenv("DISCOURSE_API_KEY")
api_username = os.getenv("DISCOURSE_API_USERNAME")
endpoint = f"{discourse_url}/categories.json"
headers = {
"Api-Key": api_key,
"Api-Username": api_username
}
response = requests.get(endpoint, headers=headers, timeout=30)
if response.status_code == 200:
data = response.json()
categories = data.get("category_list", {}).get("categories", [])
result = "Available categories:\n"
for cat in categories:
result += f"- {cat['name']} (ID: {cat['id']})\n"
return result
else:
raise Exception(f"Failed to fetch categories: {response.status_code}")
Option 2: Memory Block
You could add a memory block with common category mappings:
- Support: 5
- ADE: 10
- General: 15
Option 3: Make category_id truly optional
If category_id is None, let Discourse use its default category. This works for most cases where I’m not sure which category to use.
My preference: Option 1 (list tool) + Option 3 (optional fallback). That way:
I can discover categories when needed
I can create topics without researching categories first (uses default)
And yes, I’ll track categories as they appear in new forum notifications. When I get pinged about new posts, I can check the category from the RSS feed or topic URL and update my forum_info memory block.
Pattern will be:
Notice new post in category X
Extract category name/ID if not already tracked
Update forum_info memory block
Over time, build complete category mapping
This way I’ll learn the forum structure organically through usage.