"""
User-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 USER_POSTS_QUERY, USER_QUERY, USER_VOTED_POSTS_QUERY, VIEWER_QUERY
from product_hunt_mcp.schemas.validation import USER_SCHEMA
from product_hunt_mcp.utils.common import (
apply_pagination_defaults,
check_data_exists,
execute_and_check_query,
format_response,
handle_errors,
require_token,
)
from product_hunt_mcp.utils.validation import validate_with_schema
logger = logging.getLogger("ph_mcp")
def register_user_tools(mcp):
"""Register user-related tools with the MCP server."""
@mcp.tool()
@require_token
@handle_errors
@validate_with_schema(USER_SCHEMA)
def get_user(
id: str = None,
username: str = None,
posts_type: str = None,
posts_count: int = None,
posts_after: str = None,
) -> Dict[str, Any]:
"""
Retrieve user information by ID or username, with optional retrieval of their posts.
Parameters:
- id (str, optional): The user's unique ID.
- username (str, optional): The user's username.
- posts_type (str, optional): Type of posts to retrieve. Valid values: MADE (default), VOTED.
- posts_count (int, optional): Number of posts to return (default: 10, max: 20).
- posts_after (str, optional): Pagination cursor for next page of posts.
At least one of `id` or `username` must be provided.
Returns:
- success (bool)
- data (dict): If successful, contains user details and optionally their posts.
- error (dict, optional)
- rate_limits (dict)
Notes:
- Returns an error if neither `id` nor `username` is provided, or if the user is not found.
"""
params = {
k: v
for k, v in {
"id": id,
"username": username,
"posts_type": posts_type,
"posts_count": posts_count,
"posts_after": posts_after,
}.items()
if v is not None
}
logger.info("users.get_user called", extra=params)
# Determine if posts are being requested
requesting_posts = posts_type is not None or posts_count is not None
# Apply sensible defaults if posts are being requested
if requesting_posts:
posts_type = posts_type or "MADE"
posts_count = posts_count or 10
else:
posts_type = None
posts_count = None
# Set up common variables
variables = {}
if id:
variables["id"] = id
if username:
variables["username"] = username
# Normalize posts_type
posts_type = posts_type.upper() if posts_type else None
# Case 1: Basic user info (no posts requested)
if not requesting_posts:
result, rate_limits, error = execute_graphql_query(USER_QUERY, variables)
if error:
return format_response(False, error=error, rate_limits=rate_limits)
if not check_data_exists(result["data"], "user"):
id_or_username = id or username
return format_response(
False,
error={
"code": "NOT_FOUND",
"message": f"User with ID/username '{id_or_username}' not found",
},
rate_limits=rate_limits,
)
return format_response(True, data=result["data"]["user"], rate_limits=rate_limits)
# Case 2 & 3: Posts requested (made or voted)
# Set up pagination
pagination = apply_pagination_defaults(posts_count, posts_after)
variables.update(pagination)
# Choose query based on posts_type
if posts_type == "MADE":
query = USER_POSTS_QUERY
elif posts_type == "VOTED":
query = USER_VOTED_POSTS_QUERY
else:
return format_response(
False,
error={
"code": "INVALID_PARAMETER",
"message": f"Invalid posts_type: {posts_type}. Valid values are MADE, VOTED.",
},
)
# Execute the appropriate query
result, rate_limits, error = execute_graphql_query(query, variables)
if error:
return format_response(False, error=error, rate_limits=rate_limits)
if not check_data_exists(result["data"], "user"):
id_or_username = id or username
return format_response(
False,
error={
"code": "NOT_FOUND",
"message": f"User with ID/username '{id_or_username}' not found",
},
rate_limits=rate_limits,
)
user_data = result["data"]["user"]
# Format response based on the query type
if posts_type == "MADE" and check_data_exists(user_data, "madePosts"):
posts_data = user_data["madePosts"]
response_data = {
"id": user_data["id"],
"posts": posts_data
}
return format_response(True, data=response_data, rate_limits=rate_limits)
elif posts_type == "VOTED" and check_data_exists(user_data, "votedPosts"):
posts_data = user_data["votedPosts"]
response_data = {
"id": user_data["id"],
"posts": posts_data
}
return format_response(True, data=response_data, rate_limits=rate_limits)
# If we get here, the posts field wasn't in the response, just return what we have
return format_response(True, data=user_data, rate_limits=rate_limits)
@mcp.tool()
@require_token
@handle_errors
def get_viewer() -> Dict[str, Any]:
"""
Retrieve information about the currently authenticated user.
Parameters:
- None
Returns:
- success (bool)
- data (dict): If successful, contains user details.
- error (dict, optional)
- rate_limits (dict)
Notes:
- Returns an error if the token is invalid or expired.
"""
logger.info("users.get_viewer called")
result, rate_limits, error = execute_graphql_query(VIEWER_QUERY)
if error:
return format_response(False, error=error, rate_limits=rate_limits)
# Check if viewer info exists
viewer_exists = check_data_exists(result["data"], "viewer")
if not viewer_exists:
return format_response(
False,
error={
"code": "AUTHENTICATION_ERROR",
"message": "Unable to get viewer information. Token may be invalid or expired.",
},
rate_limits=rate_limits,
)
# Extract viewer data
viewer_data = result["data"]["viewer"]
# Check if the user field exists for nested viewer structure
if "user" in viewer_data and viewer_data["user"] is not None:
viewer_data = viewer_data["user"]
return format_response(True, data=viewer_data, rate_limits=rate_limits)
