"""
Post-related tools for the Product Hunt MCP server.
"""
import logging
from typing import Any, Dict
from product_hunt_mcp.api.client import execute_graphql_query
from product_hunt_mcp.api.queries import POST_QUERY, POSTS_QUERY
from product_hunt_mcp.schemas.validation import POST_SCHEMA, POSTS_SCHEMA
from product_hunt_mcp.utils.common import (
add_id_or_slug,
apply_pagination_defaults,
execute_and_check_query,
extract_pagination,
format_response,
handle_errors,
require_token,
)
from product_hunt_mcp.utils.validation import validate_with_schema
logger = logging.getLogger("ph_mcp")
def register_post_tools(mcp):
"""Register post-related tools with the MCP server."""
@mcp.tool()
@require_token
@handle_errors
@validate_with_schema(POST_SCHEMA)
def get_post_details(
id: str = None, slug: str = None, comments_count: int = 10, comments_after: str = None
) -> Dict[str, Any]:
"""
Retrieve detailed information about a specific Product Hunt post by ID or slug.
Parameters:
- id (str, optional): The post's unique ID.
- slug (str, optional): The post's slug (e.g., "product-hunt-api").
- comments_count (int, optional): Number of comments to return (default: 10, max: 20).
- comments_after (str, optional): Pagination cursor for fetching the next page of comments.
At least one of `id` or `slug` must be provided.
Returns:
- success (bool): Whether the request was successful.
- data (dict): If successful, contains:
- id, name, description, tagline, votes, makers, topics, media, and
- comments (paginated): { edges: [...], pageInfo: { endCursor, hasNextPage } }
- error (dict, optional): If unsuccessful, contains error code and message.
- rate_limits (dict): API rate limit information.
Notes:
- If neither `id` nor `slug` is provided, an error is returned.
- If the post is not found, an error is returned.
- The dedicated `get_post_comments` tool is deprecated; use this tool for paginated comments.
"""
params = {
k: v
for k, v in {
"id": id,
"slug": slug,
"comments_count": comments_count,
"comments_after": comments_after,
}.items()
if v is not None
}
logger.info("posts.get_post_details called", extra=params)
variables = {}
add_id_or_slug(variables, id, slug)
# Add pagination for comments if requested
if comments_count is not None:
variables["commentsCount"] = min(comments_count, 20)
if comments_after:
variables["commentsAfter"] = comments_after
# Use the utility function to execute the query and check if post exists
id_or_slug = id or slug
post_data, rate_limits, error = execute_and_check_query(
POST_QUERY, variables, "post", id_or_slug
)
if error:
return format_response(False, error=error, rate_limits=rate_limits)
return format_response(True, data=post_data, rate_limits=rate_limits)
@mcp.tool()
@require_token
@handle_errors
@validate_with_schema(POSTS_SCHEMA)
def get_posts(
featured: bool = None,
topic: str = None,
order: str = "RANKING",
count: int = 10,
after: str = None,
url: str = None,
twitter_url: str = None,
posted_before: str = None,
posted_after: str = None,
) -> Dict[str, Any]:
"""
Retrieve a list of Product Hunt posts with various filtering and sorting options.
Parameters:
- featured (bool, optional): Only return featured posts if True.
- topic (str, optional): Filter by topic slug.
- order (str, optional): Sorting order. Valid values: RANKING (default), NEWEST, VOTES, FEATURED_AT.
- count (int, optional): Number of posts to return (default: 10, max: 20).
- after (str, optional): Pagination cursor for next page.
- url (str, optional): Filter posts by URL.
- twitter_url (str, optional): Filter posts by Twitter URL.
- posted_before (str, optional): ISO datetime to filter posts posted before this date.
- posted_after (str, optional): ISO datetime to filter posts posted after this date.
Returns:
- success (bool)
- data (dict): If successful, contains:
- posts (list): List of post objects (id, name, description, etc.)
- pagination (dict): { end_cursor, has_next_page }
- error (dict, optional)
- rate_limits (dict)
Notes:
- This is not a keyword search; use filters to narrow results.
- If no posts match, `posts` will be an empty list.
- Invalid date formats return a user-friendly error.
"""
params = {
k: v
for k, v in {
"featured": featured,
"topic": topic,
"order": order,
"count": count,
"after": after,
"url": url,
"twitter_url": twitter_url,
"posted_before": posted_before,
"posted_after": posted_after,
}.items()
if v is not None
}
logger.info("posts.get_posts called", extra=params)
# Apply pagination defaults
variables = apply_pagination_defaults(count, after)
# Add order parameter
variables["order"] = order
# Add optional filters
if featured is not None:
variables["featured"] = featured
if topic:
variables["topic"] = topic
if url:
variables["url"] = url
if twitter_url:
variables["twitterUrl"] = twitter_url
if posted_before:
variables["postedBefore"] = posted_before
if posted_after:
variables["postedAfter"] = posted_after
result, rate_limits, error = execute_graphql_query(POSTS_QUERY, variables)
if error:
# If there's a GraphQL error related to date format, provide a more user-friendly message
if (
"code" in error
and error["code"] == "GRAPHQL_ERROR"
and any(
"postedBefore" in str(e) or "postedAfter" in str(e)
for e in error.get("details", [])
)
):
return format_response(
False,
error={
"code": "INVALID_DATE_FORMAT",
"message": "The provided date format is invalid. Please use ISO 8601 format (e.g., 2023-01-01T00:00:00Z)",
},
rate_limits=rate_limits,
)
return format_response(False, error=error, rate_limits=rate_limits)
# Extract posts
posts_data = result["data"]["posts"]
return format_response(
True,
data={
"posts": posts_data["edges"],
"pagination": extract_pagination(posts_data["pageInfo"]),
},
rate_limits=rate_limits,
)
