Add an FFmpeg Tool to Your AI Agent (Claude Agent SDK, OpenAI Agents SDK, LangChain)

import { query } from '@anthropic-ai/claude-agent-sdk'

for await (const message of query({
  prompt: 'Make a 720p MP4 of https://example.com/talk.mov and give me the link.',
  options: {
    mcpServers: {
      xora: {
        type: 'http',
        url: 'https://api.xora.sh/v1/mcp',
        headers: { Authorization: `Bearer ${process.env.XORA_API_KEY}` },
      },
    },
    allowedTools: ['mcp__xora__*'],
  },
})) {
  if (message.type === 'result' && message.subtype === 'success') console.log(message.result)
}

import asyncio, os
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

options = ClaudeAgentOptions(
    mcp_servers={
        "xora": {
            "type": "http",
            "url": "https://api.xora.sh/v1/mcp",
            "headers": {"Authorization": f"Bearer {os.environ['XORA_API_KEY']}"},
        }
    },
    allowed_tools=["mcp__xora__*"],
)

async def main():
    async for message in query(
        prompt="Make a 720p MP4 of https://example.com/talk.mov and give me the link.",
        options=options,
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

import asyncio, os
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async def main() -> None:
    async with MCPServerStreamableHttp(
        name="xora",
        params={
            "url": "https://api.xora.sh/v1/mcp",
            "headers": {"Authorization": f"Bearer {os.environ['XORA_API_KEY']}"},
        },
        cache_tools_list=True,
    ) as xora:
        agent = Agent(
            name="Video assistant",
            instructions="Use the Xora tools to process video. Poll get_job until the job ends.",
            mcp_servers=[xora],
        )
        result = await Runner.run(agent, "Make a 720p MP4 of https://example.com/talk.mov and give me the link.")
        print(result.final_output)

asyncio.run(main())

import asyncio, os
from fastmcp.client import Client
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter

async def main():
    client = Client("https://api.xora.sh/v1/mcp", auth=os.environ["XORA_API_KEY"])
    async with MCPAdapter(client) as adapter:
        tools = await adapter.list_tools()
        agent = create_agent("anthropic:claude-sonnet-5-5", tools)
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "Make a 720p MP4 of https://example.com/talk.mov and give me the link."}]}
        )
        print(result["messages"][-1].content)

asyncio.run(main())

import os, time, requests

API = "https://api.xora.sh/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['XORA_API_KEY']}"}

def run_ffmpeg(input_url: str, args: list[str], output_name: str) -> dict:
    """Run FFmpeg args on input_url. Use {{in_1}} for the input and {{out_1}} for the output."""
    job = requests.post(f"{API}/jobs", headers=HEADERS, json={
        "mode": "ffmpeg",
        "input_files": {"in_1": input_url},
        "output_files": {"out_1": output_name},
        "ffmpeg": {"args": args},
    }).json()
    while True:
        status = requests.get(f"{API}/jobs/{job['id']}", headers=HEADERS).json()
        if status["state"] in ("completed", "failed", "rejected", "cancelled"):
            return status
        time.sleep(3)

You can process video and audio with the Xora tools.
- Before you write an FFmpeg command for a file you have not seen, run create_job with mode "probe".
- Prefer a recipe (call list_recipes once) when one fits. Use mode "ffmpeg" with an args array for anything else.
- Output file names use letters, numbers and underscores, plus an extension.
- Send an idempotencyKey with every create_job, and reuse it if you retry the same job.
- Poll get_job until the state is completed, failed, rejected or cancelled.
- If a job fails, read error.code and error.message, fix the request, and try again once.