grok-install

# reply_bot.py — wire xAI SDK by hand
import os, time, json, logging
from xai_sdk import Client
from xai_sdk.chat import user, system, tool
from tweepy import Client as XClient
from tenacity import retry, stop_after_attempt
from ratelimit import limits, sleep_and_retry

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("reply-bot")

# --- env wiring ---
GROK_KEY = os.environ["GROK_API_KEY"]
X_BEARER = os.environ["X_BEARER_TOKEN"]
if not GROK_KEY or not X_BEARER:
    raise RuntimeError("missing keys")

grok = Client(api_key=GROK_KEY)
x    = XClient(bearer_token=X_BEARER)

# --- declare the tool by hand ---
REPLY_TOOL = {
    "type": "function",
    "function": {
        "name": "reply_to_mention",
        "description": "Reply to an X mention",
        "parameters": {
            "type": "object",
            "properties": {
                "mention_id": {"type": "string"},
                "reply_text": {"type": "string", "maxLength": 280},
            },
            "required": ["mention_id", "reply_text"],
        },
    },
}

# --- rate limit by hand ---
@sleep_and_retry
@limits(calls=200, period=86400)
@retry(stop=stop_after_attempt(3))
def post_reply(mention_id, text):
    return x.create_tweet(text=text, in_reply_to_tweet_id=mention_id)

# --- cost guard by hand ---
SPENT_TODAY = 0.0
DAILY_USD   = 3.0

def guard(usage):
    global SPENT_TODAY
    cost = usage.prompt_tokens * 3e-6 + usage.completion_tokens * 15e-6
    SPENT_TODAY += cost
    if SPENT_TODAY > DAILY_USD:
        raise RuntimeError("daily cost cap exceeded")

# --- safety scan by hand (just the start...) ---
BLOCKED = ["deepfake", "mass dm", "spam"]
def safety_check(text):
    low = text.lower()
    return not any(b in low for b in BLOCKED)

# --- main loop ---
def main():
    while True:
        mentions = x.get_users_mentions(id="me").data or []
        for m in mentions:
            chat = grok.chat.create(model="grok-4", tools=[REPLY_TOOL])
            chat.append(system("You reply to X mentions."))
            chat.append(user(m.text))
            r = chat.sample()
            guard(r.usage)
            for tc in r.tool_calls or []:
                args = json.loads(tc.function.arguments)
                if safety_check(args["reply_text"]):
                    post_reply(args["mention_id"], args["reply_text"])
        time.sleep(60)

if __name__ == "__main__":
    main()

# ...and you still need: deploy config, telemetry, structured
# output validation, Dockerfile, requirements.txt, env loader,
# pre-install scan, log rotation, parallel tool dispatch...


# reply_bot.py — wire xAI SDK by hand
import os, time, json, logging
from xai_sdk import Client
from xai_sdk.chat import user, system, tool
from tweepy import Client as XClient
from tenacity import retry, stop_after_attempt
from ratelimit import limits, sleep_and_retry