feat: add CAMB AI tool integration for speech & audio

Author: neilruaro-camb

python/packages/autogen-ext/pyproject.toml

@@ -41,6 +41,7 @@ llama-cpp = [
     "llama-cpp-python>=0.3.8",
 ]
 
+camb = ["camb-sdk>=1.0.0"]
 graphrag = ["graphrag>=2.3.0"]
 chromadb = ["chromadb>=1.0.0"]
 mem0 = ["mem0ai>=0.1.98"]

python/packages/autogen-ext/src/autogen_ext/tools/camb/__init__.py

@@ -0,0 +1,31 @@
+from ._audio_separation import AudioSeparationArgs, CambAudioSeparationTool
+from ._config import CambToolConfig
+from ._text_to_sound import CambTextToSoundTool, TextToSoundArgs
+from ._toolkit import CambAIToolkit
+from ._transcription import CambTranscriptionTool, TranscriptionArgs
+from ._translated_tts import CambTranslatedTTSTool, TranslatedTTSArgs
+from ._translation import CambTranslationTool, TranslationArgs
+from ._tts import CambTTSTool, TTSArgs
+from ._voice_clone import CambVoiceCloneTool, VoiceCloneArgs
+from ._voice_list import CambVoiceListTool, VoiceListArgs
+
+__all__ = [
+    "AudioSeparationArgs",
+    "CambAIToolkit",
+    "CambAudioSeparationTool",
+    "CambTextToSoundTool",
+    "CambToolConfig",
+    "CambTranscriptionTool",
+    "CambTranslatedTTSTool",
+    "CambTranslationTool",
+    "CambTTSTool",
+    "CambVoiceCloneTool",
+    "CambVoiceListTool",
+    "TextToSoundArgs",
+    "TranscriptionArgs",
+    "TranslatedTTSArgs",
+    "TranslationArgs",
+    "TTSArgs",
+    "VoiceCloneArgs",
+    "VoiceListArgs",
+]

python/packages/autogen-ext/src/autogen_ext/tools/camb/_audio_separation.py

@@ -0,0 +1,131 @@
+import json
+from typing import Optional
+
+from autogen_core import CancellationToken
+from pydantic import BaseModel, Field, model_validator
+from typing_extensions import Self
+
+from ._base import CambBaseTool
+from ._config import CambToolConfig
+
+
+class AudioSeparationArgs(BaseModel):
+    """Arguments for the CAMB.AI audio separation tool."""
+
+    audio_url: Optional[str] = Field(
+        default=None,
+        description="URL of the audio file to separate.",
+    )
+    audio_file_path: Optional[str] = Field(
+        default=None,
+        description="Local file path of the audio file to separate.",
+    )
+
+    @model_validator(mode="after")
+    def _validate_audio_source(self) -> "AudioSeparationArgs":
+        if not self.audio_url and not self.audio_file_path:
+            raise ValueError("Either audio_url or audio_file_path must be provided.")
+        if self.audio_url and self.audio_file_path:
+            raise ValueError("Only one of audio_url or audio_file_path should be provided.")
+        return self
+
+
+class CambAudioSeparationTool(CambBaseTool[AudioSeparationArgs, str]):
+    """Audio separation tool using CAMB.AI.
+
+    Separates vocals from background audio using the CAMB.AI audio separation API.
+    Returns a JSON string with vocals and background URLs.
+
+    .. note::
+        This tool requires the :code:`camb` extra for the :code:`autogen-ext` package.
+
+        To install:
+
+        .. code-block:: bash
+
+            pip install -U "autogen-agentchat" "autogen-ext[camb]"
+
+    Example usage:
+
+    .. code-block:: python
+
+        import asyncio
+        from autogen_core import CancellationToken
+        from autogen_ext.tools.camb import CambAudioSeparationTool, AudioSeparationArgs
+
+        async def main():
+            tool = CambAudioSeparationTool(api_key="your-api-key")
+            result = await tool.run(
+                AudioSeparationArgs(audio_file_path="/path/to/audio.mp3"),
+                CancellationToken(),
+            )
+            print(f"Separation result: {result}")
+
+        asyncio.run(main())
+    """
+
+    component_provider_override = "autogen_ext.tools.camb.CambAudioSeparationTool"
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        base_url: Optional[str] = None,
+        timeout: Optional[float] = None,
+        max_poll_attempts: int = 60,
+        poll_interval: float = 2.0,
+    ) -> None:
+        super().__init__(
+            args_type=AudioSeparationArgs,
+            return_type=str,
+            name="camb_audio_separation",
+            description=(
+                "Separate vocals from background audio using CAMB.AI. "
+                "Returns JSON with vocals and background URLs."
+            ),
+            api_key=api_key,
+            base_url=base_url,
+            timeout=timeout,
+            max_poll_attempts=max_poll_attempts,
+            poll_interval=poll_interval,
+        )
+
+    async def run(self, args: AudioSeparationArgs, cancellation_token: CancellationToken) -> str:
+        client = self._get_client()
+
+        kwargs: dict = {}
+        if args.audio_url:
+            kwargs["media_url"] = args.audio_url
+        elif args.audio_file_path:
+            kwargs["media_file"] = open(args.audio_file_path, "rb")
+
+        try:
+            task = await client.audio_separation.create_audio_separation(**kwargs)
+        finally:
+            if "media_file" in kwargs:
+                kwargs["media_file"].close()
+
+        task_id = task.task_id
+
+        status = await self._poll_task_status(
+            client.audio_separation.get_audio_separation_status,
+            task_id,
+        )
+
+        run_id = status.run_id
+        result = await client.audio_separation.get_audio_separation_run_info(run_id)
+
+        output = {
+            "foreground_audio_url": getattr(result, "foreground_audio_url", None),
+            "background_audio_url": getattr(result, "background_audio_url", None),
+        }
+        return json.dumps(output)
+
+    @classmethod
+    def _from_config(cls, config: CambToolConfig) -> Self:
+        return cls(
+            api_key=config.api_key,
+            base_url=config.base_url,
+            timeout=config.timeout,
+            max_poll_attempts=config.max_poll_attempts,
+            poll_interval=config.poll_interval,
+        )

python/packages/autogen-ext/src/autogen_ext/tools/camb/_base.py

@@ -0,0 +1,166 @@
+import asyncio
+import os
+import struct
+import tempfile
+from abc import abstractmethod
+from typing import Any, Generic, Optional, TypeVar
+
+from autogen_core import CancellationToken, Component
+from autogen_core.tools import BaseTool
+from pydantic import BaseModel
+from typing_extensions import Self
+
+from ._config import CambToolConfig
+
+ArgsT = TypeVar("ArgsT", bound=BaseModel)
+ReturnT = TypeVar("ReturnT")
+
+
+class CambBaseTool(BaseTool[ArgsT, ReturnT], Component[CambToolConfig], Generic[ArgsT, ReturnT]):
+    """Abstract base class for CAMB.AI tools.
+
+    Manages the AsyncCambAI client lifecycle and provides shared utilities
+    for polling async tasks, saving audio, and detecting audio formats.
+    Uses the ``camb-sdk`` package with its native async client.
+    """
+
+    component_type = "tool"
+    component_config_schema = CambToolConfig
+
+    def __init__(
+        self,
+        args_type: type[ArgsT],
+        return_type: type[ReturnT],
+        name: str,
+        description: str,
+        api_key: Optional[str] = None,
+        base_url: Optional[str] = None,
+        timeout: Optional[float] = None,
+        max_poll_attempts: int = 60,
+        poll_interval: float = 2.0,
+    ) -> None:
+        super().__init__(
+            args_type=args_type,
+            return_type=return_type,
+            name=name,
+            description=description,
+        )
+        self._api_key = api_key
+        self._base_url = base_url
+        self._timeout = timeout
+        self._max_poll_attempts = max_poll_attempts
+        self._poll_interval = poll_interval
+        self._client: Any = None
+
+    def _get_api_key(self) -> str:
+        """Resolve API key from parameter or environment variable."""
+        key = self._api_key or os.environ.get("CAMB_API_KEY")
+        if not key:
+            raise ValueError(
+                "CAMB.AI API key is required. Provide it via the api_key parameter "
+                "or set the CAMB_API_KEY environment variable."
+            )
+        return key
+
+    def _get_client(self) -> Any:
+        """Get or create the AsyncCambAI client (lazy initialization)."""
+        if self._client is None:
+            from camb.client import AsyncCambAI
+
+            kwargs: dict[str, Any] = {"api_key": self._get_api_key()}
+            if self._base_url:
+                kwargs["base_url"] = self._base_url
+            if self._timeout is not None:
+                kwargs["timeout"] = self._timeout
+            self._client = AsyncCambAI(**kwargs)
+        return self._client
+
+    async def _poll_task_status(
+        self,
+        status_func: Any,
+        task_id: str,
+    ) -> Any:
+        """Poll an async task until completion or failure.
+
+        Args:
+            status_func: Async function to call for status checks (e.g. client.transcription.get_transcription_task_status).
+            task_id: The task ID to poll.
+
+        Returns:
+            The final status result when the task completes.
+
+        Raises:
+            RuntimeError: If the task fails or times out.
+        """
+        for _ in range(self._max_poll_attempts):
+            result = await status_func(task_id)
+            status = getattr(result, "status", None)
+            if status is None and hasattr(result, "message"):
+                status = getattr(result.message, "status", None)
+            if status in ("SUCCESS", "complete", "completed"):
+                return result
+            if status in ("ERROR", "TIMEOUT", "PAYMENT_REQUIRED", "failed", "error"):
+                reason = getattr(result, "exception_reason", "") or ""
+                raise RuntimeError(f"CAMB.AI task failed with status: {status}. {reason}")
+            await asyncio.sleep(self._poll_interval)
+        raise RuntimeError(
+            f"CAMB.AI task timed out after {self._max_poll_attempts * self._poll_interval}s"
+        )
+
+    @staticmethod
+    def _detect_audio_format(data: bytes) -> str:
+        """Detect audio format from raw bytes."""
+        if data[:4] == b"RIFF":
+            return "wav"
+        if data[:3] == b"ID3" or data[:2] == b"\xff\xfb":
+            return "mp3"
+        if data[:4] == b"fLaC":
+            return "flac"
+        if data[:4] == b"OggS":
+            return "ogg"
+        return "wav"
+
+    @staticmethod
+    def _add_wav_header(
+        raw_data: bytes, sample_rate: int = 24000, channels: int = 1, bits_per_sample: int = 16
+    ) -> bytes:
+        """Add a WAV header to raw PCM audio data."""
+        data_size = len(raw_data)
+        header = struct.pack(
+            "<4sI4s4sIHHIIHH4sI",
+            b"RIFF",
+            36 + data_size,
+            b"WAVE",
+            b"fmt ",
+            16,
+            1,  # PCM format
+            channels,
+            sample_rate,
+            sample_rate * channels * bits_per_sample // 8,
+            channels * bits_per_sample // 8,
+            bits_per_sample,
+            b"data",
+            data_size,
+        )
+        return header + raw_data
+
+    @staticmethod
+    def _save_audio(data: bytes, extension: str = "wav") -> str:
+        """Save audio data to a temporary file and return the file path."""
+        with tempfile.NamedTemporaryFile(suffix=f".{extension}", delete=False) as f:
+            f.write(data)
+            return f.name
+
+    def _to_config(self) -> CambToolConfig:
+        return CambToolConfig(
+            api_key=self._api_key,
+            base_url=self._base_url,
+            timeout=self._timeout,
+            max_poll_attempts=self._max_poll_attempts,
+            poll_interval=self._poll_interval,
+        )
+
+    @classmethod
+    @abstractmethod
+    def _from_config(cls, config: CambToolConfig) -> Self:
+        ...

python/packages/autogen-ext/src/autogen_ext/tools/camb/_config.py

@@ -0,0 +1,21 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class CambToolConfig(BaseModel):
+    """Configuration for CAMB.AI tools.
+
+    Args:
+        api_key: CAMB.AI API key. If not provided, falls back to CAMB_API_KEY environment variable.
+        base_url: Base URL for the CAMB.AI API.
+        timeout: Request timeout in seconds.
+        max_poll_attempts: Maximum number of polling attempts for async tasks.
+        poll_interval: Interval between polling attempts in seconds.
+    """
+
+    api_key: Optional[str] = None
+    base_url: Optional[str] = None
+    timeout: Optional[float] = None
+    max_poll_attempts: int = 60
+    poll_interval: float = 2.0