How to Integrate Twitter API v2 into HoshinoBot: Complete Setup Guide

To integrate Twitter API v2 into HoshinoBot, you configure API credentials in a dedicated config file, define Service objects that map to Twitter accounts, and deploy an auto-restarting daemon that processes the filtered stream and broadcasts formatted tweets to subscribed groups.

HoshinoBot implements Twitter API v2 support through a self-contained stream module located in hoshino/modules/twitter-v2/stream. This integration uses the filtered stream endpoint to monitor specific accounts in real-time, automatically formats incoming tweets with timestamps and media, and routes them to group chats based on service subscriptions.

Configuration Setup

Before starting the daemon, you must create a configuration file with your Twitter API credentials and follow lists.

Creating the Credentials File

Copy the example configuration from hoshino/config_example/twitter.py to hoshino/config/twitter.py and populate it with your keys from the Twitter Developer Portal.

# hoshino/config/twitter.py

consumer_key = "YOUR_CONSUMER_KEY"
consumer_secret = "YOUR_CONSUMER_SECRET"
access_token_key = "YOUR_ACCESS_TOKEN"
access_token_secret = "YOUR_ACCESS_TOKEN_SECRET"
proxy = None   # Use "http://127.0.0.1:1080" if behind a firewall

follows = {
    "pcr-twitter": ["priconne_redive", "priconne_anime"],
    "kc-twitter": ["KanColle_STAFF"],
}

The follows dictionary maps service names to lists of Twitter screen names. Each key must correspond to a Service object defined in the module's service collection.

Service Registration and Daemon Architecture

The integration uses HoshinoBot's Service class to create togglable features that groups can enable or disable. A background daemon manages the persistent connection to Twitter's streaming API.

Defining Service Objects

In hoshino/modules/twitter-v2/stream/follow.py, the service_collection list declares which tweet streams are available to groups. Each Service name must match a key in your follows configuration.

# hoshino/modules/twitter-v2/stream/follow.py

from hoshino import Service

service_collection = [
    Service("pcr-twitter", help_="日服Twitter转发", enable_on_default=True, bundle="pcr订阅"),
    Service("kc-twitter", help_="舰娘推特转发", enable_on_default=False, bundle="kancolle"),
]

If a service name exists in service_collection but lacks a corresponding entry in follows, that service will initialize but never receive tweets.

Daemon Lifecycle and Auto-Recovery

The daemon starts automatically when the bot boots via the @bot.on_startup decorator in hoshino/modules/twitter-v2/stream/__init__.py. The start_daemon() function creates an asyncio task that runs stream_daemon(), which wraps the main follow_stream() function.

# hoshino/modules/twitter-v2/stream/__init__.py

@bot.on_startup
async def start_daemon():
    global daemon
    loop = asyncio.get_event_loop()
    daemon = loop.create_task(stream_daemon(follow_stream))

async def stream_daemon(stream_func):
    while True:
        try:
            await stream_func()
        except (KeyboardInterrupt, asyncio.CancelledError):
            sv.logger.info("Twitter stream daemon exited.")
            raise
        except Exception as e:
            sv.logger.exception(e)
            sv.logger.error(f"Error {type(e)} Occurred in twitter stream. Restarting stream in 1 min.")
            await asyncio.sleep(60)

This architecture guarantees high availability: if the stream encounters a network error or rate limit, the daemon logs the exception, sleeps for 60 seconds, and restarts the connection automatically.

Stream Processing and Rule Management

The follow_stream() function in follow.py handles the complex logic of converting your follow lists into Twitter API rules and processing the incoming JSON stream.

Building Filtered Stream Rules

Twitter API v2 allows a maximum of five filtered stream rules, each with a 512-byte limit. The code first aggregates all screen names from the follows configuration, deletes existing rules, then creates new ones using the cut_list() helper to chunk usernames appropriately.

# hoshino/modules/twitter-v2/stream/follow.py

follow_names = set()
for s in service_collection:
    follow_names.update(cfg.follows.get(s.name, []))

# Delete old rules first

old_rules = await client.api.tweets.search.stream.rules.get()
if old_rules.data:
    ids = [str(r.id) for r in old_rules.data]
    await client.api.tweets.search.stream.rules.post(_json={"delete": {"ids": ids}})

# Create new chunked rules

follow_name_chunks = cut_list(follow_names)
rules = [{"value": f'from:{" OR from:".join(chunk)}'} for chunk in follow_name_chunks]
await client.api.tweets.search.stream.rules.post(_json={"add": rules})

Tweet Routing and Broadcasting

The TweetRouter class maintains a mapping from screen names to FollowEntry objects, which track which services subscribe to each account and whether to filter for media-only or forward retweets.

# hoshino/modules/twitter-v2/stream/follow.py

class TweetRouter:
    def __init__(self):
        self.follows: Dict[str, FollowEntry] = defaultdict(FollowEntry)

    def add(self, service: Service, follow_names: Iterable[str]):
        for name in follow_names:
            self.follows[name].services.add(service)

When a tweet arrives, the router checks the username against this mapping, applies user-level filters, formats the content, and broadcasts to all subscribing services:

# Inside follow_stream() loop

msg = await format_tweet(tweet, client)
for s in entry.services:
    asyncio.get_event_loop().create_task(s.broadcast(msg, f" @{username} 推文", 0.2))

Tweet Formatting and Media Handling

Raw Twitter JSON is transformed into human-readable messages by format_tweet() in hoshino/modules/twitter-v2/stream/util.py. This function localizes timestamps to Asia/Shanghai format, extracts media URLs as image segments, and recursively processes retweets and quote tweets up to MAX_DEPTH = 1.

# hoshino/modules/twitter-v2/stream/util.py

def format_tweet(tweet, client):
    data = tweet.get("data")
    name = data["author_id"]
    time = format_time(data["created_at"])
    msg = f"@{name}\n{time}\n\n{data['text']}"
    
    if "media" in tweet.get("includes", {}):
        imgs = "".join([str(MessageSegment.image(m.url)) for m in tweet["includes"]["media"]])
        msg = f"{msg}\n{imgs}"
    
    # Handles retweets and quotes recursively...

    return msg

The cut_list() utility ensures compliance with Twitter's 512-byte rule limit by grouping usernames into optimally sized chunks.

Runtime Administration

Administrators can reload the Twitter stream daemon without restarting the entire bot using the reload-twitter-stream-daemon command (aliases: 重启转推, 重载转推).

# hoshino/modules/twitter-v2/stream/__init__.py

@sucmd("reload-twitter-stream-daemon", force_private=False, aliases=("重启转推", "重载转推"))
async def reload_twitter_stream_daemon(session):
    daemon.cancel()
    importlib.reload(cfg)
    await start_daemon()
    await session.send("ok")

This command cancels the existing asyncio task, reloads the configuration module to pick up new credentials or follow lists, and initializes a fresh daemon instance.

Summary

• Twitter API v2 integration in HoshinoBot is implemented as a stream module using the filtered stream endpoint.
• Configuration requires creating hoshino/config/twitter.py with credentials and a follows dictionary mapping service names to screen names.
• Service registration links configuration keys to Service objects in service_collection inside follow.py.
• Daemon architecture provides auto-restarting capabilities through stream_daemon() in __init__.py, ensuring 60-second recovery from network failures.
• Rule management respects Twitter's 5-rule, 512-byte limit using the cut_list() helper to chunk usernames.
• Tweet routing uses the TweetRouter class to broadcast formatted messages with media support to subscribed group chats.

Frequently Asked Questions

How do I obtain Twitter API v2 credentials for HoshinoBot?

You must apply for a Twitter Developer account and create a project with "Essential" or "Elevated" access in the Twitter Developer Portal. Generate Consumer Key, Consumer Secret, Access Token, and Access Token Secret from the "Keys and Tokens" section, then paste these values into hoshino/config/twitter.py.

Why are my tweets not appearing in the group chat?

First, verify the service name in service_collection exactly matches the key in your follows dictionary. Second, ensure the group has enabled the service using the appropriate enable command (e.g., enable pcr-twitter). Third, check that the Twitter screen names in your configuration are valid and the accounts are actually posting content that matches your filtered stream rules.

How does the bot handle Twitter API rate limits or disconnections?

The stream_daemon() function in hoshino/modules/twitter-v2/stream/__init__.py wraps the stream in an infinite loop with exception handling. When any error occurs—whether a network timeout, rate limit, or API disruption—the daemon logs the error, waits 60 seconds, and automatically restarts the connection without requiring manual intervention.

Can I filter tweets to only show media or ignore retweets?

Yes. The FollowEntry class supports flags like media_only and forward_retweet that you can configure when setting up the router. When processing tweets, the code checks these flags before broadcasting: if media_only is set and the tweet lacks media attachments, or if forward_retweet is false and the tweet is a retweet, the message is suppressed for that specific service.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}