admin-valentin/arcade-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Reddit Toolkit (#336)

Browse source 
| Name | Description | Package | Version |

|----------------------------|------------------------------------------------------------------|---------|---------|
| Reddit.SubmitTextPost | Submit a text-based post to a subreddit |
Reddit | 0.0.1 |
| Reddit.CommentOnPost | Comment on a Reddit post | Reddit | 0.0.1 |
| Reddit.ReplyToComment | Reply to a Reddit comment | Reddit | 0.0.1 |
| Reddit.GetPostsInSubreddit | Gets posts titles, links, and other
metadata in the specified subreddit | Reddit | 0.0.1 |
| Reddit.GetContentOfPost | Get the content (body) of a Reddit post by
its identifier. | Reddit | 0.0.1 |
| Reddit.GetContentOfMultiplePosts | Get the content (body) of multiple
Reddit posts by their identifiers. | Reddit | 0.0.1 |
| Reddit.GetTopLevelComments | Get the first page of top-level comments
of a Reddit post. | Reddit | 0.0.1 |


### Why not use an SDK?
Reddit API does not have an official SDK, although
[PRAW](https://github.com/praw-dev/praw) has large community support.

I played around with PRAW, but ultimately decided to not use an SDK.
PRAW made it incredibly easy to work with Reddit Objects, but there were
a few drawbacks that ultimately swayed me to not use it:
1. PRAW assumes that it will do the auth for you. A client ID and secret
must be passed to PRAW, but a tool only has the auth token. I was able
to hack around this by manipulating private properties - but it felt too
hacky
2. PRAW does not support Python 3.13
3. PRAW is not async. There is
[AsyncPRAW](https://github.com/praw-dev/asyncpraw), but the community
does not look active there.
This commit is contained in:
Eric Gustin 2025-04-03 13:07:10 -08:00 committed by GitHub
parent e604e8bde3
commit 7c6a739f25
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 1520 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
toolkits/reddit/.pre-commit-config.yaml Normal file
View file

@ -0,0 +1,18 @@
files: ^./
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: "v4.4.0"
hooks:
- id: check-case-conflict
- id: check-merge-conflict
- id: check-toml
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.7
hooks:
- id: ruff
args: [--fix]
- id: ruff-format

44
toolkits/reddit/.ruff.toml Normal file
View file

@ -0,0 +1,44 @@
target-version = "py39"
line-length = 100
fix = true
[lint]
select = [
# flake8-2020
"YTT",
# flake8-bandit
"S",
# flake8-bugbear
"B",
# flake8-builtins
"A",
# flake8-comprehensions
"C4",
# flake8-debugger
"T10",
# flake8-simplify
"SIM",
# isort
"I",
# mccabe
"C90",
# pycodestyle
"E", "W",
# pyflakes
"F",
# pygrep-hooks
"PGH",
# pyupgrade
"UP",
# ruff
"RUF",
# tryceratops
"TRY",
]
[lint.per-file-ignores]
"**/tests/*" = ["S101"]
[format]
preview = true
skip-magic-trailing-comma = false

21
toolkits/reddit/LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025, Arcade
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

53
toolkits/reddit/Makefile Normal file
View file

@ -0,0 +1,53 @@
.PHONY: help
help:
@echo "🛠️ reddit Commands:\n"
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
.PHONY: install
install: ## Install the poetry environment and install the pre-commit hooks
@echo "📦 Checking if Poetry is installed"
@if ! command -v poetry &> /dev/null; then \
echo "📦 Installing Poetry with pip"; \
pip install poetry==1.8.5; \
else \
echo "📦 Poetry is already installed"; \
fi
@echo "🚀 Installing package in development mode with all extras"
poetry install --all-extras
.PHONY: build
build: clean-build ## Build wheel file using poetry
@echo "🚀 Creating wheel file"
poetry build
.PHONY: clean-build
clean-build: ## clean build artifacts
@echo "🗑️ Cleaning dist directory"
rm -rf dist
.PHONY: test
test: ## Test the code with pytest
@echo "🚀 Testing code: Running pytest"
@poetry run pytest -W ignore -v --cov --cov-config=pyproject.toml --cov-report=xml
.PHONY: coverage
coverage: ## Generate coverage report
@echo "coverage report"
coverage report
@echo "Generating coverage report"
coverage html
.PHONY: bump-version
bump-version: ## Bump the version in the pyproject.toml file
@echo "🚀 Bumping version in pyproject.toml"
poetry version patch
.PHONY: check
check: ## Run code quality tools.
@echo "🚀 Checking Poetry lock file consistency with 'pyproject.toml': Running poetry check"
@poetry check
@echo "🚀 Linting code: Running pre-commit"
@poetry run pre-commit run -a
@echo "🚀 Static type checking: Running mypy"
@poetry run mypy --config-file=pyproject.toml

0
toolkits/reddit/arcade_reddit/__init__.py Normal file
View file

28
toolkits/reddit/arcade_reddit/client.py Normal file
View file

@ -0,0 +1,28 @@
from typing import Any
import httpx
class RedditClient:
BASE_URL = "https://oauth.reddit.com/"
def __init__(self, token: str):
self.token = token
async def request(self, method: str, path: str, **kwargs: Any) -> Any:
headers = {
"Authorization": f"Bearer {self.token}",
"User-Agent": "arcade-reddit",
}
async with httpx.AsyncClient() as client:
response = await client.request(
method, f"{self.BASE_URL}/{path.lstrip('/')}", headers=headers, **kwargs
)
response.raise_for_status()
return response.json()
async def get(self, path: str, **kwargs: Any) -> Any:
return await self.request("GET", path, **kwargs)
async def post(self, path: str, **kwargs: Any) -> Any:
return await self.request("POST", path, **kwargs)

47
toolkits/reddit/arcade_reddit/enums.py Normal file
View file

@ -0,0 +1,47 @@
from enum import Enum
class SubredditListingType(str, Enum):
HOT = "hot"
NEW = "new"
RISING = "rising"
TOP = "top" # time-based
CONTROVERSIAL = "controversial" # time-based
def is_time_based(self) -> bool:
return self in [SubredditListingType.TOP, SubredditListingType.CONTROVERSIAL]
class RedditTimeFilter(str, Enum):
NOW = "NOW"
TODAY = "TODAY"
THIS_WEEK = "THIS_WEEK"
THIS_MONTH = "THIS_MONTH"
THIS_YEAR = "THIS_YEAR"
ALL_TIME = "ALL_TIME"
def to_api_value(self) -> str:
_map = {
RedditTimeFilter.NOW: "hour",
RedditTimeFilter.TODAY: "day",
RedditTimeFilter.THIS_WEEK: "week",
RedditTimeFilter.THIS_MONTH: "month",
RedditTimeFilter.THIS_YEAR: "year",
RedditTimeFilter.ALL_TIME: "all",
}
return _map[self]
class RedditThingType(str, Enum):
"""The type of a Reddit 'thing'.
Typically used as a prefix for fullnames, e.g. t1_1234567890
is the fullname of a comment with id 1234567890
"""
COMMENT = "t1"
ACCOUNT = "t2"
LINK = "t3"
MESSAGE = "t4"
SUBREDDIT = "t5"
AWARD = "t6"

19
toolkits/reddit/arcade_reddit/tools/__init__.py Normal file
View file

@ -0,0 +1,19 @@
from arcade_reddit.tools.read import (
get_content_of_post,
get_posts_in_subreddit,
get_top_level_comments,
)
from arcade_reddit.tools.submit import (
comment_on_post,
reply_to_comment,
submit_text_post,
)
__all__ = [
"get_content_of_post",
"get_posts_in_subreddit",
"get_top_level_comments",
"comment_on_post",
"reply_to_comment",
"submit_text_post",
]

125
toolkits/reddit/arcade_reddit/tools/read.py Normal file
View file

@ -0,0 +1,125 @@
from typing import Annotated, Optional
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Reddit
from arcade_reddit.client import RedditClient
from arcade_reddit.enums import (
RedditTimeFilter,
SubredditListingType,
)
from arcade_reddit.utils import (
create_fullname_for_multiple_posts,
create_path_for_post,
normalize_subreddit_name,
parse_get_content_of_multiple_posts_response,
parse_get_content_of_post_response,
parse_get_posts_in_subreddit_response,
parse_get_top_level_comments_response,
remove_none_values,
)
@tool(requires_auth=Reddit(scopes=["read"]))
async def get_posts_in_subreddit(
context: ToolContext,
subreddit: Annotated[str, "The name of the subreddit to fetch posts from"],
listing: Annotated[
SubredditListingType,
(
"The type of listing to fetch. For simple listings such as 'hot', 'new', or 'rising', "
"the 'time_range' parameter is ignored. For time-based listings such as "
"'top' or 'controversial', the 'time_range' parameter is required."
),
] = SubredditListingType.HOT,
limit: Annotated[int, "The maximum number of posts to fetch. Default is 10, max is 100."] = 10,
cursor: Annotated[Optional[str], "The pagination token from a previous call"] = None,
time_range: Annotated[
RedditTimeFilter,
"The time range for filtering posts. Must be provided if the listing type is "
f"{SubredditListingType.TOP.value} or {SubredditListingType.CONTROVERSIAL.value}. "
f"Otherwise, it is ignored. Defaults to {RedditTimeFilter.TODAY.value}.",
] = RedditTimeFilter.TODAY,
) -> Annotated[dict, "A dictionary with a cursor for the next page and a list of posts"]:
"""Gets posts titles, links, and other metadata in the specified subreddit
The time_range is required if the listing type is 'top' or 'controversial'.
"""
client = RedditClient(context.get_auth_token_or_empty())
params = {"limit": limit, "after": cursor}
if listing.is_time_based():
params["t"] = time_range.to_api_value()
params = remove_none_values(params)
subreddit = normalize_subreddit_name(subreddit)
data = await client.get(f"r/{subreddit}/{listing.value}", params=params)
result = parse_get_posts_in_subreddit_response(data)
return result
@tool(requires_auth=Reddit(scopes=["read"]))
async def get_content_of_post(
context: ToolContext,
post_identifier: Annotated[
str,
"The identifier of the Reddit post. "
"The identifier may be a reddit URL to the post, a permalink to the post, "
"a fullname for the post, or a post id.",
],
) -> Annotated[dict, "The content (body) of the Reddit post"]:
"""Get the content (body) of a Reddit post by its identifier."""
client = RedditClient(context.get_auth_token_or_empty())
path = create_path_for_post(post_identifier)
data = await client.get(f"{path}.json")
result = parse_get_content_of_post_response(data)
return result
@tool(requires_auth=Reddit(scopes=["read"]))
async def get_content_of_multiple_posts(
context: ToolContext,
post_identifiers: Annotated[
list[str],
"A list of Reddit post identifiers. "
"The identifiers may be reddit URLs to the posts, permalinks to the posts, "
"fullnames for the posts, or post ids. Must be less than or equal to 100 identifiers.",
],
) -> Annotated[dict, "A dictionary containing the content of multiple Reddit posts"]:
"""Get the content (body) of multiple Reddit posts by their identifiers.
Efficiently retrieve the content of multiple posts in a single request.
Always use this tool to retrieve more than one post's content.
"""
client = RedditClient(context.get_auth_token_or_empty())
fullnames, warnings = create_fullname_for_multiple_posts(post_identifiers)
data = await client.get("api/info.json", params={"id": ",".join(fullnames)})
posts = parse_get_content_of_multiple_posts_response(data)
return {"posts": posts, "warnings": warnings}
@tool(requires_auth=Reddit(scopes=["read"]))
async def get_top_level_comments(
context: ToolContext,
post_identifier: Annotated[
str,
"The identifier of the Reddit post to fetch comments from. "
"The identifier may be a reddit URL, a permalink, a fullname, or a post id.",
],
) -> Annotated[dict, "A dictionary with a list of top level comments"]:
"""Get the first page of top-level comments of a Reddit post."""
client = RedditClient(context.get_auth_token_or_empty())
path = create_path_for_post(post_identifier)
data = await client.get(f"{path}.json")
result = parse_get_top_level_comments_response(data)
return result

113
toolkits/reddit/arcade_reddit/tools/submit.py Normal file
View file

@ -0,0 +1,113 @@
from typing import Annotated, Optional
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Reddit
from arcade_reddit.client import RedditClient
from arcade_reddit.utils import (
create_fullname_for_comment,
create_fullname_for_post,
normalize_subreddit_name,
parse_api_comment_response,
remove_none_values,
)
@tool(requires_auth=Reddit(scopes=["submit"]))
async def submit_text_post(
context: ToolContext,
subreddit: Annotated[str, "The name of the subreddit to which the post will be submitted"],
title: Annotated[str, "The title of the submission"],
body: Annotated[
Optional[str],
"The body of the post in markdown format. Should never be the same as the title",
] = None,
nsfw: Annotated[
Optional[bool],
"Indicates if the submission has content that is 'Not Safe For Work' (NSFW). "
"Default is False",
] = False,
spoiler: Annotated[
Optional[bool],
"Indicates if the post is marked as a spoiler. Default is False",
] = False,
send_replies: Annotated[
Optional[bool], "If true, sends replies to the user's inbox. Default is True"
] = True,
) -> Annotated[dict, "Response from Reddit after submission"]:
"""Submit a text-based post to a subreddit"""
client = RedditClient(context.get_auth_token_or_empty())
subreddit = normalize_subreddit_name(subreddit)
params = {
"api_type": "json",
"sr": subreddit,
"title": title,
"kind": "self",
"nsfw": nsfw,
"spoiler": spoiler,
"sendreplies": send_replies,
"text": body,
}
params = remove_none_values(params)
data = await client.post("api/submit", data=params)
return {"data": data["json"].get("data", {}), "errors": data["json"].get("errors", [])}
@tool(requires_auth=Reddit(scopes=["submit"]))
async def comment_on_post(
context: ToolContext,
post_identifier: Annotated[
str,
"The identifier of the Reddit post. "
"The identifier may be a reddit URL, a permalink, a fullname, or a post id.",
],
text: Annotated[str, "The body of the comment in markdown format"],
) -> Annotated[dict, "Response from Reddit after submission"]:
"""Comment on a Reddit post"""
client = RedditClient(context.get_auth_token_or_empty())
fullname = create_fullname_for_post(post_identifier)
params = {
"api_type": "json",
"thing_id": fullname,
"text": text,
"return_rtjson": True,
}
data = await client.post("api/comment", data=params)
return parse_api_comment_response(data)
@tool(requires_auth=Reddit(scopes=["submit"]))
async def reply_to_comment(
context: ToolContext,
comment_identifier: Annotated[
str,
"The identifier of the Reddit comment to reply to. "
"The identifier may be a comment ID, a reddit URL to the comment, "
"a permalink to the comment, or the fullname of the comment.",
],
text: Annotated[str, "The body of the reply in markdown format"],
) -> Annotated[dict, "Response from Reddit after submission"]:
"""Reply to a Reddit comment"""
client = RedditClient(context.get_auth_token_or_empty())
fullname = create_fullname_for_comment(comment_identifier)
params = {
"api_type": "json",
"thing_id": fullname,
"text": text,
"return_rtjson": True,
}
data = await client.post("api/comment", data=params)
return parse_api_comment_response(data)

375
toolkits/reddit/arcade_reddit/utils.py Normal file
View file

@ -0,0 +1,375 @@
import re
from urllib.parse import urlparse
from arcade.sdk.errors import ToolExecutionError
from arcade_reddit.enums import RedditThingType
def remove_none_values(data: dict) -> dict:
"""Remove all keys with None values from a dictionary"""
return {k: v for k, v in data.items() if v is not None}
def normalize_subreddit_name(subreddit: str) -> str:
"""Normalize a subreddit name"""
return subreddit.lower().replace("r/", "").replace(" ", "")
def _simplify_post_data(post_data: dict, include_body: bool = False) -> dict:
simplified_data = {
"id": post_data.get("id"),
"name": post_data.get("name"),
"title": post_data.get("title"),
"author": post_data.get("author"),
"subreddit": post_data.get("subreddit"),
"created_utc": post_data.get("created_utc"),
"num_comments": post_data.get("num_comments"),
"score": post_data.get("score"),
"upvote_ratio": post_data.get("upvote_ratio"),
"upvotes": post_data.get("ups"),
"permalink": post_data.get("permalink"),
"url": post_data.get("url"),
"is_video": post_data.get("is_video"),
}
if include_body:
simplified_data["body"] = post_data.get("selftext")
return simplified_data
def parse_get_posts_in_subreddit_response(data: dict) -> dict:
"""Parse the response from the Reddit API for getting posts in a subreddit
Associated Reddit API endpoints:
https://www.reddit.com/dev/api/#GET_hot
https://www.reddit.com/dev/api/#GET_new
https://www.reddit.com/dev/api/#GET_rising
https://www.reddit.com/dev/api/#GET_{sort}
Args:
data: The response from the Reddit API deserialized as a dictionary.
NOTE: The response doesn't contain the body of the posts.
Returns:
A dictionary with a cursor for the next page and a list of posts
"""
posts = []
for child in data.get("data", {}).get("children", []):
post_data = child.get("data", {})
post = _simplify_post_data(post_data)
posts.append(post)
result = {"cursor": data.get("data", {}).get("after"), "posts": posts}
return result
def parse_get_content_of_post_response(data: list) -> dict:
"""Parse the json representation of a Reddit post to get the content of a post
Args:
data: The json representation of a Reddit post
(retrieved by appending .json to the permalink)
Returns:
A dictionary with the content of the post
"""
if not data or not isinstance(data, list) or len(data) == 0:
return {}
try:
post_data = data[0].get("data", {}).get("children", [{}])[0].get("data", {})
return _simplify_post_data(post_data, include_body=True)
except (IndexError, AttributeError, KeyError):
return {}
def parse_get_content_of_multiple_posts_response(data: dict) -> list[dict]:
"""Parse the json representation of multiple Reddit posts to get the content of each post
Args:
data: The json representation of multiple Reddit posts
(retrieved from the /api/info.json endpoint)
Returns:
A dictionary with the simplified content of each post
"""
if not data or not isinstance(data, dict) or len(data) == 0:
return []
result = []
for post in data.get("data", {}).get("children", []):
post_data = post.get("data", {})
result.append(_simplify_post_data(post_data, include_body=True))
return result
def parse_get_top_level_comments_response(data: list) -> dict:
"""Parse the json representation of a Reddit post to get the top-level comments
Args:
data: The json representation of a Reddit post
Returns:
A dictionary with a list of top-level comments
"""
try:
comments_listing = data[1]["data"]["children"]
except (IndexError, KeyError):
return {"comments": [], "num_comments": 0}
comments = []
for comment in comments_listing:
if comment.get("kind") != RedditThingType.COMMENT.value:
continue
comment_data = comment.get("data", {})
comments.append({
"id": comment_data.get("id"),
"author": comment_data.get("author"),
"body": comment_data.get("body"),
"score": comment_data.get("score"),
"created_utc": comment_data.get("created_utc"),
})
return {"comments": comments, "num_comments": len(comments)}
def parse_api_comment_response(data: dict) -> dict:
"""Parse the response from the Reddit API's /api/comment endpoint
Args:
data: The response from the Reddit API deserialized as a dictionary
Returns:
A dictionary with the comment data
"""
result = {
"created_utc": data.get("created_utc"),
"name": data.get("name"),
"parent_id": data.get("parent_id"),
"permalink": data.get("permalink"),
"subreddit": data.get("subreddit"),
"subreddit_id": data.get("subreddit_id"),
"subreddit_name_prefixed": data.get("subreddit_name_prefixed"),
}
return result
def _extract_id_from_url(identifier: str, regex: str, error_msg: str) -> str:
"""
Extract an ID from a Reddit URL using the provided regular expression.
Args:
identifier: The URL string from which to extract the ID.
regex: The regular expression pattern containing a capturing group for the ID.
error_msg: The error message to use if no ID can be extracted.
Returns:
The extracted ID as a string.
Raises:
ToolExecutionError: If the URL is not a Reddit URL or the pattern does not match.
"""
parsed = urlparse(identifier)
if not parsed.netloc.endswith("reddit.com"):
raise ToolExecutionError(
message=f"Expected a reddit URL, but got: {identifier}",
developer_message="The identifier should be a valid Reddit URL.",
)
match = re.search(regex, parsed.path)
if not match:
raise ToolExecutionError(
message=f"Could not extract id from URL: {identifier}",
developer_message=error_msg,
)
return match.group(1)
def _extract_id_from_permalink(identifier: str, regex: str, error_msg: str) -> str:
"""
Extract an ID from a Reddit permalink using the provided regular expression.
Args:
identifier: The permalink string from which to extract the ID.
regex: The regular expression pattern containing a capturing group for the ID.
error_msg: The error message to use if no ID can be extracted.
Returns:
The extracted ID as a string.
Raises:
ToolExecutionError: If the pattern does not match the permalink.
"""
match = re.search(regex, identifier)
if not match:
raise ToolExecutionError(
message=f"Could not extract id from permalink: {identifier}",
developer_message=error_msg,
)
return match.group(1)
def _get_post_id(identifier: str) -> str:
"""
Retrieve the post ID from various types of Reddit post identifiers.
The identifier can be a Reddit URL to the post, a permalink for the post,
a fullname for the post (starting with 't3_'), or a raw post ID.
Args:
identifier: The Reddit post identifier.
Returns:
The post ID as a string.
Raises:
ToolExecutionError: If the identifier does not contain a valid post ID.
"""
if identifier.startswith("http://") or identifier.startswith("https://"):
return _extract_id_from_url(
identifier,
r"/comments/([A-Za-z0-9]+)",
"The reddit URL does not contain a valid post id.",
)
elif identifier.startswith("/r/"):
return _extract_id_from_permalink(
identifier,
r"/comments/([A-Za-z0-9]+)",
"The permalink does not contain a valid post id.",
)
else:
pattern = re.compile(r"^(t3_)?([A-Za-z0-9]+)$")
match = pattern.match(identifier)
if match:
return match.group(2)
raise ToolExecutionError(
message=f"Invalid identifier: {identifier}",
developer_message=(
"The identifier should be a valid Reddit URL, permalink, fullname, or post id."
),
)
def _get_comment_id(identifier: str) -> str:
"""
Retrieve the comment ID from various types of Reddit comment identifiers.
The identifier can be a Reddit URL to the comment, a permalink for the comment,
a fullname for the comment (starting with 't1_'), or a raw comment ID.
Args:
identifier: The Reddit comment identifier.
Returns:
The comment ID as a string.
Raises:
ToolExecutionError: If the identifier does not contain a valid comment ID.
"""
if identifier.startswith("http://") or identifier.startswith("https://"):
return _extract_id_from_url(
identifier,
r"/comment/([A-Za-z0-9]+)",
"The reddit URL does not contain a valid comment id.",
)
elif identifier.startswith("/r/"):
return _extract_id_from_permalink(
identifier,
r"/comment/([A-Za-z0-9]+)",
"The permalink does not contain a valid comment id.",
)
else:
if identifier.startswith("t1_"):
return identifier[3:]
if re.fullmatch(r"[A-Za-z0-9]+", identifier):
return identifier
raise ToolExecutionError(
message=f"Invalid identifier: {identifier}",
developer_message=(
"The identifier should be a valid Reddit URL, permalink, fullname, or comment id."
),
)
def create_path_for_post(identifier: str) -> str:
"""
Create a path for a Reddit post.
Args:
identifier: The identifier of the post. The identifier may be a reddit URL,
a permalink for the post, a fullname for the post, or a post id.
Returns:
The path for the post.
"""
if identifier.startswith("http://") or identifier.startswith("https://"):
parsed = urlparse(identifier)
if not parsed.netloc.endswith("reddit.com"):
raise ToolExecutionError(
message=f"Expected a reddit URL, but got: {identifier}",
developer_message="The identifier should be a valid Reddit URL.",
)
return parsed.path
if identifier.startswith("/r/"):
return identifier
post_id = _get_post_id(identifier)
return f"/comments/{post_id}"
def create_fullname_for_post(identifier: str) -> str:
"""
Create a fullname for a Reddit post.
Args:
identifier: The identifier of the post. The identifier may be a reddit URL,
a permalink for the post, a fullname for the post, or a post id.
Returns:
The fullname for the post.
"""
if identifier.startswith("t3_"):
return identifier
post_id = _get_post_id(identifier)
return f"t3_{post_id}"
def create_fullname_for_multiple_posts(post_identifiers: list[str]) -> tuple[list[str], list[dict]]:
"""
Create fullnames for multiple Reddit posts.
Args:
post_identifiers: A list of Reddit post identifiers. The identifiers may be
reddit URLs, permalinks, fullnames, or post ids.
Returns:
(fullnames, warnings): A tuple of a list of fullnames for the posts and
a list of warnings if any of the identifiers are invalid.
"""
fullnames = []
warnings = []
for identifier in post_identifiers:
try:
fullnames.append(create_fullname_for_post(identifier))
except ToolExecutionError:
message = f"'{identifier}' is not a valid Reddit post identifier."
warnings.append({"message": message, "identifier": identifier})
return fullnames, warnings
def create_fullname_for_comment(identifier: str) -> str:
"""
Create a fullname for a Reddit comment.
Args:
identifier: The identifier of the comment. The identifier may be a
reddit URL to the comment, a permalink for the comment, a fullname for
the comment, or a comment id.
Returns:
The fullname for the comment.
"""
if identifier.startswith("t1_"):
return identifier
comment_id = _get_comment_id(identifier)
return f"t1_{comment_id}"

30
toolkits/reddit/evals/additional_messages.py Normal file
View file

@ -0,0 +1,30 @@
get_post_in_subreddit_messages = [
{
"role": "user",
"content": "get 1 post from AskReddit that are contentious at this moment",
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_ZjRic1hS43o10EjutTFgotBh",
"type": "function",
"function": {
"name": "Reddit_GetPostsInSubreddit",
"arguments": '{"subreddit":"AskReddit","listing":"controversial","time_range":"NOW","limit":1}', # noqa: E501
},
}
],
},
{
"role": "tool",
"content": '{"cursor":"t3_1abcdef","posts":[{"author":"StewieGriffin","created_utc":1743457013,"id":"1abcdef","is_video":false,"name":"t3_1abcdef","num_comments":55,"permalink":"/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/","score":2,"subreddit":"AskReddit","title":"Why does my dog have four legs?","upvote_ratio":0.55,"upvotes":2,"url":"https://www.reddit.com/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/"}]}', # noqa: E501
"tool_call_id": "call_ZjRic1hS43o10EjutTFgotBh",
"name": "Reddit_GetPostsInSubreddit",
},
{
"role": "assistant",
"content": "Here's a contentious post from AskReddit that is generating some discussion:\n\n**Title:** Why does my dog have four legs?\n\n- **Author:**StewieGriffin\n- **Score:** 2\n- **Upvote Ratio:** 55%\n- **Number of Comments:** 55\n- **Posted:**[Link](https://www.reddit.com/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/)", # noqa: E501
},
]

42
toolkits/reddit/evals/critics.py Normal file
View file

@ -0,0 +1,42 @@
from dataclasses import dataclass
from typing import Any
from arcade.sdk.eval.critic import Critic
@dataclass
class AnyOfCritic(Critic):
"""
A critic that checks if the actual value matches any of the expected values.
In other words, it checks if the actual value is in the expected list.
"""
def evaluate(self, expected: list[Any], actual: Any) -> dict[str, float | bool]:
match = actual in expected
return {"match": match, "score": self.weight if match else 0.0}
@dataclass
class ListCritic(Critic):
"""
A critic for comparing two lists.
"""
def __init__(
self,
critic_field: str,
weight: float = 1.0,
order_matters: bool = True,
duplicates_matter: bool = True,
):
self.critic_field = critic_field
self.weight = weight
self.order_matters = order_matters
self.duplicates_matter = duplicates_matter
def evaluate(self, expected: list[Any], actual: list[Any]) -> dict[str, float | bool]:
match = actual == expected if self.order_matters else set(actual) == set(expected)
if self.duplicates_matter:
match = match and len(actual) == len(expected)
return {"match": match, "score": self.weight if match else 0.0}

269
toolkits/reddit/evals/eval_reddit_read.py Normal file
View file

@ -0,0 +1,269 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
from arcade.sdk.eval.critic import BinaryCritic
import arcade_reddit
from arcade_reddit.enums import RedditTimeFilter, SubredditListingType
from arcade_reddit.tools import (
get_content_of_post,
get_posts_in_subreddit,
get_top_level_comments,
)
from arcade_reddit.tools.read import get_content_of_multiple_posts
from evals.additional_messages import get_post_in_subreddit_messages
from evals.critics import AnyOfCritic, ListCritic
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
catalog = ToolCatalog()
catalog.add_module(arcade_reddit)
@tool_eval()
def reddit_get_posts_in_subreddit_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_get_posts_in_subreddit_1",
system_message=(
"You are an AI assistant with access to reddit tools. "
"Use them to help the user with their tasks."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="reddit_get_posts_in_subreddit_1",
user_message="Get 30 posts from AskReddit that are contentious at this moment.",
expected_tool_calls=[
ExpectedToolCall(
func=get_posts_in_subreddit,
args={
"subreddit": "AskReddit",
"listing": SubredditListingType.CONTROVERSIAL.value,
"limit": 30,
"cursor": None,
"time_range": RedditTimeFilter.NOW.value,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=0.3),
BinaryCritic(critic_field="listing", weight=0.2),
BinaryCritic(critic_field="limit", weight=0.2),
BinaryCritic(critic_field="cursor", weight=0.1),
BinaryCritic(critic_field="time_range", weight=0.2),
],
)
suite.add_case(
name="reddit_get_posts_in_subreddit_2",
user_message="Get the next 5 posts from AskReddit that are contentious at this moment.",
expected_tool_calls=[
ExpectedToolCall(
func=get_posts_in_subreddit,
args={
"subreddit": "AskReddit",
"listing": SubredditListingType.CONTROVERSIAL.value,
"limit": 5,
"cursor": "t3_1abcdef",
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=0.2),
BinaryCritic(critic_field="listing", weight=0.2),
BinaryCritic(critic_field="limit", weight=0.1),
BinaryCritic(critic_field="cursor", weight=0.3),
BinaryCritic(critic_field="time_range", weight=0.2),
],
additional_messages=get_post_in_subreddit_messages,
)
suite.add_case( # time-based listing, but don't provide a specific time range
name="reddit_get_posts_in_subreddit_3",
user_message="Get 5 top posts from AskReddit",
expected_tool_calls=[
ExpectedToolCall(
func=get_posts_in_subreddit,
args={
"subreddit": "AskReddit",
"listing": SubredditListingType.TOP.value,
"limit": 5,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=0.3),
BinaryCritic(critic_field="listing", weight=0.3),
BinaryCritic(critic_field="limit", weight=0.3),
],
)
suite.add_case(
name="reddit_get_posts_in_subreddit_4",
user_message="Get posts from AskReddit that are gaining traction as we speak",
expected_tool_calls=[
ExpectedToolCall(
func=get_posts_in_subreddit,
args={
"subreddit": "AskReddit",
"listing": SubredditListingType.RISING.value,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=0.3),
BinaryCritic(critic_field="listing", weight=0.7),
],
)
return suite
@tool_eval()
def reddit_get_content_of_post_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_get_content_of_post",
system_message=(
"You are an AI assistant with access to reddit tools. "
"Use them to help the user with their tasks."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="reddit_get_content_of_post_1",
user_message="Get the content of the post with the id t3_1abcdef",
expected_tool_calls=[
ExpectedToolCall(
func=get_content_of_post,
args={
"post_identifier": [ # post_identifier can be any of the following
"1abcdef",
"t3_1abcdef",
"https://www.reddit.com/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/",
"/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/",
],
},
),
],
rubric=rubric,
critics=[
AnyOfCritic(
critic_field="post_identifier",
weight=1.0,
),
],
additional_messages=get_post_in_subreddit_messages,
)
return suite
@tool_eval()
def reddit_get_content_of_multiple_posts_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_get_content_of_multiple_posts",
system_message=(
"You are an AI assistant with access to reddit tools. "
"Use them to help the user with their tasks."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="reddit_get_content_of_multiple_posts_1",
user_message=(
"Get the content of the posts t3_1abcdef, "
"https://www.reddit.com/r/AskReddit/comments/1jdfgk1vn/why_is_water_wet/, "
"t3_3abcdef, t3_4abcdef, and t3_5abcdef, t3_6abcdef, and t3_7abcdef, 4jfnsklf, "
"/r/AskReddit/comments/2dsghr/, and /r/AskReddit/comments/1jdfg35dvn/"
),
expected_tool_calls=[
ExpectedToolCall(
func=get_content_of_multiple_posts,
args={
"post_identifiers": [
"t3_1abcdef",
"https://www.reddit.com/r/AskReddit/comments/1jdfgk1vn/why_is_water_wet/",
"t3_3abcdef",
"t3_4abcdef",
"t3_5abcdef",
"t3_6abcdef",
"t3_7abcdef",
"4jfnsklf",
"/r/AskReddit/comments/2dsghr/",
"/r/AskReddit/comments/1jdfg35dvn/",
],
},
),
],
rubric=rubric,
critics=[
ListCritic(
critic_field="post_identifiers",
weight=1.0,
order_matters=False,
duplicates_matter=True,
),
],
)
return suite
@tool_eval()
def reddit_get_top_level_comments_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_get_top_level_comments",
system_message=(
"You are an AI assistant with access to reddit tools. "
"Use them to help the user with their tasks."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="reddit_get_top_level_comments_1",
user_message="What are people saying in response?",
expected_tool_calls=[
ExpectedToolCall(
func=get_top_level_comments,
args={
"post_identifier": [ # post_identifier can be any of the following
"1abcdef",
"t3_1abcdef",
"https://www.reddit.com/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/",
"/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/",
],
},
),
],
rubric=rubric,
critics=[
AnyOfCritic(
critic_field="post_identifier",
weight=1.0,
),
],
additional_messages=get_post_in_subreddit_messages,
)
return suite

106
toolkits/reddit/evals/eval_reddit_submit.py Normal file
View file

@ -0,0 +1,106 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
from arcade.sdk.eval.critic import BinaryCritic, SimilarityCritic
import arcade_reddit
from arcade_reddit.tools import (
submit_text_post,
)
from arcade_reddit.tools.submit import comment_on_post
from evals.additional_messages import get_post_in_subreddit_messages
from evals.critics import AnyOfCritic
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
catalog = ToolCatalog()
catalog.add_module(arcade_reddit)
@tool_eval()
def reddit_submit_text_post_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_submit_text_post_1",
system_message=(
"You are an AI assistant with access to reddit tools. "
"Use them to help the user with their tasks."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="reddit_submit_text_post_1",
user_message=(
"Post this in AskReddit - 'Why is the sky blue?'. I dont want replies sent to my dms"
),
expected_tool_calls=[
ExpectedToolCall(
func=submit_text_post,
args={
"subreddit": "AskReddit",
"title": "Why is the sky blue?",
"send_replies": False,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=0.3),
BinaryCritic(critic_field="title", weight=0.3),
BinaryCritic(critic_field="body", weight=0.3),
BinaryCritic(critic_field="send_replies", weight=0.1),
],
)
return suite
@tool_eval()
def reddit_comment_on_post_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_comment_on_post_1",
system_message=(
"You are an AI assistant with access to reddit tools. "
"Use them to help the user with their tasks."
),
catalog=catalog,
rubric=rubric,
)
comment = "Your dog's four-legged structure is a manifestation of tetrapodal evolution, where natural selection has optimized limb development for biomechanical stability and efficient terrestrial locomotion. Duh!" # noqa: E501
suite.add_case(
name="reddit_comment_on_post_1",
user_message=(f"here's my comment - {comment}"),
expected_tool_calls=[
ExpectedToolCall(
func=comment_on_post,
args={
"post_identifier": [ # post_identifier can be any of the following
"1abcdef",
"t3_1abcdef",
"https://www.reddit.com/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/",
"/r/AskReddit/comments/1abcdef/why_does_my_dog_have_four_legs/",
],
"text": comment,
},
),
],
rubric=rubric,
critics=[
AnyOfCritic(critic_field="post_identifier", weight=0.5),
SimilarityCritic(critic_field="text", weight=0.5),
],
additional_messages=get_post_in_subreddit_messages,
)
return suite

39
toolkits/reddit/pyproject.toml Normal file
View file

@ -0,0 +1,39 @@
[tool.poetry]
name = "arcade_reddit"
version = "0.0.1"
description = "LLM tools for interacting with Reddit"
authors = ["Arcade <dev@arcade.dev>"]
[tool.poetry.dependencies]
python = "^3.10"
arcade-ai = "^1.0.5"
[tool.poetry.dev-dependencies]
pytest = "^8.3.0"
pytest-cov = "^4.0.0"
mypy = "^1.5.1"
pre-commit = "^3.4.0"
tox = "^4.11.1"
ruff = "^0.7.4"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.mypy]
files = ["arcade_reddit/**/*.py"]
python_version = "3.10"
disallow_untyped_defs = "True"
disallow_any_unimported = "True"
no_implicit_optional = "True"
check_untyped_defs = "True"
warn_return_any = "True"
warn_unused_ignores = "True"
show_error_codes = "True"
ignore_missing_imports = "True"
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.report]
skip_empty = true

0
toolkits/reddit/tests/__init__.py Normal file
View file

191
toolkits/reddit/tests/test_utils.py Normal file
View file

@ -0,0 +1,191 @@
import pytest
from arcade.sdk.errors import ToolExecutionError
from arcade_reddit.utils import (
create_fullname_for_comment,
create_fullname_for_multiple_posts,
create_fullname_for_post,
create_path_for_post,
parse_get_content_of_multiple_posts_response,
parse_get_content_of_post_response,
parse_get_posts_in_subreddit_response,
parse_get_top_level_comments_response,
)
@pytest.mark.parametrize(
"identifier, expected",
[
("abcdef", "t1_abcdef"),
("https://www.reddit.com/r/test/comments/1234567890/comment/1abcdef/", "t1_1abcdef"),
("t1_abcdef", "t1_abcdef"),
("/r/test/comments/1234567890/comment/1abcdef/", "t1_1abcdef"),
("not-an-id", pytest.raises(ToolExecutionError)),
# Fullname: Invalid (not a Reddit comment id, but a Reddit post id)
("t2_abcdef", pytest.raises(ToolExecutionError)),
# URL: Invalid (not a Reddit url to a comment, but to a post)
("https://www.reddit.com/r/test/comments/1234567890", pytest.raises(ToolExecutionError)),
# Permalink: Invalid (not a Reddit permalink to a comment, but to a post)
("/r/test/comments/1234567890", pytest.raises(ToolExecutionError)),
],
)
def test_create_fullname_for_comment(identifier, expected) -> None:
if isinstance(expected, str):
assert create_fullname_for_comment(identifier) == expected
else:
with expected:
create_fullname_for_comment(identifier)
@pytest.mark.parametrize(
"identifier, expected",
[
("https://www.reddit.com/r/test/comments/1234abc/", "/r/test/comments/1234abc/"),
("1234abc", "/comments/1234abc"),
("/r/test/comments/1234abc/", "/r/test/comments/1234abc/"),
("t3_1234abc", "/comments/1234abc"),
# URL: invalid (non-reddit domain)
("https://www.example.com/r/test/comments/1234abc/", pytest.raises(ToolExecutionError)),
# Post ID: invalid (non-alphanumeric character)
("12!abc", pytest.raises(ToolExecutionError)),
# Permalink: invalid (missing the leading "/" so not recognized as a proper permalink)
("r/test/comments/1234abc/", pytest.raises(ToolExecutionError)),
# Fullname: invalid (contains an illegal character)
("t3_1234*abc", pytest.raises(ToolExecutionError)),
],
)
def test_create_path_for_post(identifier, expected):
if isinstance(expected, str):
result = create_path_for_post(identifier)
assert result == expected
else:
with expected:
create_path_for_post(identifier)
@pytest.mark.parametrize(
"identifier, expected",
[
("https://www.reddit.com/r/test/comments/1234abc/", "t3_1234abc"),
("1234abc", "t3_1234abc"),
("/r/test/comments/1234abc/", "t3_1234abc"),
("t3_1234abc", "t3_1234abc"),
# URL: invalid (missing "/comments/" segment)
("https://www.reddit.com/r/test/1234abc/", pytest.raises(ToolExecutionError)),
# Post ID: invalid (non-alphanumeric)
("12!abc", pytest.raises(ToolExecutionError)),
# Permalink: invalid (missing "/comments/" segment)
("/r/test/1234abc/", pytest.raises(ToolExecutionError)),
# Fullname: invalid (type prefix is for a message, not a post
("t4_1234abc", pytest.raises(ToolExecutionError)),
],
)
def test_create_fullname_for_post(identifier, expected):
if isinstance(expected, str):
result = create_fullname_for_post(identifier)
assert result == expected
else:
with expected:
create_fullname_for_post(identifier)
@pytest.mark.parametrize(
"identifiers, expected_fullnames, expected_warnings",
[
([], [], []),
(
["t3_1234abc", "https://www.reddit.com/r/test/comments/1234abc/"],
["t3_1234abc", "t3_1234abc"],
[],
),
(
["t3_1234abc", "not-a-valid-identifier"],
["t3_1234abc"],
[
{
"message": "'not-a-valid-identifier' is not a valid Reddit post identifier.",
"identifier": "not-a-valid-identifier",
}
],
),
(
["inv@lid", "not-a-valid-identifier"],
[],
[
{
"message": "'inv@lid' is not a valid Reddit post identifier.",
"identifier": "inv@lid",
},
{
"message": "'not-a-valid-identifier' is not a valid Reddit post identifier.",
"identifier": "not-a-valid-identifier",
},
],
),
],
)
def test_create_fullname_for_multiple_posts(identifiers, expected_fullnames, expected_warnings):
actual_fullnames, actual_warnings = create_fullname_for_multiple_posts(identifiers)
assert actual_fullnames == expected_fullnames
assert actual_warnings == expected_warnings
def test_parse_get_posts_in_subreddit_response_empty():
data = {}
expected = {"cursor": None, "posts": []}
result = parse_get_posts_in_subreddit_response(data)
assert result == expected
def test_parse_get_content_of_post_response_empty_and_malformed():
data = []
result = parse_get_content_of_post_response(data)
assert result == {}
data = None
result = parse_get_content_of_post_response(data)
assert result == {}
# missing expected keys
data = [{}]
expected = {
"id": None,
"name": None,
"title": None,
"author": None,
"subreddit": None,
"created_utc": None,
"num_comments": None,
"score": None,
"upvote_ratio": None,
"upvotes": None,
"permalink": None,
"url": None,
"is_video": None,
"body": None,
}
result = parse_get_content_of_post_response(data)
assert result == expected
def test_parse_get_content_of_multiple_posts_response_empty_and_malformed():
expected = []
data = {}
result = parse_get_content_of_multiple_posts_response(data)
assert result == expected
data = None
result = parse_get_content_of_multiple_posts_response(data)
assert result == expected
data = [{}]
result = parse_get_content_of_multiple_posts_response(data)
assert result == expected
def test_parse_get_top_level_comments_response_missing_data():
data = [{}]
expected = {"comments": [], "num_comments": 0}
result = parse_get_top_level_comments_response(data)
assert result == expected