Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Implement configurable API providers from configs.yaml (#43)

Browse Source 
* Implement configurable API providers from configs.yaml
* Refactor provider filtering to use WrapperHandler helper function
* Refactor API wrapper initialization to streamline configuration handling
* Refactor agent retrieval to use specific API tools directly
This commit was merged in pull request #43.
This commit is contained in:
Copilot
2025-10-27 17:53:36 +00:00
committed by GitHub
parent 6a9d8b354b
commit c7a3199f27
9 changed files with 63 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
configs.yaml
View File

@@ -32,10 +32,9 @@ models:
api:
retry_attempts: 3
retry_delay_seconds: 2
# TODO Magari implementare un sistema per settare i providers
market_providers: [BinanceWrapper, YFinanceWrapper]
news_providers: [GoogleNewsWrapper, DuckDuckGoWrapper]
social_providers: [RedditWrapper]
market_providers: [YFinanceWrapper, BinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper]
news_providers: [DuckDuckGoWrapper, GoogleNewsWrapper, NewsApiWrapper, CryptoPanicWrapper]
social_providers: [RedditWrapper, XWrapper, ChanWrapper]
agents:
strategy: Conservative

21
src/app/agents/core.py
View File

@@ -103,10 +103,9 @@ class PipelineInputs:
# Agent getters
# ======================
def get_agent_team(self) -> Team:
market, news, social = self.get_tools()
market_agent = self.team_model.get_agent(MARKET_INSTRUCTIONS, "Market Agent", tools=[market])
news_agent = self.team_model.get_agent(NEWS_INSTRUCTIONS, "News Agent", tools=[news])
social_agent = self.team_model.get_agent(SOCIAL_INSTRUCTIONS, "Socials Agent", tools=[social])
market_agent = self.team_model.get_agent(MARKET_INSTRUCTIONS, "Market Agent", tools=[MarketAPIsTool()])
news_agent = self.team_model.get_agent(NEWS_INSTRUCTIONS, "News Agent", tools=[NewsAPIsTool()])
social_agent = self.team_model.get_agent(SOCIAL_INSTRUCTIONS, "Socials Agent", tools=[SocialAPIsTool()])
return Team(
model=self.team_leader_model.get_model(TEAM_LEADER_INSTRUCTIONS),
name="CryptoAnalysisTeam",
@@ -120,20 +119,6 @@ class PipelineInputs:
def get_agent_report_generator(self) -> Agent:
return self.report_generation_model.get_agent(REPORT_GENERATION_INSTRUCTIONS, "Report Generator Agent")
def get_tools(self) -> tuple[MarketAPIsTool, NewsAPIsTool, SocialAPIsTool]:
"""
Restituisce la lista di tools disponibili per gli agenti.
"""
api = self.configs.api
market_tool = MarketAPIsTool()
market_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
news_tool = NewsAPIsTool()
news_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
social_tool = SocialAPIsTool()
social_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
return market_tool, news_tool, social_tool
def __str__(self) -> str:
return "\n".join([
f"Query Check: {self.query_analyzer_model.label}",

25
src/app/api/tools/market_tool.py
View File

@@ -2,30 +2,29 @@ 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
from app.configs import AppConfig
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
Providers can be configured in configs.yaml under api.market_providers.
"""
def __init__(self):
"""
Initialize the MarketAPIsTool with multiple market API wrappers.
The following wrappers are included in this order:
- BinanceWrapper
- YFinanceWrapper
- CoinBaseWrapper
- CryptoCompareWrapper
Initialize the MarketAPIsTool with market API wrappers configured in configs.yaml.
The order of wrappers is determined by the api.market_providers list in the configuration.
"""
wrappers: list[type[MarketWrapper]] = [BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper]
self.handler = WrapperHandler.build_wrappers(wrappers)
config = AppConfig()
self.handler = WrapperHandler.build_wrappers(
constructors=[BinanceWrapper, YFinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper],
filters=config.api.market_providers,
try_per_wrapper=config.api.retry_attempts,
retry_delay=config.api.retry_delay_seconds
)
Toolkit.__init__( # type: ignore
self,

27
src/app/api/tools/news_tool.py
View File

@@ -2,15 +2,13 @@ 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
from app.configs import AppConfig
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
This class supports retrieving top headlines and latest news articles by querying multiple sources.
Providers can be configured in configs.yaml under api.news_providers.
By default, it returns results from the first successful wrapper.
Optionally, it can be configured to collect articles from all wrappers.
@@ -19,16 +17,17 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
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.
Initialize the NewsAPIsTool with news API wrappers configured in configs.yaml.
The order of wrappers is determined by the api.news_providers list in the configuration.
"""
wrappers: list[type[NewsWrapper]] = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
self.handler = WrapperHandler.build_wrappers(wrappers)
config = AppConfig()
self.handler = WrapperHandler.build_wrappers(
constructors=[NewsApiWrapper, GoogleNewsWrapper, CryptoPanicWrapper, DuckDuckGoWrapper],
filters=config.api.news_providers,
try_per_wrapper=config.api.retry_attempts,
retry_delay=config.api.retry_delay_seconds
)
Toolkit.__init__( # type: ignore
self,

20
src/app/api/tools/social_tool.py
View File

@@ -2,13 +2,14 @@ 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 *
from app.configs import AppConfig
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
This class supports retrieving top crypto-related posts by querying multiple sources.
Providers can be configured in configs.yaml under api.social_providers.
By default, it returns results from the first successful wrapper.
Optionally, it can be configured to collect posts from all wrappers.
@@ -17,14 +18,17 @@ class SocialAPIsTool(SocialWrapper, Toolkit):
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.
Initialize the SocialAPIsTool with social media API wrappers configured in configs.yaml.
The order of wrappers is determined by the api.social_providers list in the configuration.
"""
config = AppConfig()
wrappers: list[type[SocialWrapper]] = [RedditWrapper, XWrapper, ChanWrapper]
self.handler = WrapperHandler.build_wrappers(wrappers)
self.handler = WrapperHandler.build_wrappers(
constructors=[RedditWrapper, XWrapper, ChanWrapper],
filters=config.api.social_providers,
try_per_wrapper=config.api.retry_attempts,
retry_delay=config.api.retry_delay_seconds
)
Toolkit.__init__( # type: ignore
self,

14
src/app/api/wrapper_handler.py
View File

@@ -131,13 +131,19 @@ class WrapperHandler(Generic[WrapperType]):
return f"{e} [\"{last_frame.filename}\", line {last_frame.lineno}]"
@staticmethod
def build_wrappers(constructors: list[type[WrapperClassType]], try_per_wrapper: int = 3, retry_delay: int = 2, kwargs: dict[str, Any] | None = None) -> 'WrapperHandler[WrapperClassType]':
def build_wrappers(
constructors: list[type[WrapperClassType]],
filters: list[str] | None = None,
try_per_wrapper: int = 3,
retry_delay: int = 2,
kwargs: dict[str, Any] | None = None) -> 'WrapperHandler[WrapperClassType]':
"""
Builds a WrapperHandler instance with the given wrapper constructors.
It attempts to initialize each wrapper and logs a warning if any cannot be initialized.
Only successfully initialized wrappers are included in the handler.
Args:
constructors (list[type[W]]): An iterable of wrapper classes to instantiate. e.g. [WrapperA, WrapperB]
filters (list[str] | None): Optional list of provider names to filter the constructors.
try_per_wrapper (int): Number of retries per wrapper before switching to the next.
retry_delay (int): Delay in seconds between retries.
kwargs (dict | None): Optional dictionary with keyword arguments common to all wrappers.
@@ -148,6 +154,10 @@ class WrapperHandler(Generic[WrapperType]):
"""
assert WrapperHandler.__check(constructors), f"All constructors must be classes. Received: {constructors}"
# Order of wrappers is now determined by the order in filters
if filters:
constructors = [c for name in filters for c in constructors if c.__name__ == name]
result: list[WrapperClassType] = []
for wrapper_class in constructors:
try:
@@ -156,4 +166,4 @@ class WrapperHandler(Generic[WrapperType]):
except Exception as e:
logging.warning(f"'{wrapper_class.__name__}' cannot be initialized: {e}")
return WrapperHandler(result, try_per_wrapper, retry_delay)
return WrapperHandler(result, try_per_wrapper, retry_delay)

3
src/app/configs.py
View File

@@ -57,6 +57,9 @@ class AppModel(BaseModel):
class APIConfig(BaseModel):
retry_attempts: int = 3
retry_delay_seconds: int = 2
market_providers: list[str] = []
news_providers: list[str] = []
social_providers: list[str] = []
class Strategy(BaseModel):
name: str = "Conservative"

8
tests/tools/test_market_tool.py
View File

@@ -7,14 +7,14 @@ from app.api.tools import MarketAPIsTool
@pytest.mark.api
class TestMarketAPIsTool:
def test_wrapper_initialization(self):
market_wrapper = MarketAPIsTool("EUR")
market_wrapper = MarketAPIsTool()
assert market_wrapper is not None
assert hasattr(market_wrapper, 'get_product')
assert hasattr(market_wrapper, 'get_products')
assert hasattr(market_wrapper, 'get_historical_prices')
def test_wrapper_capabilities(self):
market_wrapper = MarketAPIsTool("EUR")
market_wrapper = MarketAPIsTool()
capabilities: list[str] = []
if hasattr(market_wrapper, 'get_product'):
capabilities.append('single_product')
@@ -25,7 +25,7 @@ class TestMarketAPIsTool:
assert len(capabilities) > 0
def test_market_data_retrieval(self):
market_wrapper = MarketAPIsTool("EUR")
market_wrapper = MarketAPIsTool()
btc_product = market_wrapper.get_product("BTC")
assert btc_product is not None
assert hasattr(btc_product, 'symbol')
@@ -34,7 +34,7 @@ class TestMarketAPIsTool:
def test_error_handling(self):
try:
market_wrapper = MarketAPIsTool("EUR")
market_wrapper = MarketAPIsTool()
fake_product = market_wrapper.get_product("NONEXISTENT_CRYPTO_SYMBOL_12345")
assert fake_product is None or fake_product.price == 0
except Exception as _:

2
tests/tools/test_socials_tool.py
View File

@@ -19,7 +19,7 @@ class TestSocialAPIsTool:
assert post.title is not None
assert post.timestamp is not None
def test_social_api_tool_get_top__all_results(self):
def test_social_api_tool_get_top_all_results(self):
tool = SocialAPIsTool()
result = tool.handler.try_call_all(lambda w: w.get_top_crypto_posts(limit=2))
assert isinstance(result, dict)