# X.com Integration

> Build apps that interact with X.com (Twitter).

**URL:** https://inference.sh/docs/examples/x-integration
**Last updated:** 2026-09-24

---

Build apps that interact with X.com (Twitter).

---

## Overview

This guide shows you how to build an app that uses X.com credentials to post tweets, read timelines, and interact with the X API.

---

## Prerequisites

Users running your app need:

1. An X Developer account with OAuth 2.0 configured
2. `X_CLIENT_ID` and `X_CLIENT_SECRET` secrets configured in Settings
3. X.com credential connected (OAuth authorized)

→ See [X.com Setup](/docs/credentials/x) for user setup instructions

---

## App Configuration

### `inf.yaml`

Declare the X.com capabilities your app needs:

```yaml
namespace: myteam
name: my-x-app
description: Posts tweets using AI
category: social
kernel: python-3.11

credentials:
  - key: x.tweet.write
    description: Post tweets to X.com

  - key: x.tweet.read
    description: Read tweet data
    optional: true

  - key: x.dm.write
    description: Send direct messages
    optional: true

resources:
  gpu:
    count: 0
    type: none
  ram: 2
```

### Available Capabilities

| Capability | Description | Scopes Granted |
|------------|-------------|----------------|
| `x.tweet.read` | Read tweets | `tweet.read`, `users.read` |
| `x.tweet.write` | Post tweets | `tweet.read`, `tweet.write`, `users.read` |
| `x.user.read` | Read user profiles | `users.read` |
| `x.follow.read` | Read followers/following | `follows.read`, `users.read` |
| `x.follow.write` | Follow/unfollow users | `follows.write`, `users.read` |
| `x.dm.read` | Read direct messages | `dm.read`, `users.read` |
| `x.dm.write` | Send direct messages | `dm.write`, `users.read` |
| `x.like.read` | Read liked tweets | `like.read`, `users.read` |
| `x.like.write` | Like/unlike tweets | `like.write`, `users.read` |

---

## Environment Variables

When your app runs, these environment variables are injected:

| Variable | Description |
|----------|-------------|
| `X_ACCESS_TOKEN` | OAuth 2.0 access token (auto-refreshed) |
| `X_ACCESS_TOKEN_EXPIRES_AT` | Token expiry time (RFC3339 format) |

---

## App Code

### `inference.py`

```python
import os
import tweepy
from inferencesh import BaseApp, BaseAppInput, BaseAppOutput
from pydantic import Field

class AppInput(BaseAppInput):
    text: str = Field(description="Tweet text to post")

class AppOutput(BaseAppOutput):
    tweet_id: str = Field(description="ID of the posted tweet")
    tweet_url: str = Field(description="URL of the posted tweet")

class App(BaseApp):
    client: tweepy.Client = None

    async def setup(self, metadata):
        """Initialize the X.com client with OAuth 2.0 token."""
        access_token = os.environ.get("X_ACCESS_TOKEN")
        
        if not access_token:
            raise ValueError("X_ACCESS_TOKEN not found. Is the X.com credential connected?")
        
        # Create client with OAuth 2.0 user context
        self.client = tweepy.Client(bearer_token=access_token)
        
        metadata.log("X.com client initialized")

    async def run(self, input_data: AppInput, metadata) -> AppOutput:
        """Post a tweet."""
        metadata.log(f"Posting tweet: {input_data.text[:50]}...")
        
        # Post the tweet
        response = self.client.create_tweet(text=input_data.text)
        
        tweet_id = response.data["id"]
        tweet_url = f"https://x.com/i/web/status/{tweet_id}"
        
        metadata.log(f"Tweet posted: {tweet_url}")
        
        return AppOutput(
            tweet_id=tweet_id,
            tweet_url=tweet_url
        )

    async def unload(self):
        """Cleanup."""
        self.client = None
```

### `requirements.txt`

```
tweepy>=4.14.0
```

---

## Examples

### Post a Tweet

```python
# Simple tweet
response = self.client.create_tweet(text="Hello from my AI agent! 🤖")

# Tweet with media (requires media upload first)
media = self.client.media_upload("image.png")
response = self.client.create_tweet(
    text="Check out this image!",
    media_ids=[media.media_id]
)

# Reply to a tweet
response = self.client.create_tweet(
    text="Great point!",
    in_reply_to_tweet_id="1234567890"
)

# Quote tweet
response = self.client.create_tweet(
    text="This is so true 👆",
    quote_tweet_id="1234567890"
)
```

### Read Tweets

```python
# Get a specific tweet
tweet = self.client.get_tweet(
    id="1234567890",
    tweet_fields=["created_at", "public_metrics"]
)

# Get user's recent tweets
tweets = self.client.get_users_tweets(
    id="user_id",
    max_results=10,
    tweet_fields=["created_at", "public_metrics"]
)

# Search tweets
tweets = self.client.search_recent_tweets(
    query="AI agents",
    max_results=10
)
```

### User Operations

```python
# Get user info
user = self.client.get_user(username="elonmusk")

# Get followers
followers = self.client.get_users_followers(
    id="user_id",
    max_results=100
)

# Follow a user (requires x.follow.write)
self.client.follow_user(target_user_id="user_id")
```

### Direct Messages

```python
# Send a DM (requires x.dm.write)
self.client.create_direct_message(
    participant_id="user_id",
    text="Hello!"
)

# Get DM conversations (requires x.dm.read)
conversations = self.client.get_direct_message_events()
```

---

## Error Handling

```python
import tweepy

async def run(self, input_data: AppInput, metadata) -> AppOutput:
    try:
        response = self.client.create_tweet(text=input_data.text)
        return AppOutput(tweet_id=response.data["id"], ...)
        
    except tweepy.TweepyException as e:
        if "duplicate" in str(e).lower():
            raise ValueError("This tweet was already posted")
        elif "rate limit" in str(e).lower():
            raise ValueError("Rate limit exceeded. Try again later.")
        else:
            raise ValueError(f"X.com API error: {e}")
```

---

## Rate Limits

X.com has strict rate limits. Common limits for the free tier:

| Endpoint | Limit |
|----------|-------|
| Post tweet | 50 per 24 hours |
| Read tweets | 100 per 15 minutes |
| User lookup | 100 per 15 minutes |

Your app should handle rate limit errors gracefully.

---

## Testing Locally

1. Connect X.com in **vaults → credentials**
2. Export the token for local testing:

```bash
# Get your token from the X.com credential (one-time for testing)
export X_ACCESS_TOKEN="your-token-here"

# Test locally
belt app test --input input.json
```

---

## Deploy

```bash
belt app deploy
```

When users run your app:

1. If they haven't configured X.com → prompted to set up
2. If configured → app runs with their X.com credentials
3. Tokens auto-refresh → no manual token management needed

---

## Next Steps

- [X.com Setup](/docs/credentials/x) — User setup guide
- [Extending Apps](/docs/extend/extending-apps) — Full app development guide
- [API & SDK](/docs/api/overview) — Run your app programmatically