From 63c26a7dfd056a23b1b94cfef473a8972c5c0f40 Mon Sep 17 00:00:00 2001 From: Berack96 Date: Sun, 12 Oct 2025 17:30:18 +0200 Subject: [PATCH] Estratto i tools nella loro cartella --> api/tools --- src/app/agents/team.py | 4 +- src/app/api/markets/__init__.py | 81 +------------------------------- src/app/api/news/__init__.py | 73 +--------------------------- src/app/api/social/__init__.py | 52 +------------------- src/app/api/tools/__init__.py | 5 ++ src/app/api/tools/market_tool.py | 80 +++++++++++++++++++++++++++++++ src/app/api/tools/news_tool.py | 72 ++++++++++++++++++++++++++++ src/app/api/tools/social_tool.py | 51 ++++++++++++++++++++ 8 files changed, 212 insertions(+), 206 deletions(-) create mode 100644 src/app/api/tools/__init__.py create mode 100644 src/app/api/tools/market_tool.py create mode 100644 src/app/api/tools/news_tool.py create mode 100644 src/app/api/tools/social_tool.py diff --git a/src/app/agents/team.py b/src/app/agents/team.py index 9de5b64..4fcad4e 100644 --- a/src/app/agents/team.py +++ b/src/app/agents/team.py @@ -1,7 +1,5 @@ from agno.team import Team -from app.api.markets import MarketAPIsTool -from app.api.news import NewsAPIsTool -from app.api.social import SocialAPIsTool +from app.api.tools import * from app.agents.prompts import * from app.configs import AppConfig, AppModel diff --git a/src/app/api/markets/__init__.py b/src/app/api/markets/__init__.py index 015e548..af4d57b 100644 --- a/src/app/api/markets/__init__.py +++ b/src/app/api/markets/__init__.py @@ -1,86 +1,7 @@ -from agno.tools import Toolkit -from app.api.wrapper_handler import WrapperHandler -from app.api.core.markets import MarketWrapper, Price, ProductInfo from app.api.markets.binance import BinanceWrapper from app.api.markets.coinbase import CoinBaseWrapper from app.api.markets.cryptocompare import CryptoCompareWrapper from app.api.markets.yfinance import YFinanceWrapper -__all__ = [ "MarketAPIsTool", "BinanceWrapper", "CoinBaseWrapper", "CryptoCompareWrapper", "YFinanceWrapper", "ProductInfo", "Price" ] +__all__ = ["BinanceWrapper", "CoinBaseWrapper", "CryptoCompareWrapper", "YFinanceWrapper"] - -class MarketAPIsTool(MarketWrapper, Toolkit): - """ - Class that aggregates multiple market API wrappers and manages them using WrapperHandler. - This class supports retrieving product information and historical prices. - This class can also aggregate data from multiple sources to provide a more comprehensive view of the market. - The following wrappers are included in this order: - - BinanceWrapper - - YFinanceWrapper - - CoinBaseWrapper - - CryptoCompareWrapper - """ - - def __init__(self, currency: str = "USD"): - """ - Initialize the MarketAPIsTool with multiple market API wrappers. - The following wrappers are included in this order: - - BinanceWrapper - - YFinanceWrapper - - CoinBaseWrapper - - CryptoCompareWrapper - Args: - currency (str): Valuta in cui restituire i prezzi. Default è "USD". - """ - kwargs = {"currency": currency or "USD"} - wrappers: list[type[MarketWrapper]] = [BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper] - self.handler = WrapperHandler.build_wrappers(wrappers, kwargs=kwargs) - - Toolkit.__init__( # type: ignore - self, - name="Market APIs Toolkit", - tools=[ - self.get_product, - self.get_products, - self.get_historical_prices, - self.get_products_aggregated, - self.get_historical_prices_aggregated, - ], - ) - - def get_product(self, asset_id: str) -> ProductInfo: - return self.handler.try_call(lambda w: w.get_product(asset_id)) - def get_products(self, asset_ids: list[str]) -> list[ProductInfo]: - return self.handler.try_call(lambda w: w.get_products(asset_ids)) - def get_historical_prices(self, asset_id: str, limit: int = 100) -> list[Price]: - return self.handler.try_call(lambda w: w.get_historical_prices(asset_id, limit)) - - - def get_products_aggregated(self, asset_ids: list[str]) -> list[ProductInfo]: - """ - Restituisce i dati aggregati per una lista di asset_id.\n - Attenzione che si usano tutte le fonti, quindi potrebbe usare molte chiamate API (che potrebbero essere a pagamento). - Args: - asset_ids (list[str]): Lista di asset_id da cercare. - Returns: - list[ProductInfo]: Lista di ProductInfo aggregati. - Raises: - Exception: If all wrappers fail to provide results. - """ - all_products = self.handler.try_call_all(lambda w: w.get_products(asset_ids)) - return ProductInfo.aggregate(all_products) - - def get_historical_prices_aggregated(self, asset_id: str = "BTC", limit: int = 100) -> list[Price]: - """ - Restituisce i dati storici aggregati per un asset_id. Usa i dati di tutte le fonti disponibili e li aggrega.\n - Attenzione che si usano tutte le fonti, quindi potrebbe usare molte chiamate API (che potrebbero essere a pagamento). - Args: - asset_id (str): Asset ID da cercare. - limit (int): Numero massimo di dati storici da restituire. - Returns: - list[Price]: Lista di Price aggregati. - Raises: - Exception: If all wrappers fail to provide results. - """ - all_prices = self.handler.try_call_all(lambda w: w.get_historical_prices(asset_id, limit)) - return Price.aggregate(all_prices) diff --git a/src/app/api/news/__init__.py b/src/app/api/news/__init__.py index eb36d8a..e9fb781 100644 --- a/src/app/api/news/__init__.py +++ b/src/app/api/news/__init__.py @@ -1,78 +1,7 @@ -from agno.tools import Toolkit -from app.api.wrapper_handler import WrapperHandler -from app.api.core.news import NewsWrapper, Article from app.api.news.newsapi import NewsApiWrapper from app.api.news.googlenews import GoogleNewsWrapper from app.api.news.cryptopanic_api import CryptoPanicWrapper from app.api.news.duckduckgo import DuckDuckGoWrapper -__all__ = ["NewsAPIsTool", "NewsApiWrapper", "GoogleNewsWrapper", "CryptoPanicWrapper", "DuckDuckGoWrapper", "Article"] +__all__ = ["NewsApiWrapper", "GoogleNewsWrapper", "CryptoPanicWrapper", "DuckDuckGoWrapper"] - -class NewsAPIsTool(NewsWrapper, Toolkit): - """ - Aggregates multiple news API wrappers and manages them using WrapperHandler. - This class supports retrieving top headlines and latest news articles by querying multiple sources: - - GoogleNewsWrapper - - DuckDuckGoWrapper - - NewsApiWrapper - - CryptoPanicWrapper - - By default, it returns results from the first successful wrapper. - Optionally, it can be configured to collect articles from all wrappers. - If no wrapper succeeds, an exception is raised. - """ - - def __init__(self): - """ - Initialize the NewsAPIsTool with multiple news API wrappers. - The tool uses WrapperHandler to manage and invoke the different news API wrappers. - The following wrappers are included in this order: - - GoogleNewsWrapper. - - DuckDuckGoWrapper. - - NewsApiWrapper. - - CryptoPanicWrapper. - """ - wrappers: list[type[NewsWrapper]] = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper] - self.handler = WrapperHandler.build_wrappers(wrappers) - - Toolkit.__init__( # type: ignore - self, - name="News APIs Toolkit", - tools=[ - self.get_top_headlines, - self.get_latest_news, - self.get_top_headlines_aggregated, - self.get_latest_news_aggregated, - ], - ) - - def get_top_headlines(self, limit: int = 100) -> list[Article]: - return self.handler.try_call(lambda w: w.get_top_headlines(limit)) - def get_latest_news(self, query: str, limit: int = 100) -> list[Article]: - return self.handler.try_call(lambda w: w.get_latest_news(query, limit)) - - def get_top_headlines_aggregated(self, limit: int = 100) -> dict[str, list[Article]]: - """ - Calls get_top_headlines on all wrappers/providers and returns a dictionary mapping their names to their articles. - Args: - limit (int): Maximum number of articles to retrieve from each provider. - Returns: - dict[str, list[Article]]: A dictionary mapping providers names to their list of Articles - Raises: - Exception: If all wrappers fail to provide results. - """ - return self.handler.try_call_all(lambda w: w.get_top_headlines(limit)) - - def get_latest_news_aggregated(self, query: str, limit: int = 100) -> dict[str, list[Article]]: - """ - Calls get_latest_news on all wrappers/providers and returns a dictionary mapping their names to their articles. - Args: - query (str): The search query to find relevant news articles. - limit (int): Maximum number of articles to retrieve from each provider. - Returns: - dict[str, list[Article]]: A dictionary mapping providers names to their list of Articles - Raises: - Exception: If all wrappers fail to provide results. - """ - return self.handler.try_call_all(lambda w: w.get_latest_news(query, limit)) diff --git a/src/app/api/social/__init__.py b/src/app/api/social/__init__.py index 37b1e30..f50ca7c 100644 --- a/src/app/api/social/__init__.py +++ b/src/app/api/social/__init__.py @@ -1,53 +1,3 @@ -from agno.tools import Toolkit -from app.api.wrapper_handler import WrapperHandler -from app.api.core.social import SocialPost, SocialWrapper from app.api.social.reddit import RedditWrapper -__all__ = ["SocialAPIsTool", "RedditWrapper", "SocialPost"] - - -class SocialAPIsTool(SocialWrapper, Toolkit): - """ - Aggregates multiple social media API wrappers and manages them using WrapperHandler. - This class supports retrieving top crypto-related posts by querying multiple sources: - - RedditWrapper - - By default, it returns results from the first successful wrapper. - Optionally, it can be configured to collect posts from all wrappers. - If no wrapper succeeds, an exception is raised. - """ - - def __init__(self): - """ - Initialize the SocialAPIsTool with multiple social media API wrappers. - The tool uses WrapperHandler to manage and invoke the different social media API wrappers. - The following wrappers are included in this order: - - RedditWrapper. - """ - - wrappers: list[type[SocialWrapper]] = [RedditWrapper] - self.handler = WrapperHandler.build_wrappers(wrappers) - - Toolkit.__init__( # type: ignore - self, - name="Socials Toolkit", - tools=[ - self.get_top_crypto_posts, - self.get_top_crypto_posts_aggregated, - ], - ) - - def get_top_crypto_posts(self, limit: int = 5) -> list[SocialPost]: - return self.handler.try_call(lambda w: w.get_top_crypto_posts(limit)) - - def get_top_crypto_posts_aggregated(self, limit_per_wrapper: int = 5) -> dict[str, list[SocialPost]]: - """ - Calls get_top_crypto_posts on all wrappers/providers and returns a dictionary mapping their names to their posts. - Args: - limit_per_wrapper (int): Maximum number of posts to retrieve from each provider. - Returns: - dict[str, list[SocialPost]]: A dictionary where keys are wrapper names and values are lists of SocialPost objects. - Raises: - Exception: If all wrappers fail to provide results. - """ - return self.handler.try_call_all(lambda w: w.get_top_crypto_posts(limit_per_wrapper)) +__all__ = ["RedditWrapper"] diff --git a/src/app/api/tools/__init__.py b/src/app/api/tools/__init__.py new file mode 100644 index 0000000..da9c32a --- /dev/null +++ b/src/app/api/tools/__init__.py @@ -0,0 +1,5 @@ +from app.api.tools.market_tool import MarketAPIsTool +from app.api.tools.social_tool import SocialAPIsTool +from app.api.tools.news_tool import NewsAPIsTool + +__all__ = ["MarketAPIsTool", "NewsAPIsTool", "SocialAPIsTool"] \ No newline at end of file diff --git a/src/app/api/tools/market_tool.py b/src/app/api/tools/market_tool.py new file mode 100644 index 0000000..36f6286 --- /dev/null +++ b/src/app/api/tools/market_tool.py @@ -0,0 +1,80 @@ +from agno.tools import Toolkit +from app.api.wrapper_handler import WrapperHandler +from app.api.core.markets import MarketWrapper, Price, ProductInfo +from app.api.markets import BinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper, YFinanceWrapper + +class MarketAPIsTool(MarketWrapper, Toolkit): + """ + Class that aggregates multiple market API wrappers and manages them using WrapperHandler. + This class supports retrieving product information and historical prices. + This class can also aggregate data from multiple sources to provide a more comprehensive view of the market. + The following wrappers are included in this order: + - BinanceWrapper + - YFinanceWrapper + - CoinBaseWrapper + - CryptoCompareWrapper + """ + + def __init__(self, currency: str = "USD"): + """ + Initialize the MarketAPIsTool with multiple market API wrappers. + The following wrappers are included in this order: + - BinanceWrapper + - YFinanceWrapper + - CoinBaseWrapper + - CryptoCompareWrapper + Args: + currency (str): Valuta in cui restituire i prezzi. Default è "USD". + """ + kwargs = {"currency": currency or "USD"} + wrappers: list[type[MarketWrapper]] = [BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper] + self.handler = WrapperHandler.build_wrappers(wrappers, kwargs=kwargs) + + Toolkit.__init__( # type: ignore + self, + name="Market APIs Toolkit", + tools=[ + self.get_product, + self.get_products, + self.get_historical_prices, + self.get_products_aggregated, + self.get_historical_prices_aggregated, + ], + ) + + def get_product(self, asset_id: str) -> ProductInfo: + return self.handler.try_call(lambda w: w.get_product(asset_id)) + def get_products(self, asset_ids: list[str]) -> list[ProductInfo]: + return self.handler.try_call(lambda w: w.get_products(asset_ids)) + def get_historical_prices(self, asset_id: str, limit: int = 100) -> list[Price]: + return self.handler.try_call(lambda w: w.get_historical_prices(asset_id, limit)) + + + def get_products_aggregated(self, asset_ids: list[str]) -> list[ProductInfo]: + """ + Restituisce i dati aggregati per una lista di asset_id.\n + Attenzione che si usano tutte le fonti, quindi potrebbe usare molte chiamate API (che potrebbero essere a pagamento). + Args: + asset_ids (list[str]): Lista di asset_id da cercare. + Returns: + list[ProductInfo]: Lista di ProductInfo aggregati. + Raises: + Exception: If all wrappers fail to provide results. + """ + all_products = self.handler.try_call_all(lambda w: w.get_products(asset_ids)) + return ProductInfo.aggregate(all_products) + + def get_historical_prices_aggregated(self, asset_id: str = "BTC", limit: int = 100) -> list[Price]: + """ + Restituisce i dati storici aggregati per un asset_id. Usa i dati di tutte le fonti disponibili e li aggrega.\n + Attenzione che si usano tutte le fonti, quindi potrebbe usare molte chiamate API (che potrebbero essere a pagamento). + Args: + asset_id (str): Asset ID da cercare. + limit (int): Numero massimo di dati storici da restituire. + Returns: + list[Price]: Lista di Price aggregati. + Raises: + Exception: If all wrappers fail to provide results. + """ + all_prices = self.handler.try_call_all(lambda w: w.get_historical_prices(asset_id, limit)) + return Price.aggregate(all_prices) diff --git a/src/app/api/tools/news_tool.py b/src/app/api/tools/news_tool.py new file mode 100644 index 0000000..ab67f8b --- /dev/null +++ b/src/app/api/tools/news_tool.py @@ -0,0 +1,72 @@ +from agno.tools import Toolkit +from app.api.wrapper_handler import WrapperHandler +from app.api.core.news import NewsWrapper, Article +from app.api.news import NewsApiWrapper, GoogleNewsWrapper, CryptoPanicWrapper, DuckDuckGoWrapper + +class NewsAPIsTool(NewsWrapper, Toolkit): + """ + Aggregates multiple news API wrappers and manages them using WrapperHandler. + This class supports retrieving top headlines and latest news articles by querying multiple sources: + - GoogleNewsWrapper + - DuckDuckGoWrapper + - NewsApiWrapper + - CryptoPanicWrapper + + By default, it returns results from the first successful wrapper. + Optionally, it can be configured to collect articles from all wrappers. + If no wrapper succeeds, an exception is raised. + """ + + def __init__(self): + """ + Initialize the NewsAPIsTool with multiple news API wrappers. + The tool uses WrapperHandler to manage and invoke the different news API wrappers. + The following wrappers are included in this order: + - GoogleNewsWrapper. + - DuckDuckGoWrapper. + - NewsApiWrapper. + - CryptoPanicWrapper. + """ + wrappers: list[type[NewsWrapper]] = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper] + self.handler = WrapperHandler.build_wrappers(wrappers) + + Toolkit.__init__( # type: ignore + self, + name="News APIs Toolkit", + tools=[ + self.get_top_headlines, + self.get_latest_news, + self.get_top_headlines_aggregated, + self.get_latest_news_aggregated, + ], + ) + + def get_top_headlines(self, limit: int = 100) -> list[Article]: + return self.handler.try_call(lambda w: w.get_top_headlines(limit)) + def get_latest_news(self, query: str, limit: int = 100) -> list[Article]: + return self.handler.try_call(lambda w: w.get_latest_news(query, limit)) + + def get_top_headlines_aggregated(self, limit: int = 100) -> dict[str, list[Article]]: + """ + Calls get_top_headlines on all wrappers/providers and returns a dictionary mapping their names to their articles. + Args: + limit (int): Maximum number of articles to retrieve from each provider. + Returns: + dict[str, list[Article]]: A dictionary mapping providers names to their list of Articles + Raises: + Exception: If all wrappers fail to provide results. + """ + return self.handler.try_call_all(lambda w: w.get_top_headlines(limit)) + + def get_latest_news_aggregated(self, query: str, limit: int = 100) -> dict[str, list[Article]]: + """ + Calls get_latest_news on all wrappers/providers and returns a dictionary mapping their names to their articles. + Args: + query (str): The search query to find relevant news articles. + limit (int): Maximum number of articles to retrieve from each provider. + Returns: + dict[str, list[Article]]: A dictionary mapping providers names to their list of Articles + Raises: + Exception: If all wrappers fail to provide results. + """ + return self.handler.try_call_all(lambda w: w.get_latest_news(query, limit)) diff --git a/src/app/api/tools/social_tool.py b/src/app/api/tools/social_tool.py new file mode 100644 index 0000000..630e14d --- /dev/null +++ b/src/app/api/tools/social_tool.py @@ -0,0 +1,51 @@ +from agno.tools import Toolkit +from app.api.wrapper_handler import WrapperHandler +from app.api.core.social import SocialPost, SocialWrapper +from app.api.social import RedditWrapper + + +class SocialAPIsTool(SocialWrapper, Toolkit): + """ + Aggregates multiple social media API wrappers and manages them using WrapperHandler. + This class supports retrieving top crypto-related posts by querying multiple sources: + - RedditWrapper + + By default, it returns results from the first successful wrapper. + Optionally, it can be configured to collect posts from all wrappers. + If no wrapper succeeds, an exception is raised. + """ + + def __init__(self): + """ + Initialize the SocialAPIsTool with multiple social media API wrappers. + The tool uses WrapperHandler to manage and invoke the different social media API wrappers. + The following wrappers are included in this order: + - RedditWrapper. + """ + + wrappers: list[type[SocialWrapper]] = [RedditWrapper] + self.handler = WrapperHandler.build_wrappers(wrappers) + + Toolkit.__init__( # type: ignore + self, + name="Socials Toolkit", + tools=[ + self.get_top_crypto_posts, + self.get_top_crypto_posts_aggregated, + ], + ) + + def get_top_crypto_posts(self, limit: int = 5) -> list[SocialPost]: + return self.handler.try_call(lambda w: w.get_top_crypto_posts(limit)) + + def get_top_crypto_posts_aggregated(self, limit_per_wrapper: int = 5) -> dict[str, list[SocialPost]]: + """ + Calls get_top_crypto_posts on all wrappers/providers and returns a dictionary mapping their names to their posts. + Args: + limit_per_wrapper (int): Maximum number of posts to retrieve from each provider. + Returns: + dict[str, list[SocialPost]]: A dictionary where keys are wrapper names and values are lists of SocialPost objects. + Raises: + Exception: If all wrappers fail to provide results. + """ + return self.handler.try_call_all(lambda w: w.get_top_crypto_posts(limit_per_wrapper))