Anthropic-compatible Skills with upload

Anthropic-compatible Skills with upload.

Start the server with shell execution enabled: mistralrs serve —agent -p 1234 -m Qwen/Qwen3-4B

Then run: python3 examples/server/anthropic_skills.py

"""
Anthropic-compatible Skills with upload.

Start the server with shell execution enabled:
    mistralrs serve --agent -p 1234 -m Qwen/Qwen3-4B

Then run:
    python3 examples/server/anthropic_skills.py
"""

from pathlib import Path
import json
import os
import tempfile
import urllib.request
import uuid
import zipfile


BASE_URL = os.environ.get("MISTRALRS_BASE_URL", "http://localhost:1234")
API_KEY = os.environ.get("ANTHROPIC_API_KEY", "local")


def write_sample_skill(root: Path) -> Path:
    skill_dir = root / "invoice-auditor"
    skill_dir.mkdir()
    (skill_dir / "SKILL.md").write_text(
        """---
name: invoice-auditor
description: Checks invoice line items and totals with a bundled Python helper.
---

# Invoice Auditor

Use this command to validate the bundled invoice and write the audit note:

```bash
python3 skills/invoice-auditor/check_invoice.py skills/invoice-auditor/invoice.csv audit.txt
```

Report whether the declared total matches the sum of the line items.
""",
        encoding="utf-8",
    )
    (skill_dir / "invoice.csv").write_text(
        """item,amount
hosting,25.00
storage,12.50
support,17.50
declared_total,55.00
""",
        encoding="utf-8",
    )
    (skill_dir / "check_invoice.py").write_text(
        """import csv
import sys


with open(sys.argv[1], newline="") as handle:
    rows = list(csv.DictReader(handle))

declared = float(rows[-1]["amount"])
line_total = sum(float(row["amount"]) for row in rows[:-1])
status = "match" if line_total == declared else "mismatch"
note = f"line_total={line_total:.2f}\\ndeclared_total={declared:.2f}\\nstatus={status}\\n"
print(note, end="")

if len(sys.argv) > 2:
    with open(sys.argv[2], "w", encoding="utf-8") as handle:
        handle.write(note)
""",
        encoding="utf-8",
    )
    return skill_dir


def zip_skill(skill_dir: Path, zip_path: Path) -> None:
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive:
        for path in skill_dir.rglob("*"):
            archive.write(path, path.relative_to(skill_dir.parent))


def request_json(path: str, payload: dict) -> dict:
    request = urllib.request.Request(
        f"{BASE_URL}{path}",
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "content-type": "application/json",
            "x-api-key": API_KEY,
            "anthropic-version": "2023-06-01",
            "anthropic-beta": "code-execution-2025-08-25,skills-2025-10-02",
        },
        method="POST",
    )
    with urllib.request.urlopen(request) as response:
        return json.loads(response.read().decode("utf-8"))


def upload_skill(zip_path: Path) -> dict:
    boundary = f"----mistralrs-{uuid.uuid4().hex}"
    body = b"".join(
        [
            f"--{boundary}\r\n".encode("utf-8"),
            (
                f'Content-Disposition: form-data; name="file"; filename="{zip_path.name}"\r\n'
                "Content-Type: application/zip\r\n\r\n"
            ).encode("utf-8"),
            zip_path.read_bytes(),
            f"\r\n--{boundary}--\r\n".encode("utf-8"),
        ]
    )
    request = urllib.request.Request(
        f"{BASE_URL}/v1/skills",
        data=body,
        headers={
            "content-type": f"multipart/form-data; boundary={boundary}",
            "x-api-key": API_KEY,
            "anthropic-version": "2023-06-01",
            "anthropic-beta": "skills-2025-10-02",
        },
        method="POST",
    )
    with urllib.request.urlopen(request) as response:
        return json.loads(response.read().decode("utf-8"))


def download_file(file_id: str, output_path: Path) -> None:
    request = urllib.request.Request(
        f"{BASE_URL}/v1/files/{file_id}/content",
        headers={
            "x-api-key": API_KEY,
            "anthropic-version": "2023-06-01",
            "anthropic-beta": "files-api-2025-04-14",
        },
    )
    with urllib.request.urlopen(request) as response:
        output_path.write_bytes(response.read())


with tempfile.TemporaryDirectory() as temp_dir:
    temp_path = Path(temp_dir)
    zip_path = temp_path / "invoice-auditor.zip"
    zip_skill(write_sample_skill(temp_path), zip_path)
    skill = upload_skill(zip_path)

print(f"Uploaded skill: {skill['id']} ({skill['display_title']})")

response = request_json(
    "/v1/messages",
    {
        "model": "default",
        "max_tokens": 1024,
        "agent_permission": "auto",
        "max_tool_rounds": 6,
        "container": {
            "skills": [
                {"type": "custom", "skill_id": skill["id"], "version": "latest"}
            ]
        },
        "tools": [{"type": "code_execution_20250825", "name": "code_execution"}],
        "messages": [
            {
                "role": "user",
                "content": (
                    "Use the uploaded invoice-auditor skill. Run its helper, "
                    "write audit.txt, and summarize the result."
                ),
            }
        ],
        "files": [{"name": "audit.txt", "format": "txt"}],
    },
)

print("\nMessage response:")
for block in response["content"]:
    if block["type"] == "text":
        print(block["text"])

for file in response.get("files", []):
    output_path = Path(file["name"])
    download_file(file["id"], output_path)
    print(f"Downloaded: {output_path}")

Source: examples/server/anthropic_skills.py