Merge pull request #13 from crestalnetwork/feat/twitter-entrypoint

Feat: twitter entrypoint

.github/workflows/lint.yml

@@ -41,7 +41,7 @@ jobs:
                 - title: "Diff"
                   value: ${{ github.event.pull_request.html_url || github.event.head_commit.url }}
                 - title: "Changes"
-                  value: ${{ toJson(github.event.pull_request.title) || toJSON(github.event.head_commit.message) }}
+                  value: ${{ github.event_name == 'pull_request' && toJson(github.event.pull_request.title) || toJSON(github.event.head_commit.message) }}
 
     - uses: actions/checkout@v3
 
@@ -57,17 +57,9 @@ jobs:
       run: |
         poetry install --with dev
         
-    - name: Check formatting with Black
-      run: |
-        poetry run black --check .
-        
-    - name: Check imports with isort
-      run: |
-        poetry run isort --check-only --diff .
-        
-    - name: Run Bandit security scan
+    - name: Ruff Check
       run: |
-        poetry run bandit -r app/ skills/ skill_sets/ utils/ -ll
+        poetry run ruff check
 
     - name: Report CI Success
       if: ${{ success() }}

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## 2024-12-27
+
+### New Features
+- Twitter Skills
+
+### Improvements
+- CI/CD refactoring for better security
+
 ## 2024-12-26
 
 ### Improvements

README.md

@@ -95,6 +95,14 @@ python -m app.entrypoints.autonomous
 
 "Create Agent" and "Try it out" refer to the Docker section.
 
+## Integrations
+
+### Twitter
+[Twitter Integration](docs/twitter.md)
+
+### Coinbase
+Work in progress
+
 ## Configuration
 
 The application can be configured using environment variables or AWS Secrets Manager. Key configuration options:

app/admin/cron.py

@@ -0,0 +1,68 @@
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
+from sqlmodel import Session, update
+
+from app.config.config import config
+from app.models.agent import AgentQuota
+from app.models.db import get_engine, init_db
+
+
+def reset_daily_quotas():
+    """Reset daily quotas for all agents at UTC 00:00.
+    Resets message_count_daily and twitter_count_daily to 0.
+    """
+    with Session(get_engine()) as session:
+        stmt = update(AgentQuota).values(message_count_daily=0, twitter_count_daily=0)
+        session.exec(stmt)
+        session.commit()
+
+
+def reset_monthly_quotas():
+    """Reset monthly quotas for all agents at the start of each month.
+    Resets message_count_monthly and autonomous_count_monthly to 0.
+    """
+    with Session(get_engine()) as session:
+        stmt = update(AgentQuota).values(
+            message_count_monthly=0, autonomous_count_monthly=0
+        )
+        session.exec(stmt)
+        session.commit()
+
+
+def start_scheduler():
+    """Start the APScheduler to run quota reset jobs."""
+    scheduler = BackgroundScheduler()
+
+    # Reset daily quotas at UTC 00:00
+    scheduler.add_job(
+        reset_daily_quotas,
+        trigger=CronTrigger(hour=0, minute=0, timezone="UTC"),
+        id="reset_daily_quotas",
+        name="Reset daily quotas",
+        replace_existing=True,
+    )
+
+    # Reset monthly quotas at UTC 00:00 on the first day of each month
+    scheduler.add_job(
+        reset_monthly_quotas,
+        trigger=CronTrigger(day=1, hour=0, minute=0, timezone="UTC"),
+        id="reset_monthly_quotas",
+        name="Reset monthly quotas",
+        replace_existing=True,
+    )
+
+    scheduler.start()
+    return scheduler
+
+
+if __name__ == "__main__":
+    # Initialize infrastructure
+    init_db(**config.db)
+
+    scheduler = start_scheduler()
+    try:
+        # Keep the script running
+        while True:
+            pass
+    except (KeyboardInterrupt, SystemExit):
+        scheduler.shutdown()

app/config/config.py

@@ -3,15 +3,15 @@
 import logging
 import os
 
-from dotenv import load_dotenv
-
-# Load environment variables from .env file
-load_dotenv()
-
 import botocore.session
 from aws_secretsmanager_caching import SecretCache, SecretCacheConfig
+from dotenv import load_dotenv
 
 from utils.logging import setup_logging
+from utils.slack_alert import init_slack
+
+# Load environment variables from .env file
+load_dotenv()
 
 logger = logging.getLogger(__name__)
 
@@ -66,11 +66,16 @@ def __init__(self):
         self.cdp_api_key_name = self.load("CDP_API_KEY_NAME")
         self.cdp_api_key_private_key = self.load("CDP_API_KEY_PRIVATE_KEY")
         self.openai_api_key = self.load("OPENAI_API_KEY")
-        self.slack_token = self.load("SLACK_TOKEN")  # For alert purposes only
-        self.slack_channel = self.load("SLACK_CHANNEL")
+        self.slack_alert_token = self.load(
+            "SLACK_ALERT_TOKEN"
+        )  # For alert purposes only
+        self.slack_alert_channel = self.load("SLACK_ALERT_CHANNEL")
         # Now we know the env, set up logging
         setup_logging(self.env, self.debug)
         logger.info("config loaded")
+        # If the slack alert token exists, init it
+        if self.slack_alert_token and self.slack_alert_channel:
+            init_slack(self.slack_alert_token, self.slack_alert_channel)
 
     def load(self, key, default=None):
         """Load a secret from the secrets map or env"""

app/core/ai.py

@@ -196,7 +196,7 @@ def execute_agent(aid: str, prompt: str, thread_id: str) -> list[str]:
     # cold start
     if aid not in agents:
         initialize_agent(aid)
-        resp_debug.append(f"[ Agent cold start ... ]")
+        resp_debug.append("[ Agent cold start ... ]")
         resp_debug.append(
             f"\n------------------- start cost: {time.perf_counter() - last:.3f} seconds\n"
         )

app/entrypoints/api.py

@@ -47,6 +47,7 @@ async def lifespan(app: FastAPI):
         app: FastAPI application instance
     """
     # This part will run before the API server start
+    # Initialize infrastructure
     init_db(**config.db)
     logger.info("API server start")
     yield
@@ -160,3 +161,45 @@ def create_agent(agent: Agent, db: Session = Depends(get_db)) -> Agent:
     # TODO: change here when multiple instances deploy
     initialize_agent(agent.id)
     return latest_agent
+
+
+@app.get("/agents")
+def get_agents(db: Session = Depends(get_db)):
+    """Get all agents with their quota information.
+
+    Args:
+        db: Database session
+
+    Returns:
+        list: List of agents with their quota information
+    """
+    # Query agents and quotas together
+    query = select(Agent.id, Agent.name, AgentQuota).join(
+        AgentQuota, Agent.id == AgentQuota.id, isouter=True
+    )
+
+    results = db.exec(query).all()
+
+    # Format the response
+    agents = []
+    for result in results:
+        agent_data = {
+            "id": result.id,
+            "name": result.name,
+            "quota": {
+                "plan": result[2].plan if result[2] else "none",
+                "message_count_total": (
+                    result[2].message_count_total if result[2] else 0
+                ),
+                "message_limit_total": (
+                    result[2].message_limit_total if result[2] else 0
+                ),
+                "last_message_time": result[2].last_message_time if result[2] else None,
+                "last_autonomous_time": (
+                    result[2].last_autonomous_time if result[2] else None
+                ),
+            },
+        }
+        agents.append(agent_data)
+
+    return agents

app/entrypoints/autonomous.py

@@ -85,7 +85,7 @@ def run_autonomous_action(aid: str, prompt: str):
 
 
 if __name__ == "__main__":
-    # Initialize database connection
+    # Initialize infrastructure
     init_db(**config.db)
 
     # Initialize scheduler

app/entrypoints/twitter.py

@@ -0,0 +1,152 @@
+import logging
+import signal
+import sys
+import tweepy
+from datetime import datetime, timedelta, timezone
+from apscheduler.schedulers.blocking import BlockingScheduler
+from sqlmodel import Session, select
+
+from app.config.config import config
+from app.core.ai import execute_agent
+from app.models.agent import Agent, AgentPluginData, AgentQuota
+from app.models.db import get_engine, init_db
+
+logger = logging.getLogger(__name__)
+
+# Set debug_resp to False
+config.debug_resp = False
+
+
+def create_twitter_client(config: dict) -> tweepy.Client:
+    """Create a Twitter client from config.
+
+    Args:
+        config: Dictionary containing Twitter credentials
+
+    Returns:
+        tweepy.Client instance
+    """
+    return tweepy.Client(
+        bearer_token=config.get("bearer_token"),
+        consumer_key=config.get("consumer_key"),
+        consumer_secret=config.get("consumer_secret"),
+        access_token=config.get("access_token"),
+        access_token_secret=config.get("access_token_secret"),
+    )
+
+
+def run_twitter_agents():
+    """Get all agents from the database which twitter is enabled,
+    check their twitter config, get mentions, and process them."""
+    engine = get_engine()
+    with Session(engine) as db:
+        # Get all twitter-enabled agents
+        agents = db.exec(
+            select(Agent).where(
+                Agent.twitter_enabled == True,  # noqa: E712
+                Agent.twitter_config != None,  # noqa: E711
+            )
+        ).all()
+
+        for agent in agents:
+            try:
+                # Get agent quota
+                quota = AgentQuota.get(agent.id, db)
+
+                # Check if agent has quota
+                if not quota.has_twitter_quota(db):
+                    logger.warning(
+                        f"Agent {agent.id} has no twitter quota. "
+                        f"Daily: {quota.twitter_count_daily}/{quota.twitter_limit_daily}, "
+                        f"Total: {quota.twitter_count_total}/{quota.twitter_limit_total}"
+                    )
+                    continue
+
+                # Initialize Twitter client
+                if not agent.twitter_config:
+                    logger.warning(f"Agent {agent.id} has no valid twitter config")
+                    continue
+
+                client = create_twitter_client(agent.twitter_config)
+                me = client.get_me()
+                if not me.data:
+                    logger.error(
+                        f"Failed to get Twitter user info for agent {agent.id}"
+                    )
+                    continue
+
+                # Get last tweet id from plugin data
+                plugin_data = AgentPluginData.get(agent.id, "twitter", "entrypoint", db)
+                since_id = None
+                if plugin_data and plugin_data.data:
+                    since_id = plugin_data.data.get("last_tweet_id")
+                # Always get mentions for the last day
+                start_time = (
+                    datetime.now(tz=timezone.utc) - timedelta(days=1)
+                ).isoformat(timespec="milliseconds")
+                # Get mentions
+                mentions = client.get_users_mentions(
+                    id=me.data.id,
+                    max_results=10,
+                    since_id=since_id,
+                    start_time=start_time,
+                    tweet_fields=["created_at", "author_id", "text"],
+                )
+
+                if not mentions.data:
+                    logger.info(f"No new mentions for agent {agent.id}")
+                    continue
+
+                # Update last tweet id
+                if mentions.meta:
+                    last_tweet_id = mentions.meta.get("newest_id")
+                    plugin_data = AgentPluginData(
+                        agent_id=agent.id,
+                        plugin="twitter",
+                        key="entrypoint",
+                        data={"last_tweet_id": last_tweet_id},
+                    )
+                    plugin_data.save(db)
+                else:
+                    raise Exception(f"Failed to get last tweet id for agent {agent.id}")
+
+                # Process each mention
+                for mention in mentions.data:
+                    thread_id = f"{agent.id}-twitter-{mention.author_id}"
+                    response = execute_agent(agent.id, mention.text, thread_id)
+
+                    # Reply to the tweet
+                    client.create_tweet(
+                        text="\n".join(response), in_reply_to_tweet_id=mention.id
+                    )
+
+                # Update quota
+                quota.add_twitter(db)
+
+            except Exception as e:
+                logger.error(
+                    f"Error processing twitter mentions for agent {agent.id}: {str(e)}"
+                )
+                continue
+
+
+if __name__ == "__main__":
+    # Initialize infrastructure
+    init_db(**config.db)
+
+    # Create scheduler
+    scheduler = BlockingScheduler()
+    scheduler.add_job(run_twitter_agents, "interval", minutes=1)
+
+    # Register signal handlers
+    def signal_handler(signum, frame):
+        scheduler.shutdown()
+        sys.exit(0)
+
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    try:
+        scheduler.start()
+    except (KeyboardInterrupt, SystemExit):
+        pass