X.com Integration

1namespace: myteam2name: my-x-app3description: Posts tweets using AI4category: social5kernel: python-3.1167credentials:8  - key: x.tweet.write9    description: Post tweets to X.com1011  - key: x.tweet.read12    description: Read tweet data13    optional: true1415  - key: x.dm.write16    description: Send direct messages17    optional: true1819resources:20  gpu:21    count: 022    type: none23  ram: 2

1import os2import tweepy3from inferencesh import BaseApp, BaseAppInput, BaseAppOutput4from pydantic import Field56class AppInput(BaseAppInput):7    text: str = Field(description="Tweet text to post")89class AppOutput(BaseAppOutput):10    tweet_id: str = Field(description="ID of the posted tweet")11    tweet_url: str = Field(description="URL of the posted tweet")1213class App(BaseApp):14    client: tweepy.Client = None1516    async def setup(self, metadata):17        """Initialize the X.com client with OAuth 2.0 token."""18        access_token = os.environ.get("X_ACCESS_TOKEN")19        20        if not access_token:21            raise ValueError("X_ACCESS_TOKEN not found. Is the X.com credential connected?")22        23        # Create client with OAuth 2.0 user context24        self.client = tweepy.Client(bearer_token=access_token)25        26        metadata.log("X.com client initialized")2728    async def run(self, input_data: AppInput, metadata) -> AppOutput:29        """Post a tweet."""30        metadata.log(f"Posting tweet: {input_data.text[:50]}...")31        32        # Post the tweet33        response = self.client.create_tweet(text=input_data.text)34        35        tweet_id = response.data["id"]36        tweet_url = f"https://x.com/i/web/status/{tweet_id}"37        38        metadata.log(f"Tweet posted: {tweet_url}")39        40        return AppOutput(41            tweet_id=tweet_id,42            tweet_url=tweet_url43        )4445    async def unload(self):46        """Cleanup."""47        self.client = None

1tweepy>=4.14.0

1# Simple tweet2response = self.client.create_tweet(text="Hello from my AI agent! 🤖")34# Tweet with media (requires media upload first)5media = self.client.media_upload("image.png")6response = self.client.create_tweet(7    text="Check out this image!",8    media_ids=[media.media_id]9)1011# Reply to a tweet12response = self.client.create_tweet(13    text="Great point!",14    in_reply_to_tweet_id="1234567890"15)1617# Quote tweet18response = self.client.create_tweet(19    text="This is so true 👆",20    quote_tweet_id="1234567890"21)

1# Get a specific tweet2tweet = self.client.get_tweet(3    id="1234567890",4    tweet_fields=["created_at", "public_metrics"]5)67# Get user's recent tweets8tweets = self.client.get_users_tweets(9    id="user_id",10    max_results=10,11    tweet_fields=["created_at", "public_metrics"]12)1314# Search tweets15tweets = self.client.search_recent_tweets(16    query="AI agents",17    max_results=1018)

1# Get user info2user = self.client.get_user(username="elonmusk")34# Get followers5followers = self.client.get_users_followers(6    id="user_id",7    max_results=1008)910# Follow a user (requires x.follow.write)11self.client.follow_user(target_user_id="user_id")

1# Send a DM (requires x.dm.write)2self.client.create_direct_message(3    participant_id="user_id",4    text="Hello!"5)67# Get DM conversations (requires x.dm.read)8conversations = self.client.get_direct_message_events()

1import tweepy23async def run(self, input_data: AppInput, metadata) -> AppOutput:4    try:5        response = self.client.create_tweet(text=input_data.text)6        return AppOutput(tweet_id=response.data["id"], ...)7        8    except tweepy.TweepyException as e:9        if "duplicate" in str(e).lower():10            raise ValueError("This tweet was already posted")11        elif "rate limit" in str(e).lower():12            raise ValueError("Rate limit exceeded. Try again later.")13        else:14            raise ValueError(f"X.com API error: {e}")

1# Get your token from the X.com credential (one-time for testing)2export X_ACCESS_TOKEN="your-token-here"34# Test locally5belt app test --input input.json

1belt app deploy