Simple CharacterAI Python API Wrapper

api.py

import json

import aiohttp

#
TEST_TOKEN: str = ""
#


class Character:
    def __init__(
        self, client: "CharacterAI", data: dict[str, str], character_id: str
    ) -> None:
        self.client: "CharacterAI" = client
        self.data: dict[str, str] = data
        self.id: str = character_id

    # Properties
    @property
    def external_id(self) -> str | None:
        return self.data.get("external_id", None)

    @property
    def name(self) -> str:
        return self.ai_data.get("name")

    @property
    def ai_data(self) -> dict[str, str]:
        for participants in self.data.get("participants"):
            if not participants["is_human"]:
                return participants

    @property
    def avatar(self) -> str:
        return "https://characterai.io/i/80/static/avatars/" + self.data["messages"][
            0
        ].get("src__character__avatar_file_name")

    # Public API
    async def send(self, message: str) -> str:
        # HACK: ??
        payload: dict[str, str] = {
            "history_external_id": self.external_id,
            "character_external_id": self.id,
            "text": message,
            "tgt": self.ai_data["user"]["username"],
            "ranking_method": "random",
            "faux_chat": False,
            "staging": False,
            "model_server_address": None,
            "override_prefix": None,
            "override_rank": None,
            "rank_candidates": None,
            "filter_candidates": None,
            "prefix_limit": None,
            "prefix_token_limit": None,
            "livetune_coeff": None,
            "stream_params": None,
            "enable_tti": True,
            "initial_timeout": None,
            "insert_beginning": None,
            "translate_candidates": None,
            "stream_every_n_steps": 16,
            "chunks_to_pad": 8,
            "is_proactive": False,
        }

        async with self.client.http.post(
            "https://beta.character.ai/chat/streaming/",
            data=payload,
            headers=self.client._generate_headers(),
        ) as res:
            if not res or res.status != 200:
                raise RuntimeError(f"Failed to send message to {self.id=}")

            # The data is kinda fucked
            replies: list[dict[str, str]] = []

            for line in (await res.read()).decode().split("\n"):
                if line.strip().startswith("{"):
                    replies.append(json.loads(line.strip()))
                    continue

                try:
                    start: int = line.index(" {")
                    replies.append(json.loads(line[start - 1 :]))
                except Exception:
                    continue

            return replies[-1]["replies"][0]["text"]

    @classmethod
    def from_data(
        cls, data: dict[str, str], character_id: str, client: "CharacterAI"
    ) -> "Character":
        self: "Character" = cls(client, data, character_id)

        return self


class CharacterAI:
    def __init__(self) -> None:
        self.http: aiohttp.ClientSession = None
        self.token: str | None = None

    def _generate_headers(self, **kwargs) -> dict[str, str]:
        headers: dict[str, str] = {
            **kwargs,
            "User-Agent": "Mozilla/5.0 (X11; Linux i686; rv:108.0) Gecko/20100101 Firefox/108.0",
        }

        if self.token:
            headers |= {"Authorization": f"Token {self.token}"}

        return headers

    # Internal API
    async def _authenticate(self, access_token: str) -> None:
        async with self.http.post(
            "https://beta.character.ai/dj-rest-auth/auth0/",
            data={"access_token": access_token},
            headers=self._generate_headers(),
        ) as res:
            if not res or res.status != 200:
                raise RuntimeError("Failed to authenticate!")

            if data := await res.json():
                data: dict[str, str]

                self.token = data.get("key", None)

    # Public API
    async def get_character(self, character_id: str) -> Character:
        async with self.http.post(
            "https://beta.character.ai/chat/history/continue/",
            data={"character_external_id": character_id, "history_external_id": None},
            headers=self._generate_headers(),
        ) as res:
            if not res or res.status != 200:
                raise RuntimeError(f"Failed to get character with id: {character_id=}")

            character_data: dict[str, str] = await res.json()

            # Check status, if it says no history then create instead of continuing
            if (_ := (await res.json()).get("status")) == "No Such History":
                # No history, create character history.
                async with self.http.post(
                    "https://beta.character.ai/chat/history/create/",
                    data={
                        "character_external_id": character_id,
                        "history_external_id": None,
                    },
                    headers=self._generate_headers(),
                ) as res_create:
                    character_data = await res_create.json()

        return Character.from_data(character_data, character_id, self)

    @classmethod
    async def create(cls, token: str | None = None) -> "CharacterAI":
        if not token:
            token = TEST_TOKEN

        self = cls()

        self.http = aiohttp.ClientSession()

        # attempt to login
        await self._authenticate(access_token=token)

        return self


if __name__ == "__main__":
    import asyncio

    async def main() -> None:
        client: CharacterAI = await CharacterAI.create(TEST_TOKEN)

        if char := await client.get_character(
            "z7Y1m2mkugEb5u5vRwUELYrlULrhs3hke6Ap08KcvQY"
        ):
            while True:
                reply: str = input("> ")
                print(f"{char.name}: {await char.send(reply)}")

    asyncio.run(main())