Twitter toolset update

[Wizard1209]

No description provided.

[ViZiD]

@greptile

[greptile-apps]

Confidence Score: 4/5

• Safe to merge with minor fixes — one P1 bug around missing exception handling in _resolve_user_id that could cause unhandled exceptions under network failure conditions.
• The overall quality of the tools is high with consistent patterns, input validation, and pagination support. The single P1 issue (missing exception wrapping in _resolve_user_id) is low-severity in practice because the Letta runtime likely catches unhandled exceptions from tool execution, but it breaks the clean error-string contract used everywhere else. The remaining issues are style/documentation level.
• letta_bot/custom_tools/get_account_timeline.py — the _resolve_user_id helper needs exception handling added.

Important Files Changed

Filename	Overview
letta_bot/custom_tools/discover_accounts.py	New tool with 6 discovery methods (following, list_peers, quote_tweets, retweeted_by, mentions, followers); well-structured with a shared _api_get helper and consistent error handling throughout.
letta_bot/custom_tools/discover_topics.py	New topic-search tool; hardcodes lang:en with no parameter to override, limiting usefulness for non-English discovery. Otherwise well-implemented with proper error handling.
letta_bot/custom_tools/get_account_timeline.py	Timeline retrieval tool; _resolve_user_id helper is missing try/except for requests.exceptions.Timeout/RequestException, unlike every other API call in this PR, which would cause unhandled exceptions on network failures.
letta_bot/custom_tools/get_users_info.py	Batch user-info lookup tool; clean implementation with proper error handling, validation, and pinned tweet support. No issues found.
letta_bot/custom_tools/search_x_posts.py	Post search tool with optional media delivery via Letta injected client; properly uses except Exception to handle potential runtime failures in async media delivery. No critical issues.
letta_bot/custom_tools/x_api_request.py	Generic X API v2 wrapper; params type hint is str but the implementation silently accepts dict as well, creating an undocumented interface inconsistency.

Sequence Diagram

sequenceDiagram
    participant Agent as Letta Agent
    participant Tool as Custom Tool
    participant XAPI as X API v2

    Note over Agent,XAPI: discover_accounts / get_account_timeline / search_x_posts

    Agent->>Tool: call tool(params)
    Tool->>Tool: validate inputs & read X_API_KEY env var

    alt User ID needed (discover_accounts, get_account_timeline)
        Tool->>XAPI: GET /2/users/by/username/{user}
        XAPI-->>Tool: {data: {id, public_metrics, ...}}
    end

    Tool->>XAPI: GET /2/{endpoint} (tweets, followers, lists, etc.)
    XAPI-->>Tool: {data: [...], includes: {users, media}, meta: {next_token}}

    Tool->>Tool: filter / sort / format results

    alt attach_media=True (search_x_posts, get_account_timeline)
        Tool->>Tool: build image_parts from media URLs
        Tool->>Tool: client.agents.messages.create_async(agent_id, image_parts)
        Note right of Tool: `client` injected by Letta runtime
    end

    Tool-->>Agent: formatted string result

Loading

_{Last reviewed commit: c31b63c}

[greptile-apps]

Missing exception handling for network errors in _resolve_user_id

Unlike every other API call in this PR (_api_get in discover_accounts.py, get_users_info, search_x_posts, discover_topics), _resolve_user_id has no try/except around its requests.get call. If the request times out or a connection error occurs, a requests.exceptions.Timeout or requests.exceptions.RequestException will propagate up through get_account_timeline (which also has no catch for it) and crash the tool execution entirely — rather than returning a clean error string.

Suggested change

	def _resolve_user_id(username, bearer_token):
	"""Resolve X username to user ID."""
	import requests
	response = requests.get(
	f'{USERS_API_URL}/{username}',
	headers={'Authorization': f'Bearer {bearer_token}'},
	params={'user.fields': 'public_metrics,verified,verified_type'},
	timeout=REQUEST_TIMEOUT,
	)
	if response.status_code != 200:
	return None, f'Error: X API returned status {response.status_code} when looking up @{username}'
	data = response.json()
	# X API returns 200 with errors array when user not found
	if data.get('errors') and not data.get('data'):
	return None, f'Error: User @{username} not found'
	user_data = data.get('data')
	if not user_data:
	return None, f'Error: User @{username} not found'
	return user_data, None
	def _resolve_user_id(username, bearer_token):
	"""Resolve X username to user ID."""
	import requests
	try:
	response = requests.get(
	f'{USERS_API_URL}/{username}',
	headers={'Authorization': f'Bearer {bearer_token}'},
	params={'user.fields': 'public_metrics,verified,verified_type'},
	timeout=REQUEST_TIMEOUT,
	)
	except requests.exceptions.Timeout:
	return None, f'Error: Request to X API timed out while looking up @{username}'
	except requests.exceptions.RequestException as e:
	return None, f'Error: {e}'
	if response.status_code != 200:
	return None, f'Error: X API returned status {response.status_code} when looking up @{username}'
	data = response.json()
	# X API returns 200 with errors array when user not found
	if data.get('errors') and not data.get('data'):
	return None, f'Error: User @{username} not found'
	user_data = data.get('data')
	if not user_data:
	return None, f'Error: User @{username} not found'
	return user_data, None

[greptile-apps]

Hardcoded lang:en filter with no override option

The language filter is always appended with no parameter to control it. While the comment acknowledges this as "English language default," there's no way for the caller to discover topics in other languages. This limits the tool's usefulness for multilingual use cases.

Consider adding a language parameter (defaulting to 'en') and skipping the filter when it's empty:

Suggested change

	# English language default (most useful for discovery)
	query_parts.append('lang:en')
	# English language default (most useful for discovery)
	# lang: 'en' by default; pass empty string to search all languages

And in the function signature, add:

def discover_topics(
    keywords: str,
    exclude_accounts: str = '',
    hours_ago: int = 24,
    min_likes: int = 5,
    max_results: int = 20,
    next_token: str = '',
    language: str = 'en',
) -> str:

Then conditionally append:

if language:
    query_parts.append(f'lang:{language}')

_{Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!}

[greptile-apps]

Type hint for params is str but the code silently accepts dict

The function signature declares params: str = '', but lines 71-72 explicitly handle the case where params is already a dict. This creates a mismatch between the documented interface (JSON string) and the actual runtime behaviour. An LLM agent reading the type hint or docstring would not know to pass a dict, yet the code would accept one.

Either update the type hint to reflect what is truly accepted, or remove the dict path entirely if it's only meant to handle internal callers:

Suggested change

	def x_api_request(
	endpoint: str,
	params: str = '',
	method: str = 'GET',
	) -> str:
	def x_api_request(
	endpoint: str,
	params: str = '',
	method: str = 'GET',
	) -> str:

If dict inputs should be supported, update the hint to str | dict (and document it in the docstring).