Merge pull request #41 from kortix-ai/amazon-apikeys

Add Data Providers: Amazon, Zillow, Twitter, Yahoo, ActiveJobs
This commit is contained in:
Adam Cohen Hillel 2025-04-16 19:37:22 +01:00 committed by GitHub
commit 214089b971
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 933 additions and 50 deletions

View File

@ -139,7 +139,7 @@ You'll need the following components:
6. **RapidAPI API Key** (Optional):
- To enable API services like LinkedIn, and others, you'll need a RapidAPI key
- Each service requires individual activation in your RapidAPI account:
1. Locate the service's `base_url` in its corresponding file (e.g., `"https://linkedin-data-scraper.p.rapidapi.com"` in [`backend/agent/tools/api_services/LinkedInService.py`](backend/agent/tools/api_services/LinkedInService.py))
1. Locate the service's `base_url` in its corresponding file (e.g., `"https://linkedin-data-scraper.p.rapidapi.com"` in [`backend/agent/tools/data_providers/LinkedinProvider.py`](backend/agent/tools/data_providers/LinkedinProvider.py))
2. Visit that specific API on the RapidAPI marketplace
3. Subscribe to the service (many offer free tiers with limited requests)
4. Once subscribed, the service will be available to your agent through the API Services tool

View File

@ -68,6 +68,20 @@ You have the ability to execute operations using both Python and CLI tools:
* YOU CAN DO ANYTHING ON THE BROWSER - including clicking on elements, filling forms, submitting data, etc.
* The browser is in a sandboxed environment, so nothing to worry about.
### 2.2.6 DATA PROVIDERS
- You have access to a variety of data providers that you can use to get data for your tasks.
- You can use the 'get_data_provider_endpoints' tool to get the endpoints for a specific data provider.
- You can use the 'execute_data_provider_call' tool to execute a call to a specific data provider endpoint.
- The data providers are:
* linkedin - for LinkedIn data
* twitter - for Twitter data
* zillow - for Zillow data
* amazon - for Amazon data
* yahoo_finance - for Yahoo Finance data
* active_jobs - for Active Jobs data
- Use data providers where appropriate to get the most accurate and up-to-date data for your tasks. This is preferred over generic web scraping.
- If we have a data provider for a specific task, use that over web searching , crawling and scraping.
# 3. TOOLKIT & METHODOLOGY
## 3.1 TOOL SELECTION PRINCIPLES

View File

@ -1,8 +1,7 @@
import os
import json
from uuid import uuid4
from typing import Optional, List, Any
from datetime import datetime, timezone
from typing import Optional
from agent.tools.message_tool import MessageTool
from agent.tools.sb_deploy_tool import SandboxDeployTool
@ -14,7 +13,7 @@ from agentpress.response_processor import ProcessorConfig
from agent.tools.sb_shell_tool import SandboxShellTool
from agent.tools.sb_files_tool import SandboxFilesTool
from agent.tools.sb_browser_tool import SandboxBrowserTool
from agent.tools.api_services_tool import APIServicesTool
from agent.tools.data_providers_tool import DataProvidersTool
from agent.prompt import get_system_prompt
from sandbox.sandbox import create_sandbox, get_or_start_sandbox
from utils.billing import check_billing_status, get_account_id_from_thread
@ -74,7 +73,7 @@ async def run_agent(thread_id: str, project_id: str, stream: bool = True, thread
thread_manager.add_tool(WebSearchTool)
if os.getenv("RAPID_API_KEY"):
thread_manager.add_tool(APIServicesTool)
thread_manager.add_tool(DataProvidersTool)
xml_examples = ""
for tag_name, example in thread_manager.tool_registry.get_xml_examples().items():

View File

@ -0,0 +1,57 @@
from typing import Dict
from agent.tools.data_providers.RapidDataProviderBase import RapidDataProviderBase, EndpointSchema
class ActiveJobsProvider(RapidDataProviderBase):
def __init__(self):
endpoints: Dict[str, EndpointSchema] = {
"active_jobs": {
"route": "/active-ats-7d",
"method": "GET",
"name": "Active Jobs Search",
"description": "Get active job listings with various filter options.",
"payload": {
"limit": "Optional. Number of jobs per API call (10-100). Default is 100.",
"offset": "Optional. Offset for pagination. Default is 0.",
"title_filter": "Optional. Search terms for job title.",
"advanced_title_filter": "Optional. Advanced title filter with operators (can't be used with title_filter).",
"location_filter": "Optional. Filter by location(s). Use full names like 'United States' not 'US'.",
"description_filter": "Optional. Filter on job description content.",
"organization_filter": "Optional. Filter by company name(s).",
"description_type": "Optional. Return format for description: 'text' or 'html'. Leave empty to exclude descriptions.",
"source": "Optional. Filter by ATS source.",
"date_filter": "Optional. Filter by posting date (greater than).",
"ai_employment_type_filter": "Optional. Filter by employment type (FULL_TIME, PART_TIME, etc).",
"ai_work_arrangement_filter": "Optional. Filter by work arrangement (On-site, Hybrid, Remote OK, Remote Solely).",
"ai_experience_level_filter": "Optional. Filter by experience level (0-2, 2-5, 5-10, 10+).",
"li_organization_slug_filter": "Optional. Filter by LinkedIn company slug.",
"li_organization_slug_exclusion_filter": "Optional. Exclude LinkedIn company slugs.",
"li_industry_filter": "Optional. Filter by LinkedIn industry.",
"li_organization_specialties_filter": "Optional. Filter by LinkedIn company specialties.",
"li_organization_description_filter": "Optional. Filter by LinkedIn company description."
}
}
}
base_url = "https://active-jobs-db.p.rapidapi.com"
super().__init__(base_url, endpoints)
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
tool = ActiveJobsProvider()
# Example for searching active jobs
jobs = tool.call_endpoint(
route="active_jobs",
payload={
"limit": "10",
"offset": "0",
"title_filter": "\"Data Engineer\"",
"location_filter": "\"United States\" OR \"United Kingdom\"",
"description_type": "text"
}
)
print("Active Jobs:", jobs)

View File

@ -0,0 +1,191 @@
from typing import Dict, Optional
from agent.tools.data_providers.RapidDataProviderBase import RapidDataProviderBase, EndpointSchema
class AmazonProvider(RapidDataProviderBase):
def __init__(self):
endpoints: Dict[str, EndpointSchema] = {
"search": {
"route": "/search",
"method": "GET",
"name": "Amazon Product Search",
"description": "Search for products on Amazon with various filters and parameters.",
"payload": {
"query": "Search query (supports both free-form text queries or a product asin)",
"page": "Results page to return (default: 1)",
"country": "Sets the Amazon domain, marketplace country, language and currency (default: US)",
"sort_by": "Return the results in a specific sort order (RELEVANCE, LOWEST_PRICE, HIGHEST_PRICE, REVIEWS, NEWEST, BEST_SELLERS)",
"product_condition": "Return products in a specific condition (ALL, NEW, USED, RENEWED, COLLECTIBLE)",
"is_prime": "Only return prime products (boolean)",
"deals_and_discounts": "Return deals and discounts in a specific condition (NONE, ALL_DISCOUNTS, TODAYS_DEALS)",
"category_id": "Find products in a specific category / department (optional)",
"category": "Filter by specific numeric Amazon category (optional)",
"min_price": "Only return product offers with price greater than a certain value (optional)",
"max_price": "Only return product offers with price lower than a certain value (optional)",
"brand": "Find products with a specific brand (optional)",
"seller_id": "Find products sold by specific seller (optional)",
"four_stars_and_up": "Return product listings with ratings of 4 stars & up (optional)",
"additional_filters": "Any filters available on the Amazon page but not part of this endpoint's parameters (optional)"
}
},
"product-details": {
"route": "/product-details",
"method": "GET",
"name": "Amazon Product Details",
"description": "Get detailed information about specific Amazon products by ASIN.",
"payload": {
"asin": "Product ASIN for which to get details. Supports batching of up to 10 ASINs in a single request, separated by comma.",
"country": "Sets the Amazon domain, marketplace country, language and currency (default: US)",
"more_info_query": "A query to search and get more info about the product as part of Product Information, Customer Q&As, and Customer Reviews (optional)",
"fields": "A comma separated list of product fields to include in the response (field projection). By default all fields are returned. (optional)"
}
},
"products-by-category": {
"route": "/products-by-category",
"method": "GET",
"name": "Amazon Products by Category",
"description": "Get products from a specific Amazon category.",
"payload": {
"category_id": "The Amazon category for which to return results. Multiple category values can be separated by comma.",
"page": "Page to return (default: 1)",
"country": "Sets the Amazon domain, marketplace country, language and currency (default: US)",
"sort_by": "Return the results in a specific sort order (RELEVANCE, LOWEST_PRICE, HIGHEST_PRICE, REVIEWS, NEWEST, BEST_SELLERS)",
"min_price": "Only return product offers with price greater than a certain value (optional)",
"max_price": "Only return product offers with price lower than a certain value (optional)",
"product_condition": "Return products in a specific condition (ALL, NEW, USED, RENEWED, COLLECTIBLE)",
"brand": "Only return products of a specific brand. Multiple brands can be specified as a comma separated list (optional)",
"is_prime": "Only return prime products (boolean)",
"deals_and_discounts": "Return deals and discounts in a specific condition (NONE, ALL_DISCOUNTS, TODAYS_DEALS)",
"four_stars_and_up": "Return product listings with ratings of 4 stars & up (optional)",
"additional_filters": "Any filters available on the Amazon page but not part of this endpoint's parameters (optional)"
}
},
"product-reviews": {
"route": "/product-reviews",
"method": "GET",
"name": "Amazon Product Reviews",
"description": "Get customer reviews for a specific Amazon product by ASIN.",
"payload": {
"asin": "Product asin for which to get reviews.",
"country": "Sets the Amazon domain, marketplace country, language and currency (default: US)",
"page": "Results page to return (default: 1)",
"sort_by": "Return reviews in a specific sort order (TOP_REVIEWS, MOST_RECENT)",
"star_rating": "Only return reviews with a specific star rating (ALL, 5_STARS, 4_STARS, 3_STARS, 2_STARS, 1_STARS, POSITIVE, CRITICAL)",
"verified_purchases_only": "Only return reviews by reviewers who made a verified purchase (boolean)",
"images_or_videos_only": "Only return reviews containing images and / or videos (boolean)",
"current_format_only": "Only return reviews of the current format (product variant - e.g. Color) (boolean)"
}
},
"seller-profile": {
"route": "/seller-profile",
"method": "GET",
"name": "Amazon Seller Profile",
"description": "Get detailed information about a specific Amazon seller by Seller ID.",
"payload": {
"seller_id": "The Amazon Seller ID for which to get seller profile details",
"country": "Sets the Amazon domain, marketplace country, language and currency (default: US)",
"fields": "A comma separated list of seller profile fields to include in the response (field projection). By default all fields are returned. (optional)"
}
},
"seller-reviews": {
"route": "/seller-reviews",
"method": "GET",
"name": "Amazon Seller Reviews",
"description": "Get customer reviews for a specific Amazon seller by Seller ID.",
"payload": {
"seller_id": "The Amazon Seller ID for which to get seller reviews",
"country": "Sets the Amazon domain, marketplace country, language and currency (default: US)",
"star_rating": "Only return reviews with a specific star rating or positive / negative sentiment (ALL, 5_STARS, 4_STARS, 3_STARS, 2_STARS, 1_STARS, POSITIVE, CRITICAL)",
"page": "The page of seller feedback results to retrieve (default: 1)",
"fields": "A comma separated list of seller review fields to include in the response (field projection). By default all fields are returned. (optional)"
}
}
}
base_url = "https://real-time-amazon-data.p.rapidapi.com"
super().__init__(base_url, endpoints)
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
tool = AmazonProvider()
# Example for product search
search_result = tool.call_endpoint(
route="search",
payload={
"query": "Phone",
"page": 1,
"country": "US",
"sort_by": "RELEVANCE",
"product_condition": "ALL",
"is_prime": False,
"deals_and_discounts": "NONE"
}
)
print("Search Result:", search_result)
# Example for product details
details_result = tool.call_endpoint(
route="product-details",
payload={
"asin": "B07ZPKBL9V",
"country": "US"
}
)
print("Product Details:", details_result)
# Example for products by category
category_result = tool.call_endpoint(
route="products-by-category",
payload={
"category_id": "2478868012",
"page": 1,
"country": "US",
"sort_by": "RELEVANCE",
"product_condition": "ALL",
"is_prime": False,
"deals_and_discounts": "NONE"
}
)
print("Category Products:", category_result)
# Example for product reviews
reviews_result = tool.call_endpoint(
route="product-reviews",
payload={
"asin": "B07ZPKN6YR",
"country": "US",
"page": 1,
"sort_by": "TOP_REVIEWS",
"star_rating": "ALL",
"verified_purchases_only": False,
"images_or_videos_only": False,
"current_format_only": False
}
)
print("Product Reviews:", reviews_result)
# Example for seller profile
seller_result = tool.call_endpoint(
route="seller-profile",
payload={
"seller_id": "A02211013Q5HP3OMSZC7W",
"country": "US"
}
)
print("Seller Profile:", seller_result)
# Example for seller reviews
seller_reviews_result = tool.call_endpoint(
route="seller-reviews",
payload={
"seller_id": "A02211013Q5HP3OMSZC7W",
"country": "US",
"star_rating": "ALL",
"page": 1
}
)
print("Seller Reviews:", seller_reviews_result)

View File

@ -1,9 +1,9 @@
from typing import Dict
from agent.tools.api_services.APIServicesBase import APIServicesBase, EndpointSchema
from agent.tools.data_providers.RapidDataProviderBase import RapidDataProviderBase, EndpointSchema
class LinkedInService(APIServicesBase):
class LinkedinProvider(RapidDataProviderBase):
def __init__(self):
endpoints: Dict[str, EndpointSchema] = {
"person": {
@ -238,9 +238,9 @@ class LinkedInService(APIServicesBase):
if __name__ == "__main__":
import os
os.environ["RAPID_API_KEY"] = ""
tool = LinkedInService()
from dotenv import load_dotenv
load_dotenv()
tool = LinkedinProvider()
result = tool.call_endpoint(
route="comments_from_recent_activity",

View File

@ -11,7 +11,7 @@ class EndpointSchema(TypedDict):
payload: Dict[str, Any]
class APIServicesBase:
class RapidDataProviderBase:
def __init__(self, base_url: str, endpoints: Dict[str, EndpointSchema]):
self.base_url = base_url
self.endpoints = endpoints

View File

@ -0,0 +1,240 @@
from typing import Dict
from agent.tools.data_providers.RapidDataProviderBase import RapidDataProviderBase, EndpointSchema
class TwitterProvider(RapidDataProviderBase):
def __init__(self):
endpoints: Dict[str, EndpointSchema] = {
"user_info": {
"route": "/screenname.php",
"method": "GET",
"name": "Twitter User Info",
"description": "Get information about a Twitter user by screenname or user ID.",
"payload": {
"screenname": "Twitter username without the @ symbol",
"rest_id": "Optional Twitter user's ID. If provided, overwrites screenname parameter."
}
},
"timeline": {
"route": "/timeline.php",
"method": "GET",
"name": "User Timeline",
"description": "Get tweets from a user's timeline.",
"payload": {
"screenname": "Twitter username without the @ symbol",
"rest_id": "Optional parameter that overwrites the screenname",
"cursor": "Optional pagination cursor"
}
},
"following": {
"route": "/following.php",
"method": "GET",
"name": "User Following",
"description": "Get users that a specific user follows.",
"payload": {
"screenname": "Twitter username without the @ symbol",
"rest_id": "Optional parameter that overwrites the screenname",
"cursor": "Optional pagination cursor"
}
},
"followers": {
"route": "/followers.php",
"method": "GET",
"name": "User Followers",
"description": "Get followers of a specific user.",
"payload": {
"screenname": "Twitter username without the @ symbol",
"cursor": "Optional pagination cursor"
}
},
"search": {
"route": "/search.php",
"method": "GET",
"name": "Twitter Search",
"description": "Search for tweets with a specific query.",
"payload": {
"query": "Search query string",
"cursor": "Optional pagination cursor",
"search_type": "Optional search type (e.g. 'Top')"
}
},
"replies": {
"route": "/replies.php",
"method": "GET",
"name": "User Replies",
"description": "Get replies made by a user.",
"payload": {
"screenname": "Twitter username without the @ symbol",
"cursor": "Optional pagination cursor"
}
},
"check_retweet": {
"route": "/checkretweet.php",
"method": "GET",
"name": "Check Retweet",
"description": "Check if a user has retweeted a specific tweet.",
"payload": {
"screenname": "Twitter username without the @ symbol",
"tweet_id": "ID of the tweet to check"
}
},
"tweet": {
"route": "/tweet.php",
"method": "GET",
"name": "Get Tweet",
"description": "Get details of a specific tweet by ID.",
"payload": {
"id": "ID of the tweet"
}
},
"tweet_thread": {
"route": "/tweet_thread.php",
"method": "GET",
"name": "Get Tweet Thread",
"description": "Get a thread of tweets starting from a specific tweet ID.",
"payload": {
"id": "ID of the tweet",
"cursor": "Optional pagination cursor"
}
},
"retweets": {
"route": "/retweets.php",
"method": "GET",
"name": "Get Retweets",
"description": "Get users who retweeted a specific tweet.",
"payload": {
"id": "ID of the tweet",
"cursor": "Optional pagination cursor"
}
},
"latest_replies": {
"route": "/latest_replies.php",
"method": "GET",
"name": "Get Latest Replies",
"description": "Get the latest replies to a specific tweet.",
"payload": {
"id": "ID of the tweet",
"cursor": "Optional pagination cursor"
}
}
}
base_url = "https://twitter-api45.p.rapidapi.com"
super().__init__(base_url, endpoints)
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
tool = TwitterProvider()
# Example for getting user info
user_info = tool.call_endpoint(
route="user_info",
payload={
"screenname": "elonmusk",
# "rest_id": "44196397" # Optional, uncomment to use user ID instead of screenname
}
)
print("User Info:", user_info)
# Example for getting user timeline
timeline = tool.call_endpoint(
route="timeline",
payload={
"screenname": "elonmusk",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Timeline:", timeline)
# Example for getting user following
following = tool.call_endpoint(
route="following",
payload={
"screenname": "elonmusk",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Following:", following)
# Example for getting user followers
followers = tool.call_endpoint(
route="followers",
payload={
"screenname": "elonmusk",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Followers:", followers)
# Example for searching tweets
search_results = tool.call_endpoint(
route="search",
payload={
"query": "cybertruck",
"search_type": "Top" # Optional, defaults to Top
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Search Results:", search_results)
# Example for getting user replies
replies = tool.call_endpoint(
route="replies",
payload={
"screenname": "elonmusk",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Replies:", replies)
# Example for checking if user retweeted a tweet
check_retweet = tool.call_endpoint(
route="check_retweet",
payload={
"screenname": "elonmusk",
"tweet_id": "1671370010743263233"
}
)
print("Check Retweet:", check_retweet)
# Example for getting tweet details
tweet = tool.call_endpoint(
route="tweet",
payload={
"id": "1671370010743263233"
}
)
print("Tweet:", tweet)
# Example for getting a tweet thread
tweet_thread = tool.call_endpoint(
route="tweet_thread",
payload={
"id": "1738106896777699464",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Tweet Thread:", tweet_thread)
# Example for getting retweets of a tweet
retweets = tool.call_endpoint(
route="retweets",
payload={
"id": "1700199139470942473",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Retweets:", retweets)
# Example for getting latest replies to a tweet
latest_replies = tool.call_endpoint(
route="latest_replies",
payload={
"id": "1738106896777699464",
# "cursor": "optional-cursor-value" # Optional for pagination
}
)
print("Latest Replies:", latest_replies)

View File

@ -0,0 +1,190 @@
from typing import Dict
from agent.tools.data_providers.RapidDataProviderBase import RapidDataProviderBase, EndpointSchema
class YahooFinanceProvider(RapidDataProviderBase):
def __init__(self):
endpoints: Dict[str, EndpointSchema] = {
"get_tickers": {
"route": "/v2/markets/tickers",
"method": "GET",
"name": "Yahoo Finance Tickers",
"description": "Get financial tickers from Yahoo Finance with various filters and parameters.",
"payload": {
"page": "Page number for pagination (optional, default: 1)",
"type": "Asset class type (required): STOCKS, ETF, MUTUALFUNDS, or FUTURES",
}
},
"search": {
"route": "/v1/markets/search",
"method": "GET",
"name": "Yahoo Finance Search",
"description": "Search for financial instruments on Yahoo Finance",
"payload": {
"search": "Search term (required)",
}
},
"get_news": {
"route": "/v2/markets/news",
"method": "GET",
"name": "Yahoo Finance News",
"description": "Get news related to specific tickers from Yahoo Finance",
"payload": {
"tickers": "Stock symbol (optional, e.g., AAPL)",
"type": "News type (optional): ALL, VIDEO, or PRESS_RELEASE",
}
},
"get_stock_module": {
"route": "/v1/markets/stock/modules",
"method": "GET",
"name": "Yahoo Finance Stock Module",
"description": "Get detailed information about a specific stock module",
"payload": {
"ticker": "Company ticker symbol (required, e.g., AAPL)",
"module": "Module to retrieve (required): asset-profile, financial-data, earnings, etc.",
}
},
"get_sma": {
"route": "/v1/markets/indicators/sma",
"method": "GET",
"name": "Yahoo Finance SMA Indicator",
"description": "Get Simple Moving Average (SMA) indicator data for a stock",
"payload": {
"symbol": "Stock symbol (required, e.g., AAPL)",
"interval": "Time interval (required): 5m, 15m, 30m, 1h, 1d, 1wk, 1mo, 3mo",
"series_type": "Series type (required): open, close, high, low",
"time_period": "Number of data points used for calculation (required)",
"limit": "Limit the number of results (optional, default: 50)",
}
},
"get_rsi": {
"route": "/v1/markets/indicators/rsi",
"method": "GET",
"name": "Yahoo Finance RSI Indicator",
"description": "Get Relative Strength Index (RSI) indicator data for a stock",
"payload": {
"symbol": "Stock symbol (required, e.g., AAPL)",
"interval": "Time interval (required): 5m, 15m, 30m, 1h, 1d, 1wk, 1mo, 3mo",
"series_type": "Series type (required): open, close, high, low",
"time_period": "Number of data points used for calculation (required)",
"limit": "Limit the number of results (optional, default: 50)",
}
},
"get_earnings_calendar": {
"route": "/v1/markets/calendar/earnings",
"method": "GET",
"name": "Yahoo Finance Earnings Calendar",
"description": "Get earnings calendar data for a specific date",
"payload": {
"date": "Calendar date in yyyy-mm-dd format (optional, e.g., 2023-11-30)",
}
},
"get_insider_trades": {
"route": "/v1/markets/insider-trades",
"method": "GET",
"name": "Yahoo Finance Insider Trades",
"description": "Get recent insider trading activity",
"payload": {}
},
}
base_url = "https://yahoo-finance15.p.rapidapi.com/api"
super().__init__(base_url, endpoints)
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
tool = YahooFinanceProvider()
# Example for getting stock tickers
tickers_result = tool.call_endpoint(
route="get_tickers",
payload={
"page": 1,
"type": "STOCKS"
}
)
print("Tickers Result:", tickers_result)
# Example for searching financial instruments
search_result = tool.call_endpoint(
route="search",
payload={
"search": "AA"
}
)
print("Search Result:", search_result)
# Example for getting financial news
news_result = tool.call_endpoint(
route="get_news",
payload={
"tickers": "AAPL",
"type": "ALL"
}
)
print("News Result:", news_result)
# Example for getting stock asset profile module
stock_module_result = tool.call_endpoint(
route="get_stock_module",
payload={
"ticker": "AAPL",
"module": "asset-profile"
}
)
print("Asset Profile Result:", stock_module_result)
# Example for getting financial data module
financial_data_result = tool.call_endpoint(
route="get_stock_module",
payload={
"ticker": "AAPL",
"module": "financial-data"
}
)
print("Financial Data Result:", financial_data_result)
# Example for getting SMA indicator data
sma_result = tool.call_endpoint(
route="get_sma",
payload={
"symbol": "AAPL",
"interval": "5m",
"series_type": "close",
"time_period": "50",
"limit": "50"
}
)
print("SMA Result:", sma_result)
# Example for getting RSI indicator data
rsi_result = tool.call_endpoint(
route="get_rsi",
payload={
"symbol": "AAPL",
"interval": "5m",
"series_type": "close",
"time_period": "50",
"limit": "50"
}
)
print("RSI Result:", rsi_result)
# Example for getting earnings calendar data
earnings_calendar_result = tool.call_endpoint(
route="get_earnings_calendar",
payload={
"date": "2023-11-30"
}
)
print("Earnings Calendar Result:", earnings_calendar_result)
# Example for getting insider trades
insider_trades_result = tool.call_endpoint(
route="get_insider_trades",
payload={}
)
print("Insider Trades Result:", insider_trades_result)

View File

@ -0,0 +1,184 @@
from typing import Dict
from agent.tools.data_providers.RapidDataProviderBase import RapidDataProviderBase, EndpointSchema
class ZillowProvider(RapidDataProviderBase):
def __init__(self):
endpoints: Dict[str, EndpointSchema] = {
"search": {
"route": "/search",
"method": "GET",
"name": "Zillow Property Search",
"description": "Search for properties by neighborhood, city, or ZIP code with various filters.",
"payload": {
"location": "Location can be an address, neighborhood, city, or ZIP code (required)",
"page": "Page number for pagination (optional, default: 0)",
"output": "Output format: json, csv, xlsx (optional, default: json)",
"status": "Status of properties: forSale, forRent, recentlySold (optional, default: forSale)",
"sortSelection": "Sorting criteria (optional, default: priorityscore)",
"listing_type": "Listing type: by_agent, by_owner_other (optional, default: by_agent)",
"doz": "Days on Zillow: any, 1, 7, 14, 30, 90, 6m, 12m, 24m, 36m (optional, default: any)",
"price_min": "Minimum price (optional)",
"price_max": "Maximum price (optional)",
"sqft_min": "Minimum square footage (optional)",
"sqft_max": "Maximum square footage (optional)",
"beds_min": "Minimum number of bedrooms (optional)",
"beds_max": "Maximum number of bedrooms (optional)",
"baths_min": "Minimum number of bathrooms (optional)",
"baths_max": "Maximum number of bathrooms (optional)",
"built_min": "Minimum year built (optional)",
"built_max": "Maximum year built (optional)",
"lotSize_min": "Minimum lot size in sqft (optional)",
"lotSize_max": "Maximum lot size in sqft (optional)",
"keywords": "Keywords to search for (optional)"
}
},
"search_address": {
"route": "/search_address",
"method": "GET",
"name": "Zillow Address Search",
"description": "Search for a specific property by its full address.",
"payload": {
"address": "Full property address (required)"
}
},
"propertyV2": {
"route": "/propertyV2",
"method": "GET",
"name": "Zillow Property Details",
"description": "Get detailed information about a specific property by zpid or URL.",
"payload": {
"zpid": "Zillow property ID (optional if URL is provided)",
"url": "Property details URL (optional if zpid is provided)"
}
},
"zestimate_history": {
"route": "/zestimate_history",
"method": "GET",
"name": "Zillow Zestimate History",
"description": "Get historical Zestimate values for a specific property.",
"payload": {
"zpid": "Zillow property ID (optional if URL is provided)",
"url": "Property details URL (optional if zpid is provided)"
}
},
"similar_properties": {
"route": "/similar_properties",
"method": "GET",
"name": "Zillow Similar Properties",
"description": "Find properties similar to a specific property.",
"payload": {
"zpid": "Zillow property ID (optional if URL or address is provided)",
"url": "Property details URL (optional if zpid or address is provided)",
"address": "Property address (optional if zpid or URL is provided)"
}
},
"mortgage_rates": {
"route": "/mortgage/rates",
"method": "GET",
"name": "Zillow Mortgage Rates",
"description": "Get current mortgage rates for different loan programs and conditions.",
"payload": {
"program": "Loan program (required): Fixed30Year, Fixed20Year, Fixed15Year, Fixed10Year, ARM3, ARM5, ARM7, etc.",
"state": "State abbreviation (optional, default: US)",
"refinance": "Whether this is for refinancing (optional, default: false)",
"loanType": "Type of loan: Conventional, etc. (optional)",
"loanAmount": "Loan amount category: Micro, SmallConforming, Conforming, SuperConforming, Jumbo (optional)",
"loanToValue": "Loan to value ratio: Normal, High, VeryHigh (optional)",
"creditScore": "Credit score category: Low, High, VeryHigh (optional)",
"duration": "Duration in days (optional, default: 30)"
}
},
}
base_url = "https://zillow56.p.rapidapi.com"
super().__init__(base_url, endpoints)
if __name__ == "__main__":
from dotenv import load_dotenv
from time import sleep
load_dotenv()
tool = ZillowProvider()
# Example for searching properties in Houston
search_result = tool.call_endpoint(
route="search",
payload={
"location": "houston, tx",
"status": "forSale",
"sortSelection": "priorityscore",
"listing_type": "by_agent",
"doz": "any"
}
)
print("Search Result:", search_result)
print("***")
print("***")
print("***")
sleep(1)
# Example for searching by address
address_result = tool.call_endpoint(
route="search_address",
payload={
"address": "1161 Natchez Dr College Station Texas 77845"
}
)
print("Address Search Result:", address_result)
print("***")
print("***")
print("***")
sleep(1)
# Example for getting property details
property_result = tool.call_endpoint(
route="propertyV2",
payload={
"zpid": "7594920"
}
)
print("Property Details Result:", property_result)
sleep(1)
print("***")
print("***")
print("***")
# Example for getting zestimate history
zestimate_result = tool.call_endpoint(
route="zestimate_history",
payload={
"zpid": "20476226"
}
)
print("Zestimate History Result:", zestimate_result)
sleep(1)
print("***")
print("***")
print("***")
# Example for getting similar properties
similar_result = tool.call_endpoint(
route="similar_properties",
payload={
"zpid": "28253016"
}
)
print("Similar Properties Result:", similar_result)
sleep(1)
print("***")
print("***")
print("***")
# Example for getting mortgage rates
mortgage_result = tool.call_endpoint(
route="mortgage_rates",
payload={
"program": "Fixed30Year",
"state": "US",
"refinance": "false",
"loanType": "Conventional",
"loanAmount": "Conforming",
"loanToValue": "Normal",
"creditScore": "Low",
"duration": "30"
}
)
print("Mortgage Rates Result:", mortgage_result)

View File

@ -1,29 +1,37 @@
import json
from agentpress.tool import Tool, ToolResult, openapi_schema, xml_schema
from agent.tools.api_services.LinkedInService import LinkedInService
from agent.tools.data_providers.LinkedinProvider import LinkedinProvider
from agent.tools.data_providers.YahooFinanceProvider import YahooFinanceProvider
from agent.tools.data_providers.AmazonProvider import AmazonProvider
from agent.tools.data_providers.ZillowProvider import ZillowProvider
from agent.tools.data_providers.TwitterProvider import TwitterProvider
class APIServicesTool(Tool):
"""Tool for making requests to various API services."""
class DataProvidersTool(Tool):
"""Tool for making requests to various data providers."""
def __init__(self):
super().__init__()
self.register_apis = {
"linkedin": LinkedInService()
self.register_data_providers = {
"linkedin": LinkedinProvider(),
"yahoo_finance": YahooFinanceProvider(),
"amazon": AmazonProvider(),
"zillow": ZillowProvider(),
"twitter": TwitterProvider()
}
@openapi_schema({
"type": "function",
"function": {
"name": "get_api_service_endpoints",
"description": "Get available endpoints for a specific API service",
"name": "get_data_provider_endpoints",
"description": "Get available endpoints for a specific data provider",
"parameters": {
"type": "object",
"properties": {
"service_name": {
"type": "string",
"description": "The name of the API service (e.g., 'linkedin')"
"description": "The name of the data provider (e.g., 'linkedin', 'twitter', 'zillow', 'amazon', 'yahoo_finance')"
}
},
"required": ["service_name"]
@ -31,44 +39,44 @@ class APIServicesTool(Tool):
}
})
@xml_schema(
tag_name="get-api-service-endpoints",
tag_name="get-data-provider-endpoints",
mappings=[
{"param_name": "service_name", "node_type": "attribute", "path": "."}
],
example='''
<!--
The get-api-service-endpoints tool returns available endpoints for a specific API service.
The get-data-provider-endpoints tool returns available endpoints for a specific data provider.
Use this tool when you need to discover what endpoints are available.
-->
<!-- Example to get LinkedIn API endpoints -->
<get-api-service-endpoints service_name="linkedin">
</get-api-service-endpoints>
<get-data-provider-endpoints service_name="linkedin">
</get-data-provider-endpoints>
'''
)
async def get_api_service_endpoints(
async def get_data_provider_endpoints(
self,
service_name: str
) -> ToolResult:
"""
Get available endpoints for a specific API service.
Get available endpoints for a specific data provider.
Parameters:
- service_name: The name of the API service (e.g., 'linkedin')
- service_name: The name of the data provider (e.g., 'linkedin')
"""
try:
if not service_name:
return self.fail_response("API name is required.")
return self.fail_response("Data provider name is required.")
if service_name not in self.register_apis:
return self.fail_response(f"API '{service_name}' not found. Available APIs: {list(self.register_apis.keys())}")
if service_name not in self.register_data_providers:
return self.fail_response(f"Data provider '{service_name}' not found. Available data providers: {list(self.register_data_providers.keys())}")
endpoints = self.register_apis[service_name].get_endpoints()
endpoints = self.register_data_providers[service_name].get_endpoints()
return self.success_response(endpoints)
except Exception as e:
error_message = str(e)
simplified_message = f"Error getting API endpoints: {error_message[:200]}"
simplified_message = f"Error getting data provider endpoints: {error_message[:200]}"
if len(error_message) > 200:
simplified_message += "..."
return self.fail_response(simplified_message)
@ -76,8 +84,8 @@ Use this tool when you need to discover what endpoints are available.
@openapi_schema({
"type": "function",
"function": {
"name": "execute_api_call",
"description": "Execute a call to a specific API endpoint",
"name": "execute_data_provider_call",
"description": "Execute a call to a specific data provider endpoint",
"parameters": {
"type": "object",
"properties": {
@ -99,7 +107,7 @@ Use this tool when you need to discover what endpoints are available.
}
})
@xml_schema(
tag_name="execute-api-call",
tag_name="execute-data-provider-call",
mappings=[
{"param_name": "service_name", "node_type": "attribute", "path": "service_name"},
{"param_name": "route", "node_type": "attribute", "path": "route"},
@ -107,30 +115,30 @@ Use this tool when you need to discover what endpoints are available.
],
example='''
<!--
The execute-api-call tool makes a request to a specific API endpoint.
Use this tool when you need to call an API endpoint with specific parameters.
The route must be a valid endpoint key obtained from get-api-service-endpoints tool!!
The execute-data-provider-call tool makes a request to a specific data provider endpoint.
Use this tool when you need to call an data provider endpoint with specific parameters.
The route must be a valid endpoint key obtained from get-data-provider-endpoints tool!!
-->
<!-- Example to call linkedIn service with the specific route person -->
<execute-api-call service_name="linkedin" route="person">
<execute-data-provider-call service_name="linkedin" route="person">
{"link": "https://www.linkedin.com/in/johndoe/"}
</execute-api-call>
</execute-data-provider-call>
'''
)
async def execute_api_call(
async def execute_data_provider_call(
self,
service_name: str,
route: str,
payload: str # this actually a json string
) -> ToolResult:
"""
Execute a call to a specific API endpoint.
Execute a call to a specific data provider endpoint.
Parameters:
- service_name: The name of the API service (e.g., 'linkedin')
- service_name: The name of the data provider (e.g., 'linkedin')
- route: The key of the endpoint to call
- payload: The payload to send with the API call
- payload: The payload to send with the data provider call
"""
try:
payload = json.loads(payload)
@ -141,24 +149,24 @@ Use this tool when you need to discover what endpoints are available.
if not route:
return self.fail_response("route is required.")
if service_name not in self.register_apis:
return self.fail_response(f"API '{service_name}' not found. Available APIs: {list(self.register_apis.keys())}")
if service_name not in self.register_data_providers:
return self.fail_response(f"API '{service_name}' not found. Available APIs: {list(self.register_data_providers.keys())}")
api_service = self.register_apis[service_name]
data_provider = self.register_data_providers[service_name]
if route == service_name:
return self.fail_response(f"route '{route}' is the same as service_name '{service_name}'. YOU FUCKING IDIOT!")
if route not in api_service.get_endpoints().keys():
return self.fail_response(f"Endpoint '{route}' not found in {service_name} API.")
if route not in data_provider.get_endpoints().keys():
return self.fail_response(f"Endpoint '{route}' not found in {service_name} data provider.")
result = api_service.call_endpoint(route, payload)
result = data_provider.call_endpoint(route, payload)
return self.success_response(result)
except Exception as e:
error_message = str(e)
print(error_message)
simplified_message = f"Error executing API call: {error_message[:200]}"
simplified_message = f"Error executing data provider call: {error_message[:200]}"
if len(error_message) > 200:
simplified_message += "..."
return self.fail_response(simplified_message)