# AgentBody X API Reference

Base URL: `https://api.agentbody.io`. Every request uses `Authorization: Bearer $AGENTBODY_API_KEY` and documented `snake_case` query fields.

| Route | Required fields | Optional fields | Purpose |
|---|---|---|---|
| `GET /v1/twitter/search` | `query` | `cursor` | Search public posts |
| `GET /v1/twitter/trending` | none | `country` | Read current trends |
| `GET /v1/twitter/post` | `post_id` | none | Read one post |
| `GET /v1/twitter/profile` | `username` or `user_id` | none | Read one profile |
| `GET /v1/twitter/profile/posts` | `username` or `user_id` | `cursor` | Read profile posts |
| `GET /v1/twitter/profile/media` | `username` or `user_id` | `cursor` | Read profile media |
| `GET /v1/twitter/post/comments` | `post_id` | `cursor` | Read replies/comments |

The Gateway returns direct business JSON on success and `{"error":{"code":"...","message":"..."}}` on failure. Preserve source URLs, identity fields, timestamps, metrics, and cursors exactly as returned. Do not infer unsupported fields.

#!/usr/bin/env python3
"""Contract tests for the AgentBody X research client."""

import importlib.util
import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
import urllib.error


MODULE_PATH = Path(__file__).with_name("x_client.py")
SPEC = importlib.util.spec_from_file_location("x_client", MODULE_PATH)
x_client = importlib.util.module_from_spec(SPEC)
assert SPEC.loader
SPEC.loader.exec_module(x_client)


class XClientContractTests(unittest.TestCase):
    def test_loads_key_from_agentbody_credentials(self):
        with tempfile.TemporaryDirectory() as home:
            credentials = Path(home) / ".agentbody" / "credentials"
            credentials.parent.mkdir()
            credentials.write_text('AGENTBODY_API_KEY="saved-key" # note\n', encoding="utf-8")
            with patch.dict(os.environ, {}, clear=True), patch("os.path.expanduser", return_value=home):
                self.assertEqual(x_client.resolve_api_key(), "saved-key")

    def test_local_credentials_override_process_environment(self):
        with tempfile.TemporaryDirectory() as home:
            credentials = Path(home) / ".agentbody" / "credentials"
            credentials.parent.mkdir()
            credentials.write_text("AGENTBODY_API_KEY=local-key\n", encoding="utf-8")
            with patch.dict(os.environ, {"AGENTBODY_API_KEY": "agent-key"}, clear=True), patch("os.path.expanduser", return_value=home):
                self.assertEqual(x_client.resolve_api_key(), "local-key")

    def test_supported_commands_map_to_agentbody_routes_and_snake_case(self):
        client = x_client.AgentBodyXClient("key")
        captured = []
        client._request = lambda path, params: captured.append((path, params)) or {"items": []}

        client.search("AI agents", "cursor-1")
        client.trending("US")
        client.post("123")
        client.profile("OpenAI")
        client.profile_posts("OpenAI", "cursor-2")
        client.profile_media("OpenAI", "cursor-3")
        client.post_comments("123", "cursor-4")

        self.assertEqual(captured, [
            ("/v1/twitter/search", {"query": "AI agents", "cursor": "cursor-1"}),
            ("/v1/twitter/trending", {"country": "US"}),
            ("/v1/twitter/post", {"post_id": "123"}),
            ("/v1/twitter/profile", {"username": "OpenAI"}),
            ("/v1/twitter/profile/posts", {"username": "OpenAI", "cursor": "cursor-2"}),
            ("/v1/twitter/profile/media", {"username": "OpenAI", "cursor": "cursor-3"}),
            ("/v1/twitter/post/comments", {"post_id": "123", "cursor": "cursor-4"}),
        ])

    def test_account_errors_are_actionable_and_do_not_expose_raw_body(self):
        client = x_client.AgentBodyXClient("key")
        unauthorized = urllib.error.HTTPError("url", 401, "Unauthorized", {}, None)
        unauthorized.read = lambda: json.dumps({"secret": "raw"}).encode()
        with patch("urllib.request.urlopen", side_effect=unauthorized):
            result = client.search("AI")
        self.assertEqual(result["error"]["code"], "UNAUTHORIZED")
        self.assertIn("https://agentbody.io/login", result["error"]["message"])
        self.assertNotIn("raw", result["error"]["message"])

        insufficient = urllib.error.HTTPError("url", 402, "Payment Required", {}, None)
        insufficient.read = lambda: b"provider details"
        with patch("urllib.request.urlopen", side_effect=insufficient):
            result = client.search("AI")
        self.assertEqual(result["error"]["code"], "INSUFFICIENT_BALANCE")
        self.assertIn("https://agentbody.io/console/billing", result["error"]["message"])


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3
"""AgentBody X/Twitter read client for agent workflows."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import sys
from typing import Any
import urllib.error
import urllib.parse
import urllib.request


BASE_URL = "https://api.agentbody.io"
REQUEST_TIMEOUT = 60
LOGIN_URL = "https://agentbody.io/login"
BILLING_URL = "https://agentbody.io/console/billing"


def _credential_value(path: Path) -> str:
    try:
        lines = path.read_text(encoding="utf-8-sig", errors="replace").splitlines()
    except OSError:
        return ""
    for raw_line in lines:
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[len("export "):].lstrip()
        name, separator, value = line.partition("=")
        if not separator or name.strip() != "AGENTBODY_API_KEY":
            continue
        value = value.strip()
        if value[:1] in ("'", '"'):
            end = value.find(value[0], 1)
            value = value[1:end] if end != -1 else value[1:]
        else:
            for marker in (" #", "\t#"):
                value = value.split(marker, 1)[0]
            value = value.strip()
        if value and "�" not in value:
            return value
    return ""


def resolve_api_key() -> str:
    """Resolve the current AgentBody key without reading sibling profiles."""
    home = Path(os.path.expanduser("~"))
    key = _credential_value(home / ".agentbody" / "credentials")
    if key:
        return key

    key = os.environ.get("AGENTBODY_API_KEY", "").strip()
    if key:
        return key

    candidates = []
    hermes_home = Path(os.environ.get("HERMES_HOME") or home / ".hermes")
    profile = os.environ.get("HERMES_PROFILE", "").strip()
    if profile:
        candidates.append(hermes_home / "profiles" / profile / ".env")
    candidates.append(hermes_home / ".env")
    for path in candidates:
        key = _credential_value(path)
        if key:
            return key
    return ""


class AgentBodyXClient:
    def __init__(self, api_key: str | None = None):
        self.api_key = api_key or resolve_api_key()
        if not self.api_key:
            raise ValueError(
                "AGENTBODY_API_KEY is not configured. Sign in or create an account, "
                f"create a key, and complete one-time setup: {LOGIN_URL}"
            )

    def _request(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
        query = urllib.parse.urlencode({key: value for key, value in params.items() if value is not None})
        url = f"{BASE_URL}{path}" + (f"?{query}" if query else "")
        request = urllib.request.Request(
            url,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Accept": "application/json",
                "User-Agent": "AgentBody-X-Research/1.0",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
                body = response.read().decode("utf-8")
                return json.loads(body) if body else {}
        except urllib.error.HTTPError as error:
            if error.code == 401:
                return {"error": {"code": "UNAUTHORIZED", "message": f"Sign in or create an AgentBody account and configure a key: {LOGIN_URL}"}}
            if error.code == 402:
                return {"error": {"code": "INSUFFICIENT_BALANCE", "message": f"Your AgentBody balance is insufficient. Recharge here: {BILLING_URL}"}}
            return {"error": {"code": f"HTTP_{error.code}", "message": "AgentBody request failed."}}
        except urllib.error.URLError:
            return {"error": {"code": "NETWORK_ERROR", "message": "AgentBody could not be reached."}}

    def search(self, query: str, cursor: str | None = None) -> dict[str, Any]:
        return self._request("/v1/twitter/search", {"query": query, "cursor": cursor})

    def trending(self, country: str | None = None) -> dict[str, Any]:
        return self._request("/v1/twitter/trending", {"country": country})

    def post(self, post_id: str) -> dict[str, Any]:
        return self._request("/v1/twitter/post", {"post_id": post_id})

    def profile(self, username: str) -> dict[str, Any]:
        return self._request("/v1/twitter/profile", {"username": username})

    def profile_posts(self, username: str, cursor: str | None = None) -> dict[str, Any]:
        return self._request("/v1/twitter/profile/posts", {"username": username, "cursor": cursor})

    def profile_media(self, username: str, cursor: str | None = None) -> dict[str, Any]:
        return self._request("/v1/twitter/profile/media", {"username": username, "cursor": cursor})

    def post_comments(self, post_id: str, cursor: str | None = None) -> dict[str, Any]:
        return self._request("/v1/twitter/post/comments", {"post_id": post_id, "cursor": cursor})


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Read current public X/Twitter data through AgentBody.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    command = subparsers.add_parser("search", help="Search public posts.")
    command.add_argument("--query", "-q", required=True)
    command.add_argument("--cursor")

    command = subparsers.add_parser("trending", help="Get trending topics.")
    command.add_argument("--country")

    command = subparsers.add_parser("post", help="Get one public post.")
    command.add_argument("--post-id", required=True)

    command = subparsers.add_parser("profile", help="Get one public profile.")
    command.add_argument("--username", required=True)

    for name, help_text in (("profile-posts", "Get public profile posts."), ("profile-media", "Get public profile media.")):
        command = subparsers.add_parser(name, help=help_text)
        command.add_argument("--username", required=True)
        command.add_argument("--cursor")

    command = subparsers.add_parser("comments", help="Get public post comments.")
    command.add_argument("--post-id", required=True)
    command.add_argument("--cursor")
    return parser


def main() -> None:
    args = build_parser().parse_args()
    try:
        client = AgentBodyXClient()
    except ValueError as error:
        print(json.dumps({"error": {"code": "UNAUTHORIZED", "message": str(error)}}))
        raise SystemExit(1) from error

    handlers = {
        "search": lambda: client.search(args.query, args.cursor),
        "trending": lambda: client.trending(args.country),
        "post": lambda: client.post(args.post_id),
        "profile": lambda: client.profile(args.username),
        "profile-posts": lambda: client.profile_posts(args.username, args.cursor),
        "profile-media": lambda: client.profile_media(args.username, args.cursor),
        "comments": lambda: client.post_comments(args.post_id, args.cursor),
    }
    result = handlers[args.command]()
    print(json.dumps(result, ensure_ascii=False, indent=2))
    raise SystemExit(1 if "error" in result else 0)


if __name__ == "__main__":
    main()