Advanced: Connect Xiaozhi AI to Claude and Codex Without an API Key

What you'll get

Let the Xiaozhi AI on an ESP32 answer questions using your own Claude and Codex (ChatGPT) subscriptions, without applying for an API key or paying per token.

Approach: add 3 LLM providers to the Xiaozhi server, xiaozhi-esp32-server:

Provider	What it does
claude_cli	Starts the locally logged-in claude command line and reads the answer as a stream
codex_cli	Starts the locally logged-in codex command line and reads the answer
fallback	Tries providers in order: if Claude errors, it automatically switches to Codex, and the failing provider cools down for 300 seconds

Key point: OAuth sign-in is handled entirely by the official CLI itself. The server simply "types a line in the command line for you"; it does not read, copy, or refresh any tokens. When a token expires, the CLI renews it itself.

Before you start

1. An ESP32 dev board already flashed with Xiaozhi firmware, able to connect to Wi-Fi and hold a conversation. If it isn't flashed yet, first flash it following the instructions in the Xiaozhi firmware repository.
2. xiaozhi-esp32-server deployed from source (not Docker). The CLI login state is stored in your local user directory, which the Docker container cannot see, so the server must run directly on the computer where you logged in to the CLIs. The code is in the main/xiaozhi-server directory of the repository.
3. Install and sign in to both CLIs:

# Claude Code: after installing, run claude once and sign in with your Claude account as prompted
npm install -g @anthropic-ai/claude-code
claude

# Codex: sign in with your ChatGPT account
npm install -g @openai/codex
codex login

First, try each one in the terminal to confirm it answers:

echo "你好" | claude -p --model claude-haiku-5-5
echo "你好" | codex exec --skip-git-repo-check -

4. Point the device's OTA address to your server, for example http://your-computer-IP:8003/xiaozhi/ota/, so the device connects to the server you deployed yourself.

Subscription accounts are only suitable for personal use. Do not expose this server so others can share your account.

How Xiaozhi calls the LLM

Each LLM on the Xiaozhi server is an LLMProvider class in core/providers/llm/<type>/<type>.py. If the config says type: openai, it loads openai/openai.py.

Each provider must implement two methods, following the approach in core/providers/llm/openai:

• response(session_id, dialogue): normal conversation, yields text piece by piece, sending each piece on to speech synthesis as it is generated.
• response_with_functions(session_id, dialogue, functions): conversation with tools, yields (text, tool calls). When a tool needs to be called, text is None, and tool calls is a list where each item has id, function.name, and function.arguments.

So what we need to do is: assemble dialogue into a single prompt, hand it to the CLI, and split the CLI's output back into these two formats.

Step 1: shared module cli_common.py

The two CLI providers share three things:

• Building the prompt: pull the system prompt out separately and assemble the rest of the conversation into a "User: ... Assistant: ..." transcript.
• Running the subprocess: start the CLI, pass the conversation in via stdin, and parse the JSON it outputs line by line; kill the process on timeout; record the time to first token and the total time.
• Tool calls: the CLI's built-in tools are already turned off, so we write functions into the prompt and require the model to output only <tool_call>{"name":"...","arguments":{...}}</tool_call> when it needs a tool. During streaming, once this tag appears, cut it off so it isn't spoken; after the output finishes, parse it into a tool call and return it.

Create core/providers/llm/cli_common.py:

"""Shared by claude_cli and codex_cli: builds the prompt, runs the subprocess, and parses <tool_call>."""
import json
import re
import subprocess
import threading
import time
import uuid
from types import SimpleNamespace

from config.logger import setup_logging

TAG = __name__
logger = setup_logging()

TOOL_RULE = """You can call the tools listed below. When you need to call a tool, output only:
<tool_call>{"name": "tool_name", "arguments": {arguments}}</tool_call>
Do not output any other text, do not explain, and do not read the tool call aloud to the user.
When no tool is needed, answer directly in short, conversational language.

Tool list (one per line, JSON):
"""


def tools_text(functions):
    lines = []
    for f in functions or []:
        fn = f.get("function", f)
        lines.append(json.dumps({
            "name": fn.get("name"),
            "description": fn.get("description", ""),
            "parameters": fn.get("parameters", {}),
        }, ensure_ascii=False))
    return "\n".join(lines)


def _get(obj, key, default=None):
    return obj.get(key, default) if isinstance(obj, dict) else getattr(obj, key, default)


def build_prompt(dialogue, functions=None):
    """Split an OpenAI-format dialogue into (system, prompt)."""
    system, turns = [], []
    for m in dialogue:
        role, content = m.get("role"), m.get("content") or ""
        if role == "system":
            system.append(content)
        elif role == "assistant" and m.get("tool_calls"):
            for tc in m["tool_calls"]:
                fn = _get(tc, "function", {})
                call = {"name": _get(fn, "name"), "arguments": json.loads(_get(fn, "arguments") or "{}")}
                turns.append("Assistant: <tool_call>" + json.dumps(call, ensure_ascii=False) + "</tool_call>")
            if content:
                turns.append("Assistant: " + content)
        elif role == "tool":
            turns.append("Tool result: " + content)
        else:
            turns.append(("User: " if role == "user" else "Assistant: ") + content)
    if functions:
        system.append(TOOL_RULE + tools_text(functions))
    prompt = "Below is the dialogue transcript. Reply to the last message as the \"Assistant\", and output only the reply content itself.\n\n" + "\n".join(turns)
    return "\n\n".join(system), prompt


def run_cli(name, cmd, stdin_text, parse, timeout=60, cwd=None):
    """Start the CLI, read JSON line by line, and yield text fragments from parse(obj). Records time to first token and total time."""
    t0 = time.time()
    p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE, text=True, encoding="utf-8", cwd=cwd)
    err = []
    threading.Thread(target=lambda: err.append(p.stderr.read()), daemon=True).start()
    timer = threading.Timer(timeout, p.kill)
    timer.start()
    first = None
    try:
        p.stdin.write(stdin_text)
        p.stdin.close()
        for line in p.stdout:
            line = line.strip()
            if not line.startswith("{"):
                continue
            try:
                obj = json.loads(line)
            except ValueError:
                continue
            for text in parse(obj):
                if text:
                    if first is None:
                        first = time.time()
                    yield text
        rc = p.wait()
        if rc != 0 or first is None:
            raise RuntimeError(f"{name} exited with code {rc} and produced no output: {''.join(err)[-300:]}")
        logger.bind(tag=TAG).info(
            f"{name} first token {(first - t0) * 1000:.0f} ms, total {(time.time() - t0) * 1000:.0f} ms")
    finally:
        timer.cancel()
        if p.poll() is None:
            p.kill()


START = "<tool_call>"


def _cut(buf):
    """Return (length that is safe to speak, whether <tool_call> has appeared)."""
    i = buf.find(START)
    if i >= 0:
        return i, True
    for k in range(min(len(START) - 1, len(buf)), 0, -1):
        if START.startswith(buf[-k:]):  # The end may be half a tag, so hold it back
            return len(buf) - k, False
    return len(buf), False


def parse_tool_calls(text):
    calls = []
    for raw in re.findall(r"<tool_call>\s*(\{.*?\})\s*(?:</tool_call>|$)", text, re.S):
        try:
            data = json.loads(raw)
        except ValueError:
            continue
        calls.append(SimpleNamespace(
            index=len(calls),
            id=uuid.uuid4().hex,
            function=SimpleNamespace(
                name=data.get("name"),
                arguments=json.dumps(data.get("arguments") or {}, ensure_ascii=False),
            ),
        ))
    return calls


def with_tool_calls(chunks):
    """Text streams out as (text, None) as usual; on reaching <tool_call>, hold it back, and at the end yield (None, [tool_call])."""
    buf, tool = "", None
    for c in chunks:
        if tool is not None:
            tool += c
            continue
        buf += c
        n, found = _cut(buf)
        if n:
            yield buf[:n], None
        if found:
            tool, buf = buf[n:], ""
        else:
            buf = buf[n:]
    if tool is None:
        if buf:
            yield buf, None
        return
    calls = parse_tool_calls(tool)
    if calls:
        yield None, calls
    else:
        logger.bind(tag=TAG).warning(f"Malformed tool call, discarded: {tool[:200]}")

with_tool_calls has one detail worth noting: text arrives in small pieces, and a <tool_call> tag can be split across two pieces. So if the end of the buffer looks like a partial tag (for example <tool), it is held back and not spoken until the next piece arrives and settles the question.

Step 2: claude_cli

Create core/providers/llm/claude_cli/claude_cli.py. The command is:

claude -p --output-format stream-json --verbose --model claude-haiku-5-5 --tools ""

Add a few more flags to make it faster and cleaner:

• --include-partial-messages: streams output character by character. Without it, you wait for the whole sentence to be written before getting any text.
• --tools "": turns off all of Claude Code's built-in tools (reading files, running commands, and so on). It only talks.
• --strict-mcp-config: does not load the MCP servers configured on your machine, so startup is faster.
• --system-prompt: replaces Claude Code's default long system prompt with XiaoZhi's role prompt.

from config.logger import setup_logging
from core.providers.llm.base import LLMProviderBase
from core.providers.llm.cli_common import build_prompt, run_cli, with_tool_calls

TAG = __name__
logger = setup_logging()


class LLMProvider(LLMProviderBase):
    """Calls the locally logged-in claude CLI. OAuth is handled by the CLI itself; the token is not touched here."""

    def __init__(self, config):
        self.bin = config.get("bin") or "claude"
        self.model = config.get("model_name") or "claude-haiku-5-5"
        self.timeout = int(config.get("timeout") or 60)

    def _cmd(self, system):
        return [
            self.bin, "-p",
            "--output-format", "stream-json", "--verbose",
            "--include-partial-messages",      # word-by-word output, faster first word
            "--model", self.model,
            "--tools", "",                     # turn off all built-in Claude Code tools
            "--strict-mcp-config",             # do not load local MCP servers
            "--no-session-persistence",        # do not keep session records locally
            "--system-prompt", system or "You are a voice assistant; answer briefly and conversationally.",
        ]

    @staticmethod
    def _parser():
        seen_delta = [False]

        def parse(o):
            t = o.get("type")
            if t == "stream_event":
                e = o.get("event") or {}
                d = e.get("delta") or {}
                if e.get("type") == "content_block_delta" and d.get("type") == "text_delta":
                    seen_delta[0] = True
                    yield d.get("text", "")
            elif t == "assistant" and not seen_delta[0]:   # older versions have no word-by-word events
                for b in (o.get("message") or {}).get("content", []):
                    if b.get("type") == "text":
                        yield b.get("text", "")
            elif t == "result" and o.get("is_error"):
                raise RuntimeError(f"claude error: {o.get('result')}")
        return parse

    def _stream(self, dialogue, functions=None):
        system, prompt = build_prompt(dialogue, functions)
        return run_cli("claude_cli", self._cmd(system), prompt, self._parser(), self.timeout)

    def response(self, session_id, dialogue, **kwargs):
        yield from self._stream(dialogue)

    def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
        yield from with_tool_calls(self._stream(dialogue, functions))

Step 3: codex_cli

Create core/providers/llm/codex_cli/codex_cli.py. The command is:

codex exec --json --skip-git-repo-check -

The final - means the prompt is read from stdin. --json outputs one event per line, and the answer is in the line where type is item.completed and item.type is agent_message.

Three additional safeguards:

• --sandbox read-only: Codex is a coding agent that runs commands. Here it is only allowed to talk, not to change anything.
• Run it in an empty temporary directory so it cannot read the server's code and configuration.
• model_reasoning_effort=low: voice conversation does not need deep reasoning, and lowering it makes answers faster.

import tempfile

from config.logger import setup_logging
from core.providers.llm.base import LLMProviderBase
from core.providers.llm.cli_common import build_prompt, run_cli, with_tool_calls

TAG = __name__
logger = setup_logging()


class LLMProvider(LLMProviderBase):
    """Calls the locally logged-in codex CLI (codex login). OAuth is handled by the CLI itself."""

    def __init__(self, config):
        self.bin = config.get("bin") or "codex"
        self.model = config.get("model_name")
        self.effort = config.get("reasoning_effort") or "low"
        self.timeout = int(config.get("timeout") or 90)
        self.workdir = tempfile.mkdtemp(prefix="xiaozhi-codex-")   # empty directory, so it cannot read project files

    def _cmd(self):
        cmd = [self.bin, "exec", "--json", "--skip-git-repo-check",
               "--sandbox", "read-only",
               "-c", f"model_reasoning_effort={self.effort}"]
        if self.model:
            cmd += ["-m", self.model]
        return cmd + ["-"]          # "-" means the prompt is read from stdin

    @staticmethod
    def parse(o):
        t = o.get("type")
        item = o.get("item") or {}
        if t == "item.completed" and item.get("type") == "agent_message":
            yield item.get("text", "")
        elif t in ("turn.failed", "error"):
            raise RuntimeError(f"codex error: {o.get('error') or o.get('message')}")

    def _stream(self, dialogue, functions=None):
        system, prompt = build_prompt(dialogue, functions)
        # codex exec has no separate system parameter, so it is prepended
        text = system + "\n\nDo not run any commands, do not read or write files, just answer directly.\n\n" + prompt
        return run_cli("codex_cli", self._cmd(), text, self.parse, self.timeout, cwd=self.workdir)

    def response(self, session_id, dialogue, **kwargs):
        yield from self._stream(dialogue)

    def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
        yield from with_tool_calls(self._stream(dialogue, functions))

codex exec --json does not currently output word by word; the whole answer is returned only after it is fully written, so its time to first word is basically the same as its total time. Placing it second in the chain as a backup is a good fit.

Step 4: fallback

Create core/providers/llm/fallback/fallback.py. It does not call a model itself; it only hands the request to the other providers in the order given by chain:

• If the previous one errors before it has spoken its first word (not logged in, command not found, timeout, quota used up), switch to the next one.
• The provider that failed is recorded with a "cooldown until" time, and for 300 seconds it is skipped directly, so it does not have to fail once before every sentence.
• If it errors after half of the answer has already been spoken, do not switch. Switching to another model would start over from the beginning, and the user would hear the opening twice.

"}

import os
import time

from config.config_loader import get_project_dir, merge_configs, read_config
from config.logger import setup_logging
from core.providers.llm.base import LLMProviderBase
from core.utils.llm import create_instance

TAG = __name__
logger = setup_logging()


class LLMProvider(LLMProviderBase):
    """Try providers in chain order; a provider that errors goes into cooldown for cooldown seconds and is skipped during that time."""

    def __init__(self, config):
        self.cooldown = int(config.get("cooldown") or 300)
        llm_conf = self._llm_section()
        self.providers = []
        for name in config.get("chain") or []:
            conf = llm_conf.get(name)
            if not conf:
                logger.bind(tag=TAG).error(f"fallback: 找不到 LLM.{name} 的配置，已跳过")
                continue
            self.providers.append((name, create_instance(conf.get("type", name), conf)))
        if not self.providers:
            raise ValueError("fallback: chain 里没有可用的 LLM")
        self.down_until = {}

    @staticmethod
    def _llm_section():
        d = get_project_dir()
        conf = read_config(d + "config.yaml")
        custom = d + "data/.config.yaml"
        if os.path.exists(custom):
            conf = merge_configs(conf, read_config(custom))
        return conf.get("LLM", {})

    def _run(self, call):
        now = time.time()
        ready = [(n, p) for n, p in self.providers if self.down_until.get(n, 0) <= now]
        last = None
        for name, p in ready or self.providers:   # If all are in cooldown, retry them all anyway; better than staying silent
            started = False
            try:
                for item in call(p):
                    started = True
                    yield item
                self.down_until.pop(name, None)
                return
            except Exception as e:
                last = e
                self.down_until[name] = time.time() + self.cooldown
                logger.bind(tag=TAG).warning(f"{name} 出错，冷却 {self.cooldown} 秒：{e}")
                if started:   # Part of the reply has already been spoken; cannot switch to another model and start over
                    return
        raise last or RuntimeError("fallback: 没有可用的 LLM")

    def response(self, session_id, dialogue, **kwargs):
        yield from self._run(lambda p: p.response(session_id, dialogue, **kwargs))

    def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
        yield from self._run(
            lambda p: p.response_with_functions(session_id, dialogue, functions=functions, **kwargs))

fallback only receives its own section of the config, so it reads config.yaml and data/.config.yaml itself and looks up the claude_cli and codex_cli configs by name. If you deploy configuration through the management console (manager-api), change it to read from the console, or write the sub-configs directly into fallback.

Step 5: Change the config

Edit data/.config.yaml, add the three LLMs, and set the selected LLM to fallback. Intent recognition must use function_call so that tool calls go through response_with_functions:

selected_module:
  LLM: fallback
  Intent: function_call

LLM:
  claude_cli:
    type: claude_cli
    model_name: claude-haiku-5-5
    timeout: 60
  codex_cli:
    type: codex_cli
    reasoning_effort: low
    timeout: 90
  fallback:
    type: fallback
    chain: [claude_cli, codex_cli]
    cooldown: 300

After the change, restart the server:

cd main/xiaozhi-server
python app.py

Step 6: Test

Test the providers on their own, without a device

In the main/xiaozhi-server directory, create test_cli_llm.py. It calls fallback directly and tests plain conversation, tool calls, and the reply after a tool result comes back:

import sys, time
from core.utils.llm import create_instance
from config.config_loader import read_config
name = sys.argv[1] if len(sys.argv) > 1 else "fallback"
llm = create_instance(read_config("data/.config.yaml")["LLM"][name]["type"], read_config("data/.config.yaml")["LLM"][name])
sysmsg = {"role": "system", "content": "你是小智，一个语音助手，回答简短口语化，不超过两句话。"}
print("== 普通对话")
t = time.time(); print("".join(llm.response("t", [sysmsg, {"role": "user", "content": "你好，介绍一下你自己"}])), f"({time.time()-t:.1f}s)")
tools = [{"type": "function", "function": {"name": "get_weather", "description": "查询某个城市的天气",
          "parameters": {"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}}}]
print("== 工具调用")
for text, calls in llm.response_with_functions("t", [sysmsg, {"role": "user", "content": "北京今天天气怎么样"}], functions=tools):
    if text: print("念出来:", repr(text))
    if calls: print("工具:", [(c.function.name, c.function.arguments) for c in calls])
print("== 工具结果之后")
d = [sysmsg, {"role": "user", "content": "北京今天天气怎么样"},
     {"role": "assistant", "content": "", "tool_calls": [{"id": "1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"location\": \"北京\"}"}}]},
     {"role": "tool", "tool_call_id": "1", "content": "北京：晴，12~22 度，北风 2 级"}]
print("".join(t or "" for t, _ in llm.response_with_functions("t", d, functions=tools)))

python test_cli_llm.py fallback

Output from our test on a Mac (Claude subscription, Haiku 5.5):

Note the tool-call section: the <tool_call> the model output was not spoken at all. It became the get_weather call directly.

Test on a device

1. Plain conversation: say "你好，介绍一下你自己" to Xiaozhi. The log should show claude_cli 首字 … ms.
2. Tool call: say "今天北京天气怎么样" or "放一首歌" (requires the matching plugin to be enabled under Intent.function_call.functions). The log should show "检测到 1 个工具调用", and Xiaozhi should not speak the <tool_call>.
3. Automatic switching: temporarily add a line bin: claude-not-here (a nonexistent command) under claude_cli, restart, and ask a question. The log will show:

WARNING claude_cli 出错，冷却 300 秒：[Errno 2] No such file or directory: 'claude-not-here'
INFO codex_cli 首字 … ms，总耗时 … ms

The reply still comes through, just from Codex instead. After testing, delete that line.

Check latency

grep -E "首字|冷却" tmp/server.log | tail -20

• First-token latency: from sending the request to receiving the first character. Xiaozhi starts speech synthesis as soon as it gets the first sentence, so this is the stretch users feel most.
• Total time: until the full reply has been written.

In our tests, Claude (Haiku 5.5) had first-token latency of about 1.1 to 2.3 seconds and total time of 2 to 3 seconds. About 1 second of that is the overhead of starting the CLI process each time, which is slower than calling the API directly; the user's total wait also includes speech recognition and speech synthesis. Codex does not output word by word and generally has higher latency, so it suits being the backup.

Common questions

• Log shows "exit code 1, no output": most likely the CLI is not logged in, or the login has expired. Run claude or codex login manually once under the same system user. If the server runs as a different user (for example a systemd service), that user also needs to be logged in.
• Error No such file or directory: 'claude': the server process cannot find the command on its PATH. Use the full path in the config with bin:, for example bin: /Users/you/.local/bin/claude (find it with which claude).
• XiaoZhi reads out <tool_call>: this means the turn went through response instead of response_with_functions. Check whether selected_module.Intent is function_call.
• Replies are too long: state clearly in XiaoZhi's role prompt that replies must not exceed two sentences. For voice scenarios, shorter is better.
• Subscription quota used up: the CLI reports an error, fallback automatically switches to the next one, and it retries after a 300-second cooldown.

References

• xiaozhi-esp32-server: XiaoZhi server
• xiaozhi-esp32: XiaoZhi firmware
• Claude Code CLI reference
• Codex CLI