feat: Scaffolding for Twitter Langchain Toolkit

twitter-langchain/README.md

@@ -0,0 +1 @@
+# Twitter (X) Langchain Toolkit

twitter-langchain/pyproject.toml

@@ -0,0 +1,49 @@
+[tool.poetry]
+name = "twitter-langchain"
+version = "0.0.1"
+description = "Twitter Langchain Toolkit"
+authors = ["John Peterson <[email protected]>"]
+license = "Apache-2.0"
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.10"
+pydantic = "^2.9.2"
+langchain = "^0.3.7"
+tweepy = "^4.14.0"
+
+
+[tool.poetry.group.dev.dependencies]
+ruff = "^0.7.2"
+python-lsp-server = "^1.12.0"
+ruff-lsp = "^0.0.58"
+pytest = "^8.3.3"
+pytest-cov = "^6.0.0"
+mypy = "^1.13.0"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.mypy]
+python_version = "3.10"
+strict = true
+ignore_missing_imports = true
+
+[tool.ruff]
+line-length = 100
+target-version = "py310"
+exclude = ["./build/**", "./dist/**", "./docs/**"]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "W", "D", "UP", "B", "C4", "SIM", "RUF"]
+ignore = ["D213", "D203", "D100", "D104", "D107", "E501"]
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"
+skip-magic-trailing-comma = false
+line-ending = "auto"
+
+[tool.ruff.lint.isort]
+known-first-party = ["cdp_agentkit_core", "twitter_langchain"]

twitter-langchain/twitter_langchain/twitter_action.py

@@ -0,0 +1,41 @@
+"""Tool allows agents to interact with the Twitter API.
+
+To use this tool, you must first set as environment variables:
+    # TODO: List ENV VARs required.
+
+"""
+
+from typing import Any
+
+from langchain_core.callbacks import CallbackManagerForToolRun
+from langchain_core.tools import BaseTool
+from pydantic import BaseModel
+
+from twitter_langchain.twitter_api_wrapper import TwitterApiWrapper
+
+
+class TwitterAction(BaseTool):  # type: ignore[override]
+    """Tool for interacting with the Twitter API."""
+
+    twitter_apiwrapper: TwitterApiWrapper
+    mode: str
+    name: str = ""
+    description: str = ""
+    args_schema: type[BaseModel] | None = None
+
+    def _run(
+        self,
+        instructions: str | None = "",
+        run_manager: CallbackManagerForToolRun | None = None,
+        **kwargs: Any,
+    ) -> str:
+        """Use the Twitter (X) API to run an operation."""
+        if not instructions or instructions == "{}":
+            # Catch other forms of empty input that GPT-4 likes to send.
+            instructions = ""
+        if self.args_schema is not None:
+            validated_input_data = self.args_schema(**kwargs)
+            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)

twitter-langchain/twitter_langchain/twitter_api_wrapper.py

@@ -0,0 +1,19 @@
+"""Util that calls Twitter API."""
+
+from typing import Any
+
+from pydantic import BaseModel, model_validator
+
+
+class TwitterApiWrapper(BaseModel):
+    """Wrapper for Twitter API."""
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_environment(cls, values: dict) -> Any:
+        """TODO: Implement."""
+        pass
+
+    def run(self, mode: str, **kwargs) -> str:
+        """Run the action via the Twitter API."""
+        raise ValueError("Invalid mode" + mode)

twitter-langchain/twitter_langchain/twitter_toolkit.py

@@ -0,0 +1,129 @@
+"""TwitterToolkit."""
+
+from langchain_core.tools import BaseTool
+from langchain_core.tools.base import BaseToolkit
+from twitterlangchain import TwitterApiWrapper
+
+from twitter_langchain import TwitterAction
+
+
+class TwitterToolkit(BaseToolkit):
+    """Twitter (X) Toolkit.
+
+    *Security Note*: This toolkit contains tools that can read and modify
+        the state of a service; e.g., by creating, deleting, or updating,
+        reading underlying data.
+
+        For example, this toolkit can be used post messages on Twitter (X).
+
+        See [Security](https://python.langchain.com/docs/security) for more information.
+
+    Setup:
+        See detailed installation instructions here:
+        https://python.langchain.com/docs/integrations/tools/twitter/#installation
+
+        You will need to set the following environment
+        variables:
+
+        .. code-block:: bash
+
+            # TODO: Add relevant ENV VARs
+
+    Instantiate:
+        .. code-block:: python
+
+            from twitter_langchain import TwitterToolkit
+            from twitter_langchain import TwitterAgentkitWrapper
+
+            twitter = TwitterAgentkitWrapper()
+            twitter_toolkit = TwitterToolkit.from_twitter_api_wrapper(twitter)
+
+    Tools:
+        .. code-block:: python
+
+            tools = twitter_toolkit.get_tools()
+            for tool in tools:
+                print(tool.name)
+
+        .. code-block:: none
+
+            # TODO: Add list of available tools.
+
+    Use within an agent:
+        .. code-block:: python
+
+            from langchain_openai import ChatOpenAI
+            from langgraph.prebuilt import create_react_agent
+
+            # Select example tool
+            tools = [tool for tool in toolkit.get_tools() if tool.name == "post_tweet"]
+            assert len(tools) == 1
+
+            llm = ChatOpenAI(model="gpt-4o-mini")
+            agent_executor = create_react_agent(llm, tools)
+
+            example_query = "Post a hello tweet to the world"
+
+            events = agent_executor.stream(
+                {"messages": [("user", example_query)]},
+                stream_mode="values",
+            )
+            for event in events:
+                event["messages"][-1].pretty_print()
+
+        .. code-block:: none
+
+             ================================[1m Human Message [0m=================================
+
+            Post a hello tweet to the world
+            ==================================[1m Ai Message [0m==================================
+            Tool Calls:
+            post_tweet (call_iSYJVaM7uchfNHOMJoVPQsOi)
+            Call ID: call_iSYJVaM7uchfNHOMJoVPQsOi
+            Args:
+                no_input: "hello world"
+            =================================[1m Tool Message [0m=================================
+            Name: post_tweet
+
+            ...
+            ==================================[1m Ai Message [0m==================================
+
+            I posted the tweet "hello world".
+
+    Parameters
+    ----------
+        tools: List[BaseTool]. The tools in the toolkit. Default is an empty list.
+
+    """
+
+    tools: list[BaseTool] = []  # noqa: RUF012
+
+    @classmethod
+    def from_twitter_api_wrapper(cls, twitter_api_wrapper: TwitterApiWrapper) -> "TwitterToolkit":
+        """Create a TwitterToolkit from a TwitterApiWrapper.
+
+        Args:
+            twitter_api_wrapper: TwitterApiWrapper. The Twitter (X) API wrapper.
+
+        Returns:
+            TwitterToolkit. The Twitter toolkit.
+
+        """
+        actions: list[dict] = []
+
+        tools = [
+            TwitterAction(
+                name=action["name"],
+                description=action["description"],
+                mode=action["mode"],
+                twitter_api_wrapper=twitter_api_wrapper,
+                args_schema=action.get("args_schema", None),
+            )
+            for action in actions
+        ]
+
+        return cls(tools=tools)  # type: ignore[arg-type]
+
+    def get_tools(self) -> list[BaseTool]:
+        """Get the tools in the toolkit."""
+        return self.tools