feat: twitter langchain refinements (#43)

* first pass refinemetns to match the rest * formatting * forgot to commit twitter action * rmoving unnecessairy tweep import validation * formatting and import validation * changelog --------- Co-authored-by: John Peterson <[email protected]>
coinbase · Nov 15, 2024 · d8a0020 · d8a0020
1 parent 0fbaa69
commit d8a0020
Show file tree

Hide file tree

Showing 11 changed files with 109 additions and 93 deletions.
diff --git a/cdp-agentkit-core/CHANGELOG.md b/cdp-agentkit-core/CHANGELOG.md
@@ -4,6 +4,7 @@
 
 - Added `wow_buy_token` and `wow_sell_token`.
 - Added `token_uri` to `wow_create_token` action for custom token metadata.
+- Refactor twitter actions to conform to extendable `twitter-langchain` updates.
 
 ## [0.0.3] - 2024-11-09
 

diff --git a/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/__init__.py b/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/__init__.py
@@ -1,14 +1,21 @@
-from cdp_agentkit_core.actions.social.twitter.account_details import (
-    ACCOUNT_DETAILS_PROMPT as ACCOUNT_DETAILS_PROMPT,
-)
-from cdp_agentkit_core.actions.social.twitter.account_details import (
-    AccountDetailsInput as AccountDetailsInput,
-)
-from cdp_agentkit_core.actions.social.twitter.account_details import (
-    account_details as account_details,
-)
-from cdp_agentkit_core.actions.social.twitter.post_tweet import (
-    POST_TWEET_PROMPT as POST_TWEET_PROMPT,
-)
-from cdp_agentkit_core.actions.social.twitter.post_tweet import PostTweetInput as PostTweetInput
-from cdp_agentkit_core.actions.social.twitter.post_tweet import post_tweet as post_tweet
+from cdp_agentkit_core.actions.social.twitter.account_details import AccountDetailsAction
+from cdp_agentkit_core.actions.social.twitter.action import TwitterAction
+from cdp_agentkit_core.actions.social.twitter.post_tweet import PostTweetAction
+
+
+def get_all_twitter_actions() -> list[type[TwitterAction]]:
+    actions = []
+    for action in TwitterAction.__subclasses__():
+        actions.append(action())
+
+    return actions
+
+
+TWITTER_ACTIONS = get_all_twitter_actions()
+
+__all__ = [
+    "TwitterAction",
+    "AccountDetailsAction",
+    "PostTweetAction",
+    "TWITTER_ACTIONS",
+]
diff --git a/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/account_details.py b/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/account_details.py
@@ -1,5 +1,9 @@
+from collections.abc import Callable
+
 import tweepy
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
+
+from cdp_agentkit_core.actions.social.twitter.action import TwitterAction
 
 ACCOUNT_DETAILS_PROMPT = """
 This tool will return account details for the currently authenticated Twitter (X) user context."""
@@ -8,15 +12,13 @@
 class AccountDetailsInput(BaseModel):
     """Input argument schema for Twitter account details action."""
 
-    no_input: str = Field(
-        "",
-        description="No input required, e.g. `` (empty string).",
-    )
-
 
 def account_details(client: tweepy.Client) -> str:
     """Get the authenticated Twitter (X) user account details.
 
+    Args:
+        client (tweepy.Client): The Twitter (X) client used to authenticate with.
+
     Returns:
         str: A message containing account details for the authenticated user context.
 
@@ -36,3 +38,12 @@ def account_details(client: tweepy.Client) -> str:
         message = f"Error retrieving authenticated user account details: {e}"
 
     return message
+
+
+class AccountDetailsAction(TwitterAction):
+    """Twitter (X) account details action."""
+
+    name: str = "account_details"
+    description: str = ACCOUNT_DETAILS_PROMPT
+    args_schema: type[BaseModel] | None = AccountDetailsInput
+    func: Callable[..., str] = account_details
diff --git a/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/action.py b/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/action.py
@@ -0,0 +1,12 @@
+from collections.abc import Callable
+
+from pydantic import BaseModel
+
+
+class TwitterAction(BaseModel):
+    """Twitter Action Base Class."""
+
+    name: str
+    description: str
+    args_schema: type[BaseModel] | None = None
+    func: Callable[..., str]
diff --git a/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/post_tweet.py b/cdp-agentkit-core/cdp_agentkit_core/actions/social/twitter/post_tweet.py
@@ -1,6 +1,10 @@
+from collections.abc import Callable
+
 import tweepy
 from pydantic import BaseModel, Field
 
+from cdp_agentkit_core.actions.social.twitter import TwitterAction
+
 POST_TWEET_PROMPT = """
 This tool will post a tweet on Twitter. The tool takes the text of the tweet as input. Tweets can be maximum 280 characters."""
 
@@ -34,3 +38,12 @@ def post_tweet(client: tweepy.Client, tweet: str) -> str:
         message = f"Error posting to Twitter:\n{tweet}\n{e}"
 
     return message
+
+
+class PostTweetAction(TwitterAction):
+    """Twitter (X) post tweet action."""
+
+    name: str = "post_tweet"
+    description: str = POST_TWEET_PROMPT
+    args_schema: type[BaseModel] | None = PostTweetInput
+    func: Callable[..., str] = post_tweet
diff --git a/twitter-langchain/README.md b/twitter-langchain/README.md
@@ -21,11 +21,11 @@ pip install twitter-langchain
 Set the following environment variables:
 
 ```bash
-OPENAI_API_KEY=<your-openai-api-key>
-TWITTER_API_KEY=<your-api-key>
-TWITTER_API_SECRET=<your-api-secret>
-TWITTER_ACCESS_TOKEN=<your-access-token>
-TWITTER_ACCESS_TOKEN_SECRET=<your-access-token-secret>
+export OPENAI_API_KEY=<your-openai-api-key>
+export TWITTER_API_KEY=<your-api-key>
+export TWITTER_API_SECRET=<your-api-secret>
+export TWITTER_ACCESS_TOKEN=<your-access-token>
+export TWITTER_ACCESS_TOKEN_SECRET=<your-access-token-secret>
 ```
 
 ## Usage

diff --git a/twitter-langchain/examples/account_details/account_details.py b/twitter-langchain/examples/account_details/account_details.py
@@ -1,4 +1,3 @@
-
 from langchain_core.messages import HumanMessage
 from langchain_openai import ChatOpenAI
 from langgraph.prebuilt import create_react_agent

diff --git a/twitter-langchain/twitter_langchain/__init__.py b/twitter-langchain/twitter_langchain/__init__.py
@@ -1,11 +1,11 @@
 """CDP Twitter Toolkit."""
 
-from twitter_langchain.twitter_action import TwitterAction
 from twitter_langchain.twitter_api_wrapper import TwitterApiWrapper
+from twitter_langchain.twitter_tool import TwitterTool
 from twitter_langchain.twitter_toolkit import TwitterToolkit
 
 __all__ = [
-    "TwitterAction",
     "TwitterApiWrapper",
+    "TwitterTool",
     "TwitterToolkit",
 ]
diff --git a/twitter-langchain/twitter_langchain/twitter_api_wrapper.py b/twitter-langchain/twitter_langchain/twitter_api_wrapper.py
@@ -1,11 +1,10 @@
 """Util that calls Twitter API."""
 
+import inspect
+from collections.abc import Callable
 from typing import Any
 
-from cdp_agentkit_core.actions.social.twitter import (
-    account_details,
-    post_tweet,
-)
+import tweepy
 from langchain_core.utils import get_from_dict_or_env
 from pydantic import BaseModel, model_validator
 
@@ -22,13 +21,17 @@ def validate_environment(cls, values: dict) -> Any:
         api_key = get_from_dict_or_env(values, "twitter_api_key", "TWITTER_API_KEY")
         api_secret = get_from_dict_or_env(values, "twitter_api_secret", "TWITTER_API_SECRET")
         access_token = get_from_dict_or_env(values, "twitter_access_token", "TWITTER_ACCESS_TOKEN")
-        access_token_secret = get_from_dict_or_env(values, "twitter_access_token_secret", "TWITTER_ACCESS_TOKEN_SECRET")
+        access_token_secret = get_from_dict_or_env(
+            values, "twitter_access_token_secret", "TWITTER_ACCESS_TOKEN_SECRET"
+        )
 
         try:
             import tweepy
         except Exception:
             raise ImportError(
-                "Tweepy Twitter SDK is not installed. " "Please install it with `pip install tweepy`"
+                "Tweepy Twitter SDK is not installed. "
+
+"Please install it with `pip install tweepy`"
             ) from None
 
         client = tweepy.Client(
@@ -46,34 +49,12 @@ def validate_environment(cls, values: dict) -> Any:
 
         return values
 
-    def account_details_wrapper(self) -> str:
-        """Get the authenticated Twitter (X) user account details.
-
-        Returns:
-            str: A message containing account details for the authenticated user context in JSON format.
-
-        """
-        return account_details(client=self.client)
-
-    def post_tweet_wrapper(self, tweet: str) -> str:
-        """Post tweet to Twitter.
+    def run_action(self, func: Callable[..., str], **kwargs) -> str:
+        """Run a Twitter Action."""
+        func_signature = inspect.signature(func)
+        first_kwarg = next(iter(func_signature.parameters.values()), None)
 
-        Args:
-            client (tweepy.Client): The tweepy client to use.
-            tweet (str): The text of the tweet to post to twitter. Tweets can be maximum 280 characters.
-
-        Returns:
-            str: A message containing the result of the post action and the tweet.
-
-        """
-        return post_tweet(client=self.client, tweet=tweet)
-
-    def run(self, mode: str, **kwargs) -> str:
-        """Run the action via the Twitter API."""
-        if mode == "account_details":
-            return self.account_details_wrapper()
-        elif mode == "post_tweet":
-            return self.post_tweet_wrapper(**kwargs)
+        if first_kwarg and first_kwarg.annotation is tweepy.Client:
+            return func(self.client, **kwargs)
         else:
-            raise ValueError("Invalid mode: " + mode)
-
+            return func(**kwargs)
diff --git a/...chain/twitter_langchain/twitter_action.py → ...ngchain/twitter_langchain/twitter_tool.py b/...chain/twitter_langchain/twitter_action.py → ...ngchain/twitter_langchain/twitter_tool.py
@@ -1,5 +1,15 @@
-"""Tool allows agents to interact with the Twitter API."""
+"""Tool allows agents to interact with the Twitter API.
 
+To use this tool, you must first set as environment variables:
+    OPENAI_API_KEY
+    TWITTER_API_KEY
+    TWITTER_API_SECRET
+    TWITTER_ACCESS_TOKEN
+    TWITTER_ACCESS_TOKEN_SECRET
+
+"""
+
+from collections.abc import Callable
 from typing import Any
 
 from langchain_core.callbacks import CallbackManagerForToolRun
@@ -9,14 +19,14 @@
 from twitter_langchain.twitter_api_wrapper import TwitterApiWrapper
 
 
-class TwitterAction(BaseTool):  # type: ignore[override]
+class TwitterTool(BaseTool):  # type: ignore[override]
     """Tool for interacting with the Twitter API."""
 
     twitter_api_wrapper: TwitterApiWrapper
-    mode: str
     name: str = ""
     description: str = ""
     args_schema: type[BaseModel] | None = None
+    func: Callable[..., str]
 
     def _run(
         self,
@@ -33,4 +43,4 @@ def _run(
             parsed_input_args = validated_input_data.model_dump()
         else:
             parsed_input_args = {"instructions": instructions}
-        return self.twitter_api_wrapper.run(self.mode, **parsed_input_args)
+        return self.twitter_api_wrapper.run_action(self.func, **parsed_input_args)
diff --git a/twitter-langchain/twitter_langchain/twitter_toolkit.py b/twitter-langchain/twitter_langchain/twitter_toolkit.py
@@ -1,16 +1,11 @@
 """TwitterToolkit."""
 
-from cdp_agentkit_core.actions.social.twitter import (
-    ACCOUNT_DETAILS_PROMPT,
-    POST_TWEET_PROMPT,
-    AccountDetailsInput,
-    PostTweetInput,
-)
+from cdp_agentkit_core.actions.social.twitter import TWITTER_ACTIONS
 from langchain_core.tools import BaseTool
 from langchain_core.tools.base import BaseToolkit
 
-from twitter_langchain.twitter_action import TwitterAction
 from twitter_langchain.twitter_api_wrapper import TwitterApiWrapper
+from twitter_langchain.twitter_tool import TwitterTool
 
 
 class TwitterToolkit(BaseToolkit):
@@ -122,28 +117,15 @@ def from_twitter_api_wrapper(cls, twitter_api_wrapper: TwitterApiWrapper) -> "Tw
             TwitterToolkit. The Twitter toolkit.
 
         """
-        actions: list[dict] = [
-            {
-                "mode": "account_details",
-                "name": "account_details",
-                "description": ACCOUNT_DETAILS_PROMPT,
-                "args_schema": AccountDetailsInput,
-            },
-            {
-                "mode": "post_tweet",
-                "name": "post_tweet",
-                "description": POST_TWEET_PROMPT,
-                "args_schema": PostTweetInput,
-            },
-        ]
+        actions = TWITTER_ACTIONS
 
         tools = [
-            TwitterAction(
-                name=action["name"],
-                description=action["description"],
-                mode=action["mode"],
+            TwitterTool(
+                name=action.name,
+                description=action.description,
                 twitter_api_wrapper=twitter_api_wrapper,
-                args_schema=action.get("args_schema", None),
+                args_schema=action.args_schema,
+                func=action.func,
             )
             for action in actions
         ]