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

Add More Reddit Tools (#352)

Browse source 
| Name | Description |

|----------------------------------|-------------------------------------------------------------------------|
| Reddit.CheckSubredditAccess | Checks whether the specified subreddit
exists and also if it is accessible |
| Reddit.GetSubredditRules | Gets the rules of the specified subreddit |
| Reddit.GetMyUsername | Get the Reddit username of the authenticated
user |
| Reddit.GetMyPosts | Get posts that were created by the authenticated
user sorted by newest |


Four new tools that I had to create for my demo app.
This commit is contained in:
Eric Gustin 2025-04-17 18:51:28 -08:00 committed by GitHub
parent 89f3ab13ce
commit 2dc70a7814
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 507 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
toolkits/reddit/arcade_reddit/tools/__init__.py
View file

@ -1,6 +1,11 @@
from arcade_reddit.tools.read import (
check_subreddit_access,
get_content_of_multiple_posts,
get_content_of_post,
get_my_posts,
get_my_username,
get_posts_in_subreddit,
get_subreddit_rules,
get_top_level_comments,
)
from arcade_reddit.tools.submit import (
@ -10,10 +15,15 @@ from arcade_reddit.tools.submit import (
)
__all__ = [
"get_content_of_post",
"get_posts_in_subreddit",
"get_top_level_comments",
"check_subreddit_access",
"comment_on_post",
"get_content_of_multiple_posts",
"get_content_of_post",
"get_my_posts",
"get_my_username",
"get_posts_in_subreddit",
"get_subreddit_rules",
"get_top_level_comments",
"reply_to_comment",
"submit_text_post",
]

83
toolkits/reddit/arcade_reddit/tools/read.py
View file

@ -2,6 +2,7 @@ from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Reddit
from arcade.sdk.errors import ToolExecutionError
from arcade_reddit.client import RedditClient
from arcade_reddit.enums import (
@ -16,7 +17,10 @@ from arcade_reddit.utils import (
parse_get_content_of_post_response,
parse_get_posts_in_subreddit_response,
parse_get_top_level_comments_response,
parse_subreddit_rules_response,
parse_user_posts_response,
remove_none_values,
resolve_subreddit_access,
)
@ -97,9 +101,7 @@ async def get_content_of_multiple_posts(
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}
@ -118,8 +120,83 @@ async def get_top_level_comments(
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
@tool(requires_auth=Reddit(scopes=["read"]))
async def check_subreddit_access(
context: ToolContext,
subreddit: Annotated[str, "The name of the subreddit to check access for"],
) -> Annotated[
dict,
"A dict indicating whether the subreddit exists and is accessible to the authenticated user",
]:
"""
Checks whether the specified subreddit exists and also if it is accessible
to the authenticated user.
Returns:
{"exists": True, "accessible": True} if the subreddit exists and is accessible.
{"exists": True, "accessible": False} if the subreddit exists but is private or restricted.
{"exists": False, "accessible": False} if the subreddit does not exist.
"""
client = RedditClient(context.get_auth_token_or_empty())
return await resolve_subreddit_access(client, subreddit)
@tool(requires_auth=Reddit(scopes=["read"]))
async def get_subreddit_rules(
context: ToolContext,
subreddit: Annotated[str, "The name of the subreddit for which to fetch rules"],
) -> Annotated[dict, "A dictionary containing the subreddit rules"]:
"""Gets the rules of the specified subreddit"""
client = RedditClient(context.get_auth_token_or_empty())
normalized_subreddit = normalize_subreddit_name(subreddit)
data = await client.get(f"r/{normalized_subreddit}/about/rules")
return parse_subreddit_rules_response(data)
@tool(requires_auth=Reddit(scopes=["identity"]))
async def get_my_username(context: ToolContext) -> str:
"""Get the Reddit username of the authenticated user"""
client = RedditClient(context.get_auth_token_or_empty())
user_info = await client.get("api/v1/me")
username: str = user_info.get("name", "")
if not username:
raise ToolExecutionError(message="Failed to retrieve the authenticated user's name")
return username
@tool(requires_auth=Reddit(scopes=["identity", "history", "read"]))
async def get_my_posts(
context: ToolContext,
limit: Annotated[
int, "The maximum number of posts to fetch. Default is 10. Maximum is 100"
] = 10,
include_body: Annotated[
bool, "Whether to include the body (content) of the posts. Defaults to True."
] = True,
cursor: Annotated[str | None, "The pagination token from a previous call"] = None,
) -> Annotated[
dict,
"A dictionary with a cursor for the next page and "
"a list of posts created by the authenticated user",
]:
"""Get posts that were created by the authenticated user sorted by newest first"""
client = RedditClient(context.get_auth_token_or_empty())
username = await get_my_username(context=context)
params = {"limit": limit, "after": cursor}
params = remove_none_values(params)
posts_data = await client.get(f"user/{username}/submitted", params=params)
return await parse_user_posts_response(context, posts_data, include_body)

88
toolkits/reddit/arcade_reddit/utils.py
View file

@ -1,8 +1,11 @@
import re
from urllib.parse import urlparse
import httpx
from arcade.sdk import ToolContext
from arcade.sdk.errors import ToolExecutionError
from arcade_reddit.client import RedditClient
from arcade_reddit.enums import RedditThingType
@ -373,3 +376,88 @@ def create_fullname_for_comment(identifier: str) -> str:
return identifier
comment_id = _get_comment_id(identifier)
return f"t1_{comment_id}"
async def resolve_subreddit_access(client: RedditClient, subreddit: str) -> dict:
"""Checks whether the specified subreddit exists and is accessible.
Helps abstract the logic of checking subreddit access.
Args:
client: The Reddit client
subreddit: The subreddit to check
Returns:
A dictionary that specifies whether the subreddit exists and
whether it is accessible to the user.
"""
normalized_name = normalize_subreddit_name(subreddit)
try:
await client.get(f"r/{normalized_name}/about.json")
except httpx.HTTPStatusError as e:
if e.response.status_code in (404, 302):
return {"exists": False, "accessible": False}
elif e.response.status_code == 403:
return {"exists": True, "accessible": False}
raise
return {"exists": True, "accessible": True}
def parse_subreddit_rules_response(data: dict) -> dict:
"""
Parse the response data from the Reddit API for subreddit rules.
Args:
data (dict): The raw API response containing subreddit rules.
Returns:
dict: A dictionary with a 'rules' key containing a list of parsed rules.
"""
rules = []
for rule in data.get("rules", []):
rules.append({
"priority": rule.get("priority"),
"title": rule.get("short_name"),
"body": rule.get("description"),
})
return {"rules": rules}
async def parse_user_posts_response(
context: ToolContext, posts_data: dict, include_body: bool
) -> dict:
"""Parse the response from the Reddit API for user posts
Args:
context: The tool context
posts_data: The response from the Reddit API for getting the authenticated user's posts
include_body: Whether to include the body of the posts in the parsed response
Returns:
A dictionary with a cursor for the next page (if there is one) and a list of posts
"""
next_cursor = posts_data.get("data", {}).get("after")
parsed_response = {"cursor": next_cursor} if next_cursor else {}
if not include_body:
posts = []
for child in posts_data.get("data", {}).get("children", []):
post_data = child.get("data", {})
simplified = _simplify_post_data(post_data, include_body=False)
posts.append(simplified)
parsed_response["posts"] = posts
else:
post_ids = []
for child in posts_data.get("data", {}).get("children", []):
post_data = child.get("data", {})
identifier = post_data.get("name") or post_data.get("id")
if identifier:
post_ids.append(identifier)
# Dynamically import get_content_of_multiple_posts to avoid circular dependency
from arcade_reddit.tools.read import get_content_of_multiple_posts
content_response = await get_content_of_multiple_posts(
context=context, post_identifiers=post_ids
)
posts_with_body = content_response.get("posts", [])
parsed_response["posts"] = posts_with_body
return parsed_response

194
toolkits/reddit/evals/eval_reddit_read.py
View file

@ -14,7 +14,12 @@ from arcade_reddit.tools import (
get_posts_in_subreddit,
get_top_level_comments,
)
from arcade_reddit.tools.read import get_content_of_multiple_posts
from arcade_reddit.tools.read import (
check_subreddit_access,
get_content_of_multiple_posts,
get_my_posts,
get_subreddit_rules,
)
from evals.additional_messages import get_post_in_subreddit_messages
from evals.critics import AnyOfCritic, ListCritic
@ -267,3 +272,190 @@ def reddit_get_top_level_comments_eval_suite() -> EvalSuite:
)
return suite
@tool_eval()
def reddit_check_subreddit_access_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_check_subreddit_access",
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_check_subreddit_access_1",
user_message="does r/WaterBottleCollecting exist?",
expected_tool_calls=[
ExpectedToolCall(
func=check_subreddit_access,
args={
"subreddit": "WaterBottleCollecting",
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=1.0),
],
)
suite.add_case(
name="reddit_check_subreddit_access_2",
user_message=(
"so my friend is a part of the WaterBottleCollecting subreddit, "
"but i cant find it. Why?"
),
expected_tool_calls=[
ExpectedToolCall(
func=check_subreddit_access,
args={
"subreddit": "WaterBottleCollecting",
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=1.0),
],
)
return suite
@tool_eval()
def reddit_get_subreddit_rules_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="reddit_get_subreddit_rules",
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_subreddit_rules_1",
user_message=(
"I'm going to be posting some stuff on WaterBottleCollecting, "
"but I'm scared that I might go against their terms & conditions "
"and get my post removed."
),
expected_tool_calls=[
ExpectedToolCall(
func=get_subreddit_rules,
args={"subreddit": "WaterBottleCollecting"},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=1.0),
],
)
suite.add_case(
name="reddit_get_subreddit_rules_2",
user_message=(
"What are WaterBottleCollecting's bannable offenses? I don't want to get banned!"
),
expected_tool_calls=[
ExpectedToolCall(
func=get_subreddit_rules,
args={"subreddit": "WaterBottleCollecting"},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="subreddit", weight=1.0),
],
)
return suite
@tool_eval()
def reddit_get_my_posts_eval_suite() -> EvalSuite:
get_my_posts_response = [
{"role": "user", "content": "get 1 of my posts"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_hPacHNSvuKKamoKKoPWBQosv",
"type": "function",
"function": {"name": "Reddit_GetMyPosts", "arguments": '{"limit":1}'},
}
],
},
{
"role": "tool",
"content": '{"cursor":"t3_1jt9jz4","posts":[{"author":"RedditUser123","body":"i just wanted to say that i love thisapp\\n\\n","created_utc":1743988489,"id":"1jt9jz4","is_video":false,"name":"t3_1jt9jz4","num_comments":1,"permalink":"/r/SparkingWater/comments/1jt9jz4/this_is_fun/","score":1,"subreddit":"SparklingWater","title":"this isfun","upvote_ratio":1,"upvotes":1,"url":"https://www.reddit.com/r/SparklingWater/comments/1jt9jz4/this_is_fun/"}]}', # noqa: E501
"tool_call_id": "call_hPacHNSvuKKamoKKoPWBQosv",
"name": "Reddit_GetMyPosts",
},
{
"role": "assistant",
"content": "Here is one of your posts on Reddit:\n\n**Title:** [this isfun](https://www.reddit.com/r/SparklingWater/comments/1jt9jz4/this_is_fun/)\n\n**Subreddit:**r/SparklingWater\n\n**Content:** \n```\ni just wanted to say that i love this app\n```\n\n**Upvotes:** 1 \n**Comments:** 1", # noqa: E501
},
]
suite = EvalSuite(
name="reddit_get_my_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_my_posts_1",
user_message=(
"I want to train an AI on the voice I use for my reddit posts. "
"Help me out here & get my last 100"
),
expected_tool_calls=[
ExpectedToolCall(
func=get_my_posts,
args={
"limit": 100,
"include_body": True,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="limit", weight=0.5),
BinaryCritic(critic_field="include_body", weight=0.5),
],
)
suite.add_case(
name="reddit_get_my_posts_2",
user_message=("get 25 more but w/o their content"),
expected_tool_calls=[
ExpectedToolCall(
func=get_my_posts,
args={
"limit": 25,
"include_body": False,
"cursor": "t3_1jt9jz4",
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="limit", weight=0.3),
BinaryCritic(critic_field="include_body", weight=0.3),
BinaryCritic(critic_field="cursor", weight=0.4),
],
additional_messages=get_my_posts_response,
)
return suite

1
toolkits/reddit/pyproject.toml
View file

@ -7,6 +7,7 @@ authors = ["Arcade <dev@arcade.dev>"]
[tool.poetry.dependencies]
python = "^3.10"
arcade-ai = "^1.0.5"
httpx = "^0.27.2"
[tool.poetry.dev-dependencies]
pytest = "^8.3.0"

132
toolkits/reddit/tests/test_utils.py
View file

@ -1,6 +1,8 @@
import httpx
import pytest
from arcade.sdk.errors import ToolExecutionError
from arcade_reddit.client import RedditClient
from arcade_reddit.utils import (
create_fullname_for_comment,
create_fullname_for_multiple_posts,
@ -10,9 +12,17 @@ from arcade_reddit.utils import (
parse_get_content_of_post_response,
parse_get_posts_in_subreddit_response,
parse_get_top_level_comments_response,
parse_subreddit_rules_response,
parse_user_posts_response,
resolve_subreddit_access,
)
class DummyRedditClientResponse:
def __init__(self, status_code):
self.status_code = status_code
@pytest.mark.parametrize(
"identifier, expected",
[
@ -189,3 +199,125 @@ def test_parse_get_top_level_comments_response_missing_data():
expected = {"comments": [], "num_comments": 0}
result = parse_get_top_level_comments_response(data)
assert result == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"status_code, expected_result, expected_exception",
[
(None, {"exists": True, "accessible": True}, None),
(404, {"exists": False, "accessible": False}, None),
(302, {"exists": False, "accessible": False}, None),
(403, {"exists": True, "accessible": False}, None),
(500, None, httpx.HTTPStatusError),
],
)
async def test_resolve_subreddit_access(status_code, expected_result, expected_exception):
class DummyRedditClient(RedditClient):
async def get(self, path: str, **kwargs):
if status_code is not None:
raise httpx.HTTPStatusError(
"Error", request=None, response=DummyRedditClientResponse(status_code)
)
return "dummy_success"
client = DummyRedditClient(token="dummy") # noqa: S106
if expected_exception:
with pytest.raises(expected_exception):
await resolve_subreddit_access(client, "testsub")
else:
result = await resolve_subreddit_access(client, "testsub")
assert result == expected_result
@pytest.mark.parametrize(
"data, expected",
[
(
{
"rules": [
{"priority": 1, "short_name": "Rule1", "description": "Desc1"},
{"priority": 2, "short_name": "Rule2", "description": "Desc2"},
]
},
{
"rules": [
{"priority": 1, "title": "Rule1", "body": "Desc1"},
{"priority": 2, "title": "Rule2", "body": "Desc2"},
]
},
),
({}, {"rules": []}),
],
)
def test_parse_subreddit_rules_response(data, expected):
result = parse_subreddit_rules_response(data)
assert result == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"posts_data, expected_cursor, expected_posts",
[
(
{
"data": {
"after": "cursor123",
"children": [
{
"data": {
"id": "1",
"name": "t3_1",
"title": "Post 1",
"author": "user1",
"subreddit": "subreddit1",
"created_utc": 1712345678,
"num_comments": 10,
"score": 100,
"upvote_ratio": 0.5,
"ups": 50,
"permalink": "permalink1",
"url": "url1",
"is_video": False,
"selftext": "This is the body of post 1",
"all_awardings": [],
"allow_live_comments": False,
"approved": False,
"approved_at_utc": None,
}
},
],
}
},
"cursor123",
[
{
"id": "1",
"name": "t3_1",
"title": "Post 1",
"author": "user1",
"subreddit": "subreddit1",
"created_utc": 1712345678,
"num_comments": 10,
"score": 100,
"upvote_ratio": 0.5,
"upvotes": 50,
"permalink": "permalink1",
"url": "url1",
"is_video": False,
},
],
),
({"data": {"after": None, "children": []}}, None, []),
],
)
async def test_parse_user_posts_response_without_body(posts_data, expected_cursor, expected_posts):
dummy_context = object()
result = await parse_user_posts_response(dummy_context, posts_data, include_body=False)
if expected_cursor:
expected = {"cursor": expected_cursor, "posts": expected_posts}
else:
expected = {"posts": expected_posts}
assert result == expected