diff --git a/README.md b/README.md index f1dbe0e..25e6ba0 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,8 @@ through [Ollama](https://ollama.com), and when it needs current information it c (`lcode mcp add atlassian`), image generation with ComfyUI, or any other MCP server. Browser sign-in (OAuth) is built in. - **Sees images.** Attach a screenshot or mockup with `@path` and lcode looks at it, using the vision part of your model. With ComfyUI (`lcode mcp add comfyui`) it can also generate and edit - images with local models such as FLUX, SDXL and Qwen-Image. + images with local models such as FLUX, SDXL and Qwen-Image, or, in the cloud, images, videos + and speech with xAI's Grok Imagine (`lcode mcp add xai`). - **Web search when needed.** Looks up the latest versions, docs and error messages with Ollama web search, Brave, Tavily or your own SearXNG, and reads pages as clean text. - **Private.** The model runs on your machine, lcode never uploads your files and has no telemetry. diff --git a/docs/mcp.md b/docs/mcp.md index 262cc53..68f7383 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -29,6 +29,7 @@ and find the code for the top one."* | `github` | Repositories, issues, pull requests, Actions | a GitHub token (or the GitHub CLI) | | `playwright` | A real (headless) browser: open pages, click, fill forms | `npx` | | `comfyui` | Generate and edit images with local models (FLUX, SDXL, SD 1.5, Qwen-Image) | `uvx` and [ComfyUI](#images-with-comfyui) running locally | +| `xai` | Generate and edit images, and generate videos and speech, with xAI's Grok Imagine (paid, in the cloud) | `uvx`, an xAI API key ([below](#images-videos-and-speech-with-xai)) | | `context7` | Up-to-date docs and examples for thousands of libraries | nothing (an API key is optional) | | `sentry` | Errors, issues, traces and releases | a Sentry account (browser sign-in) | | `postgres` | Schemas, read-only queries, query performance | `uvx`, a connection URL | @@ -137,6 +138,36 @@ size, prompt), which can take several minutes of trial and error. Once an image lcode to save that workflow in your project (for example `assets/icon.workflow.json`) and reuse it: later images are a single `run_workflow` call. +### Images, videos and speech with xAI + +xAI's [Grok Imagine](https://docs.x.ai/developers/model-capabilities/imagine) and +[text to speech](https://docs.x.ai/developers/models/text-to-speech) APIs make images, videos and +speech in the cloud: no GPU needed, but each call is billed to your xAI account. lcode ships an MCP +server for them, `lcode-mcp-xai`. + +```bash +lcode mcp add xai # asks for your xAI API key (create one at https://console.x.ai) +``` + +The key is typed hidden and kept in `~/.config/lcode/mcp.json`, readable only by you. If +`XAI_API_KEY` is already set in your environment, lcode uses that and doesn't store the key. + +| Tool | | Costs | +|---|---|---| +| `generate_image` | Images from a description: 1–10 at a time, aspect ratio, 1k or 2k | $0.02–0.05 an image | +| `edit_image` | Change a local image as described | the same | +| `generate_video` | A 1–15 second MP4 from a description, optionally starting from a local image, up to 1080p. Waits for it (a minute or more) | $0.02–0.08 a second | +| `video_status` | Fetch a video that wasn't ready within 10 minutes | | +| `text_to_speech` | Speech in one of 26 voices and 20 languages, as MP3 or WAV | per character | + +Files are saved in `generated/` in the project, or where you ask (*"generate an app icon and save +it as assets/icon.png"*). xAI has no music generation API. + +Your prompts, and the images you edit or animate, go to xAI. lcode asks before each call, since each +one costs money. To stop asking for some tools, list them under `"allow"` for the server in +`mcp.json`, for example `"allow": ["generate_image", "text_to_speech"]`. xAI applies its own usage +policies to what it generates; when it refuses a prompt, lcode passes its message to the model. + ### Metabase Metabase has its own MCP server inside every instance (Metabase 60 and later), at diff --git a/pyproject.toml b/pyproject.toml index 36abd97..07cc448 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ Changelog = "https://github.com/nasser1941/lcode/blob/main/CHANGELOG.md" lcode = "lcode.cli:main" lcode-mcp-valohai = "lcode.servers.valohai:main" lcode-mcp-encord = "lcode.servers.encord:main" +lcode-mcp-xai = "lcode.servers.xai:main" [dependency-groups] dev = ["pytest>=8", "ruff>=0.6"] @@ -88,6 +89,7 @@ ignore = [ "src/lcode/action.py" = ["E501"] "src/lcode/acp.py" = ["E501"] "src/lcode/update.py" = ["E501"] +"src/lcode/servers/xai.py" = ["E501"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/src/lcode/mcp/catalog.toml b/src/lcode/mcp/catalog.toml index 5f6a434..fc92410 100644 --- a/src/lcode/mcp/catalog.toml +++ b/src/lcode/mcp/catalog.toml @@ -281,3 +281,24 @@ secret = true var = "VALOHAI_HOST" prompt = "Valohai address (Enter for app.valohai.com; for a self-hosted installation, its URL)" optional = true + +[xai] +name = "xAI Grok Imagine" +description = "Generate and edit images, and generate videos and speech, with xAI's Grok Imagine and voice API (paid, in the cloud)" +homepage = "https://nasser1941.github.io/lcode/mcp/#xai" +requires = ["uvx"] +setup = """Uses xAI's API: your prompts, and images you edit or animate, go to xAI, and every call is billed to your xAI +account (an image $0.02-0.05, video $0.02-0.08 a second, speech per character). Create an API key at https://console.x.ai. +Files are saved in generated/ in the project. lcode asks before each call; to stop asking for some tools, list them +under "allow" for this server in mcp.json. xAI has no music generation.""" + +[xai.server] +command = "uvx" +args = ["--isolated", "--python", "3.12", "--from", "lcode-cli", "lcode-mcp-xai"] +env = { XAI_API_KEY = "${XAI_API_KEY}" } +timeout = 900 # a video takes minutes + +[[xai.inputs]] +var = "XAI_API_KEY" +prompt = "xAI API key (create one at https://console.x.ai)" +secret = true diff --git a/src/lcode/servers/xai.py b/src/lcode/servers/xai.py new file mode 100644 index 0000000..ad8da23 --- /dev/null +++ b/src/lcode/servers/xai.py @@ -0,0 +1,289 @@ +"""An MCP server for xAI's Grok Imagine and voice APIs: images, image edits, videos and speech. + +xAI has an HTTP API but no MCP server, so lcode ships this one. It needs an API key (XAI_API_KEY, +from https://console.x.ai) and saves what it makes in the project, in `generated/` unless a path is +given. The prompts, and any image it edits or animates, go to xAI, which bills per image, per +second of video and per character of speech, and applies its own usage policies. + +Run: lcode-mcp-xai (lcode mcp add xai sets it up) +""" + +from __future__ import annotations + +import argparse +import base64 +import mimetypes +import os +import re +import time +from pathlib import Path + +import requests + +from lcode import __version__ +from lcode.mcp.server import Server, ToolFailure + +API = "https://api.x.ai/v1" +TIMEOUT = 120 +IMAGE_MODELS = ["grok-imagine-image-2.0", "grok-imagine-image-quality", "grok-imagine-image"] +VIDEO_MODELS = ["grok-imagine-video-1.5", "grok-imagine-video-1.5-lite", "grok-imagine-video"] +ASPECT_RATIOS = ["auto", "1:1", "16:9", "9:16", "4:3", "3:4", "3:2", "2:3", "2:1", "1:2", "21:9"] +VOICES = ["eve", "ara", "leo", "rex", "sal", "luna", "iris", "orion", "helix", "zagan", "carina", "altair", "zenith", + "perseus", "helios", "lux", "kepler", "rigel", "cosmo", "celeste", "ursa", "sirius", "lumen", "castor", + "naksh", "atlas"] # fmt: skip +AUDIO_FORMATS = {"mp3": ".mp3", "wav": ".wav"} +VIDEO_WAIT = 600 # seconds generate_video waits before handing back the request id +POLL = 5 + + +class XAI: + def __init__(self, key: str, base: str = API, output: Path | None = None): + self.base = base.rstrip("/") + self.output = output or Path("generated") + self.session = requests.Session() + self.session.headers.update({"Authorization": f"Bearer {key}", "Content-Type": "application/json"}) + + def request(self, method: str, path: str, **kwargs) -> requests.Response: + try: + r = self.session.request(method, f"{self.base}{path}", timeout=TIMEOUT, **kwargs) + except requests.RequestException as e: + raise ToolFailure(f"can't reach xAI: {e}") from e + if r.status_code in (401, 403): + raise ToolFailure( + f"xAI refused the request ({r.status_code}): {_error(r)}. Check XAI_API_KEY (console.x.ai)." + ) + if r.status_code == 429: + raise ToolFailure(f"xAI's rate limit or credit limit was reached: {_error(r)}") + if r.status_code >= 400: + raise ToolFailure(f"xAI error {r.status_code}: {_error(r)}") + return r + + def post(self, path: str, body: dict) -> dict: + return self.request("POST", path, json=body).json() + + def download(self, url: str) -> bytes: + try: + r = requests.get(url, timeout=TIMEOUT) + r.raise_for_status() + except requests.RequestException as e: + raise ToolFailure(f"couldn't download the result from xAI: {e}") from e + return r.content + + def target(self, path: str, stem: str, suffix: str) -> Path: + """Where to save: the given path, or a new file in the output folder.""" + if path: + target = Path(path).expanduser() + if not target.suffix: + target = target.with_suffix(suffix) + else: + name = re.sub(r"[^a-z0-9]+", "-", stem.lower()).strip("-")[:40] or "xai" + target = self.output / f"{name}-{time.strftime('%Y%m%d-%H%M%S')}{suffix}" + target.parent.mkdir(parents=True, exist_ok=True) + n, unique = 2, target + while unique.exists(): + unique = target.with_name(f"{target.stem}-{n}{target.suffix}") + n += 1 + return unique + + +def _error(r: requests.Response) -> str: + try: + data = r.json() + except ValueError: + return r.text[:300] + error = data.get("error") if isinstance(data, dict) else None + if isinstance(error, dict): + return str(error.get("message") or error)[:300] + return str(error or data)[:300] + + +def data_url(path: str) -> str: + """A local image as a data: URL, for edits and image-to-video.""" + file = Path(path).expanduser() + if not file.is_file(): + raise ToolFailure(f"{path} doesn't exist") + mime = mimetypes.guess_type(file.name)[0] or "image/png" + if mime not in ("image/png", "image/jpeg", "image/webp"): + raise ToolFailure(f"{path} isn't a PNG, JPEG or WebP image") + return f"data:{mime};base64,{base64.b64encode(file.read_bytes()).decode()}" + + +def image_suffix(data: bytes) -> str: + if data[:3] == b"\xff\xd8\xff": + return ".jpg" + if data[:4] == b"RIFF" and data[8:12] == b"WEBP": + return ".webp" + return ".png" + + +def save_images(api: XAI, reply: dict, prompt: str, path: str) -> list[Path]: + saved = [] + items = reply.get("data") or [] + if not items: + raise ToolFailure(f"xAI returned no image: {str(reply)[:300]}") + for i, item in enumerate(items): + data = base64.b64decode(item["b64_json"]) if item.get("b64_json") else api.download(item["url"]) + many = path and len(items) > 1 + target = api.target(f"{Path(path).with_suffix('')}-{i + 1}" if many else path, prompt, image_suffix(data)) + target.write_bytes(data) + saved.append(target) + return saved + + +def build(api: XAI) -> Server: + server = Server( + "xai", + __version__, + "xAI's Grok Imagine and voice: generate and edit images, generate videos (from text or a start image) " + "and speech. Results are saved as files in the project; each call costs money on the user's xAI account. " + "xAI has no music generation.", + ) + path_arg = {"type": "string", "description": "Where to save it (default: a new file in generated/)"} + + @server.tool( + "generate_image", + "Generate images from a description with Grok Imagine and save them. Costs $0.02–0.05 per image.", + { + "prompt": {"type": "string", "description": "What the image shows, in detail"}, + "n": {"type": "integer", "description": "How many images, 1–10 (default 1)"}, + "aspect_ratio": {"type": "string", "enum": ASPECT_RATIOS, "description": "Default auto"}, + "resolution": {"type": "string", "enum": ["1k", "2k"], "description": "Default 1k"}, + "model": {"type": "string", "enum": IMAGE_MODELS, "description": f"Default {IMAGE_MODELS[0]}"}, + "path": path_arg, + }, + ["prompt"], + ) + def generate_image(prompt: str, n: int = 1, aspect_ratio: str = "auto", resolution: str = "1k", + model: str = IMAGE_MODELS[0], path: str = "") -> str: # fmt: skip + body = {"model": model, "prompt": prompt, "n": max(1, min(10, int(n))), "response_format": "b64_json"} + if aspect_ratio and aspect_ratio != "auto": + body["aspect_ratio"] = aspect_ratio + if resolution and resolution != "1k": + body["resolution"] = resolution + saved = save_images(api, api.post("/images/generations", body), prompt, path) + return f"Saved {len(saved)} image(s) from {model}:\n" + "\n".join(f"- {p}" for p in saved) + + @server.tool( + "edit_image", + "Change an existing image (a local file) as described, with Grok Imagine, and save the result.", + { + "image": {"type": "string", "description": "Path of the image to edit (PNG, JPEG or WebP)"}, + "prompt": {"type": "string", "description": "What to change"}, + "model": {"type": "string", "enum": IMAGE_MODELS, "description": f"Default {IMAGE_MODELS[0]}"}, + "path": path_arg, + }, + ["image", "prompt"], + ) + def edit_image(image: str, prompt: str, model: str = IMAGE_MODELS[0], path: str = "") -> str: + body = {"model": model, "prompt": prompt, "image": {"url": data_url(image), "type": "image_url"}, + "response_format": "b64_json"} # fmt: skip + saved = save_images(api, api.post("/images/edits", body), prompt, path) + return f"Saved the edited image: {saved[0]}" + + def finish_video(request_id: str, prompt: str, path: str, wait: float) -> str: + deadline = time.monotonic() + wait + while True: + state = api.request("GET", f"/videos/{request_id}").json() + status = state.get("status") + if status == "done": + video = state.get("video") or {} + target = api.target(path, prompt or request_id, ".mp4") + target.write_bytes(api.download(video["url"])) + return f"Saved the video ({video.get('duration', '?')} s, {state.get('model', '')}): {target}" + if status in ("failed", "expired"): + raise ToolFailure(f"xAI couldn't make the video ({status}): {state.get('error') or state}") + if time.monotonic() > deadline: + return ( + f"The video isn't ready yet (request {request_id}). Check later with " + f'video_status(request_id="{request_id}"), which downloads it when it\'s done.' + ) + time.sleep(POLL) + + @server.tool( + "generate_video", + "Generate a video with Grok Imagine, from a description and optionally a start image, and save it as " + "MP4. Takes a minute or more. Costs $0.02–0.08 per second of video.", + { + "prompt": {"type": "string", "description": "What happens in the video"}, + "image": {"type": "string", "description": "A local image to start from (optional)"}, + "duration": {"type": "integer", "description": "Seconds, 1–15 (default 8)"}, + "aspect_ratio": { + "type": "string", + "enum": [r for r in ASPECT_RATIOS if r not in ("auto", "2:1", "1:2", "21:9")], + }, + "resolution": {"type": "string", "enum": ["480p", "720p", "1080p"], "description": "Default 720p"}, + "model": {"type": "string", "enum": VIDEO_MODELS, "description": f"Default {VIDEO_MODELS[0]}"}, + "path": path_arg, + }, + ["prompt"], + ) + def generate_video(prompt: str, image: str = "", duration: int = 8, aspect_ratio: str = "16:9", + resolution: str = "720p", model: str = VIDEO_MODELS[0], path: str = "") -> str: # fmt: skip + body: dict = {"model": model, "prompt": prompt, "duration": max(1, min(15, int(duration))), + "aspect_ratio": aspect_ratio, "resolution": resolution} # fmt: skip + if image: + body["image"] = {"url": data_url(image)} + request_id = api.post("/videos/generations", body).get("request_id") + if not request_id: + raise ToolFailure("xAI didn't start the video") + return finish_video(str(request_id), prompt, path, VIDEO_WAIT) + + @server.tool( + "video_status", + "Check a video that generate_video started and that wasn't ready yet; saves it when it's done.", + {"request_id": {"type": "string"}, "path": path_arg}, + ["request_id"], + ) + def video_status(request_id: str, path: str = "") -> str: + return finish_video(request_id, "", path, 0) + + @server.tool( + "text_to_speech", + "Turn text into speech with an xAI voice and save it as an audio file. Billed per character.", + { + "text": {"type": "string", "description": "What to say (up to 60,000 characters)"}, + "voice": {"type": "string", "enum": VOICES, "description": "Default eve"}, + "language": {"type": "string", "description": "BCP-47 code such as en, de, fr, ja, or auto (default)"}, + "format": {"type": "string", "enum": list(AUDIO_FORMATS), "description": "Default mp3"}, + "speed": {"type": "number", "description": "1.0 is normal"}, + "path": path_arg, + }, + ["text"], + ) + def text_to_speech(text: str, voice: str = "eve", language: str = "auto", format: str = "mp3", + speed: float = 1.0, path: str = "") -> str: # fmt: skip + codec = format if format in AUDIO_FORMATS else "mp3" + body = {"text": text, "voice_id": voice.lower(), "language": language or "auto", + "output_format": {"codec": codec}, "speed": float(speed or 1.0)} # fmt: skip + r = api.request("POST", "/tts", json=body) + duration = "" + if "json" in r.headers.get("Content-Type", ""): + data = r.json() + audio = base64.b64decode(data.get("audio", "")) + duration = f", {data['duration']:.1f} s" if isinstance(data.get("duration"), (int, float)) else "" + else: + audio = r.content + if not audio: + raise ToolFailure("xAI returned no audio") + target = api.target(path, text[:40], AUDIO_FORMATS[codec]) + target.write_bytes(audio) + return f"Saved the speech ({voice}{duration}): {target}" + + return server + + +def main(argv: list[str] | None = None) -> None: + parser = argparse.ArgumentParser( + prog="lcode-mcp-xai", description="MCP server for xAI's image, video and speech APIs (stdio)." + ) + parser.add_argument("--output", default=os.environ.get("LCODE_XAI_OUTPUT", "generated"), + help="folder for the files it makes (default: generated/ in the project)") # fmt: skip + args = parser.parse_args(argv) + key = os.environ.get("XAI_API_KEY") + if not key: + parser.error("set XAI_API_KEY to an xAI API key (https://console.x.ai)") + build(XAI(key, os.environ.get("XAI_API_BASE", API), Path(args.output))).serve() + + +if __name__ == "__main__": + main() diff --git a/tests/test_xai.py b/tests/test_xai.py new file mode 100644 index 0000000..8c91863 --- /dev/null +++ b/tests/test_xai.py @@ -0,0 +1,163 @@ +import base64 +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import ClassVar + +import pytest + +from lcode.mcp import catalog +from lcode.servers import xai + +PNG = b"\x89PNG\r\n\x1a\n" + b"pixels" +JPEG = b"\xff\xd8\xff\xe0" + b"jpeg" +MP4 = b"\x00\x00\x00\x18ftypmp42video" +MP3 = b"ID3" + b"sound" + + +class FakeXAI(BaseHTTPRequestHandler): + """xAI's image, video and speech endpoints, as far as lcode-mcp-xai uses them.""" + + requests: ClassVar[list] = [] + polls: ClassVar[int] = 0 + video_ready_after: ClassVar[int] = 1 + + def log_message(self, *args): + pass + + def send(self, status, body, content_type="application/json"): + data = body if isinstance(body, bytes) else json.dumps(body).encode() + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def authorized(self): + if self.headers.get("Authorization") != "Bearer good-key": + self.send(401, {"error": {"message": "Incorrect API key provided"}}) + return False + return True + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + type(self).requests.append((self.path, body)) + if not self.authorized(): + return + if self.path == "/v1/images/generations": + if "forbidden" in body["prompt"]: + self.send(400, {"error": {"message": "Generated image rejected by content moderation."}}) + return + images = [{"b64_json": base64.b64encode(PNG).decode()} for _ in range(body.get("n", 1))] + self.send(200, {"data": images}) + elif self.path == "/v1/images/edits": + self.send(200, {"data": [{"url": f"http://{self.headers['Host']}/files/edited.jpg"}]}) + elif self.path == "/v1/videos/generations": + self.send(200, {"request_id": "req-1"}) + elif self.path == "/v1/tts": + self.send(200, {"audio": base64.b64encode(MP3).decode(), "content_type": "audio/mpeg", "duration": 1.5}) + else: + self.send(404, {"error": "no such endpoint"}) + + def do_GET(self): + if self.path == "/files/edited.jpg": + self.send(200, JPEG, "image/jpeg") + elif self.path == "/files/video.mp4": + self.send(200, MP4, "video/mp4") + elif self.path == "/v1/videos/req-1" and self.authorized(): + type(self).polls += 1 + if type(self).polls <= type(self).video_ready_after: + self.send(200, {"status": "pending"}) + else: + url = f"http://{self.headers['Host']}/files/video.mp4" + video = {"url": url, "duration": 6} + self.send(200, {"status": "done", "video": video, "model": "grok-imagine-video-1.5"}) + + +@pytest.fixture +def api(tmp_path, monkeypatch): + FakeXAI.requests, FakeXAI.polls, FakeXAI.video_ready_after = [], 0, 1 + monkeypatch.setattr(xai, "POLL", 0.01) + httpd = ThreadingHTTPServer(("127.0.0.1", 0), FakeXAI) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + yield xai.XAI("good-key", f"http://127.0.0.1:{httpd.server_address[1]}/v1", tmp_path / "generated") + httpd.shutdown() + + +def call(server, name, **arguments): + result = server.call(name, arguments) + return result["content"][0]["text"], result["isError"] + + +def test_images_are_generated_and_saved(api, tmp_path): + server = xai.build(api) + assert [t.name for t in server.listed()] == [ + "generate_image", "edit_image", "generate_video", "video_status", "text_to_speech"] # fmt: skip + text, error = call(server, "generate_image", prompt="A red fox in the snow", n=2, aspect_ratio="16:9") + assert not error and text.startswith("Saved 2 image(s) from grok-imagine-image-2.0:") + saved = sorted((tmp_path / "generated").glob("a-red-fox-in-the-snow-*.png")) + assert len(saved) == 2 and all(p.read_bytes() == PNG for p in saved) + path, body = FakeXAI.requests[0] + assert path == "/v1/images/generations" and body == { + "model": "grok-imagine-image-2.0", "prompt": "A red fox in the snow", "n": 2, + "response_format": "b64_json", "aspect_ratio": "16:9"} # fmt: skip + text, _ = call(server, "generate_image", prompt="logo", path=str(tmp_path / "assets" / "logo")) + assert text.endswith(f"- {tmp_path / 'assets' / 'logo.png'}") # the right extension is added + text, _ = call(server, "generate_image", prompt="logo", path=str(tmp_path / "assets" / "logo.png")) + assert text.endswith(f"- {tmp_path / 'assets' / 'logo-2.png'}") # nothing is overwritten + + +def test_an_image_is_edited_from_a_local_file(api, tmp_path): + source = tmp_path / "photo.png" + source.write_bytes(PNG) + text, error = call(xai.build(api), "edit_image", image=str(source), prompt="Make it a pencil sketch") + assert not error and text.startswith("Saved the edited image:") and text.endswith(".jpg") + _, body = FakeXAI.requests[0] + assert body["image"] == {"url": "data:image/png;base64," + base64.b64encode(PNG).decode(), "type": "image_url"} + text, error = call(xai.build(api), "edit_image", image=str(tmp_path / "missing.png"), prompt="x") + assert error and "doesn't exist" in text + + +def test_a_video_is_generated_waited_for_and_downloaded(api, tmp_path, monkeypatch): + start = tmp_path / "first.jpg" + start.write_bytes(JPEG) + server = xai.build(api) + text, error = call(server, "generate_video", prompt="The fox runs away", image=str(start), duration=6) + assert not error and text.startswith("Saved the video (6 s, grok-imagine-video-1.5):") and text.endswith(".mp4") + _, body = FakeXAI.requests[0] + assert body["duration"] == 6 and body["image"]["url"].startswith("data:image/jpeg;base64,") + assert next((tmp_path / "generated").glob("*.mp4")).read_bytes() == MP4 and FakeXAI.polls == 2 + FakeXAI.polls, FakeXAI.video_ready_after = 0, 10**6 + monkeypatch.setattr(xai, "VIDEO_WAIT", 0) + text, _ = call(server, "generate_video", prompt="slow") + assert "isn't ready yet (request req-1)" in text and 'video_status(request_id="req-1")' in text + FakeXAI.video_ready_after = 0 + text, _ = call(server, "video_status", request_id="req-1", path=str(tmp_path / "clip.mp4")) + assert text.endswith(str(tmp_path / "clip.mp4")) and (tmp_path / "clip.mp4").read_bytes() == MP4 + + +def test_speech_is_saved_as_audio(api, tmp_path): + text, error = call(xai.build(api), "text_to_speech", text="Hello from lcode", voice="Leo", language="en") + assert not error and text.startswith("Saved the speech (Leo, 1.5 s):") and text.endswith(".mp3") + _, body = FakeXAI.requests[0] + assert body == {"text": "Hello from lcode", "voice_id": "leo", "language": "en", "output_format": {"codec": "mp3"}, + "speed": 1.0} # fmt: skip + assert next((tmp_path / "generated").glob("hello-from-lcode-*.mp3")).read_bytes() == MP3 + + +def test_xai_errors_reach_the_model(api, tmp_path): + text, error = call(xai.build(api), "generate_image", prompt="something forbidden") + assert error and text == "xAI error 400: Generated image rejected by content moderation." + bad = xai.XAI("wrong", api.base, tmp_path) + text, error = call(xai.build(bad), "text_to_speech", text="hi") + assert error and "xAI refused the request (401): Incorrect API key provided" in text and "XAI_API_KEY" in text + + +def test_the_preset_asks_for_the_key(): + preset = catalog.load()["xai"] + [key] = preset.inputs + assert key.var == "XAI_API_KEY" and key.secret + server = catalog.build(preset, {"XAI_API_KEY": "typed-" + "key"}) + assert server["env"] == {"XAI_API_KEY": "typed-key"} and server["args"][-1] == "lcode-mcp-xai" + assert catalog.build(preset, {"XAI_API_KEY": None})["env"] == {"XAI_API_KEY": "${XAI_API_KEY}"} + assert server["timeout"] == 900