Configurazioni dell'app (#27)

* Prompt messi in una cartella apposta
* Aggiorna importazioni demo per riflettere la nuova struttura delle cartelle API
* Aggiunto configurazione dell'applicazione
* Spostato ChatManager in app.interface
* Update README.md
* Aggiornato config per app & api
* Rinominato il modulo NewsAPI
* fix main infinite loop
* API base --> core
* pattern singleton per AppConfig.
* Estratto i tools nella loro cartella --> api/tools
* fix main KeyboardInterrupt
* update tests
* Docker & libs
* fix copilot suggestions
This commit was merged in pull request #27.
This commit is contained in:
Giacomo Bertolazzi
2025-10-12 18:05:43 +02:00
committed by GitHub
parent 093a7f5a48
commit 862525cc62
49 changed files with 718 additions and 564 deletions

View File

@@ -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))