Batch API

BatchClient performs provider API operations. BatchStore persists job state in SQLite. BatchManager combines them with named callbacks and monitoring. These are delayed jobs, not ordinary streaming generations. See Batch Processing for usage.

Batch support is limited to the OpenAI/Azure and Anthropic routes implemented by the batch client; do not infer support from a provider's text-generation support. Endpoint availability, quotas, and pricing are provider-specific. Register callbacks again before resuming stored jobs, and close the manager when finished. The reference follows the current source checkout.

nagents.BatchManager

High-level batch processing manager with persistence.

Combines BatchClient (API operations) with BatchStore (persistence) to provide a complete batch processing solution that survives agent restarts.

Features: - Automatic persistence of batch jobs - Named callback handlers for result processing - Resume pending batches on startup - Background monitoring of active batches

Source code in src/nagents/batch/manager.py

class BatchManager:
    """
    High-level batch processing manager with persistence.

    Combines BatchClient (API operations) with BatchStore (persistence)
    to provide a complete batch processing solution that survives
    agent restarts.

    Features:
    - Automatic persistence of batch jobs
    - Named callback handlers for result processing
    - Resume pending batches on startup
    - Background monitoring of active batches
    """

    def __init__(
        self,
        api_key: str,
        provider_type: ProviderType,
        model: str,
        db_path: Path,
        *,
        base_url: str | None = None,
        poll_interval: float = 60.0,
    ):
        """
        Initialize the batch manager.

        Args:
            api_key: API key for the provider
            provider_type: OPENAI_COMPATIBLE or ANTHROPIC
            model: Default model for batch requests
            db_path: Path to SQLite database for persistence
            base_url: Optional custom base URL
            poll_interval: Seconds between status checks (default 60)
        """
        self.api_key = api_key
        self.provider_type = provider_type
        self.model = model
        self.base_url = base_url
        self.poll_interval = poll_interval

        # Initialize components
        self._store = BatchStore(db_path)
        self._client = BatchClient(
            provider_type=provider_type,
            api_key=api_key,
            model=model,
            base_url=base_url,
        )

        # Callback registry
        self._callbacks: dict[str, BatchCallback] = {}

        # Background task tracking
        self._monitor_tasks: dict[str, asyncio.Task[None]] = {}
        self._initialized = False

    async def initialize(self) -> None:
        """Initialize the store and resume pending batches."""
        if self._initialized:
            return

        await self._store.initialize()
        self._initialized = True
        logger.info("BatchManager initialized")

    def register_callback(
        self,
        name: str,
    ) -> Callable[[BatchCallback], BatchCallback]:
        """
        Decorator to register a callback handler.

        Args:
            name: Unique name for this callback

        Example:
            @manager.register_callback("process_summaries")
            async def handler(results: list[BatchResult], context: dict):
                for result in results:
                    save_summary(result.custom_id, result.content)
        """

        def decorator(func: BatchCallback) -> BatchCallback:
            self._callbacks[name] = func
            logger.debug(f"Registered batch callback: {name}")
            return func

        return decorator

    def add_callback(self, name: str, callback: BatchCallback) -> None:
        """
        Add a callback handler directly.

        Args:
            name: Unique name for this callback
            callback: Async function to handle results
        """
        self._callbacks[name] = callback
        logger.debug(f"Added batch callback: {name}")

    async def create_batch(
        self,
        requests: list[BatchRequest],
        *,
        config: BatchConfig | None = None,
        tools: list[ToolDefinition] | None = None,
        generation_config: GenerationConfig | None = None,
        callback_name: str | None = None,
        callback_context: dict[str, Any] | None = None,
        store_requests: bool = True,
        auto_monitor: bool = True,
    ) -> BatchJob:
        """
        Create a batch job with persistence.

        Args:
            requests: List of BatchRequest objects
            config: Optional batch configuration
            tools: Optional default tools for all requests
            generation_config: Optional default generation config
            callback_name: Name of registered callback for results
            callback_context: Context dict passed to callback
            store_requests: Whether to store requests for retry (default True)
            auto_monitor: Whether to start background monitoring (default True)

        Returns:
            The created BatchJob
        """
        await self.initialize()

        # Validate callback if provided
        if callback_name and callback_name not in self._callbacks:
            raise ValueError(f"Unknown callback: {callback_name}. Available: {list(self._callbacks.keys())}")

        # Create batch via API
        job = await self._client.create_batch(requests, config, tools, generation_config)

        # Persist to database
        await self._store.save_batch(
            job,
            requests=requests if store_requests else None,
            model=self.model,
            base_url=self.base_url or "",
            callback_name=callback_name,
            callback_context=callback_context,
        )

        logger.info(f"Created batch {job.id} with {len(requests)} requests")

        # Start background monitoring
        if auto_monitor:
            self._start_monitor(job.id)

        return job

    async def get_batch(self, batch_id: str) -> BatchJob | None:
        """
        Get a batch job by ID.

        Fetches from API and updates local store.

        Args:
            batch_id: The batch ID

        Returns:
            Updated BatchJob or None if not found
        """
        await self.initialize()

        try:
            job = await self._client.get_batch(batch_id)
            await self._store.update_status(job)
            return job
        except Exception as e:
            logger.warning(f"Failed to get batch {batch_id} from API: {e}")
            # Fall back to stored data
            return await self._store.get_batch(batch_id)

    async def get_results(self, batch_id: str) -> list[BatchResult]:
        """
        Get results for a completed batch.

        Args:
            batch_id: The batch ID

        Returns:
            List of BatchResult objects
        """
        await self.initialize()

        job = await self._client.get_batch(batch_id)
        if job.status not in (BatchStatus.COMPLETED, BatchStatus.ENDED):
            raise ValueError(f"Batch {batch_id} is not completed (status={job.status})")

        results = []
        async for result in self._client.get_results(job):
            results.append(result)
            # Save individual result
            await self._store.save_request_result(
                batch_id,
                result.custom_id,
                result.result_type,
                content=result.content,
                error=result.error_message,
            )

        return results

    async def cancel_batch(self, batch_id: str) -> BatchJob:
        """
        Cancel a batch job.

        Args:
            batch_id: The batch ID

        Returns:
            Updated BatchJob
        """
        await self.initialize()

        job = await self._client.cancel_batch(batch_id)
        await self._store.update_status(job)

        # Stop monitoring if active
        if batch_id in self._monitor_tasks:
            self._monitor_tasks[batch_id].cancel()
            del self._monitor_tasks[batch_id]

        logger.info(f"Cancelled batch {batch_id}")
        return job

    async def resume_all(self) -> list[str]:
        """
        Resume monitoring all pending batches.

        Call this on startup to resume any batches that were
        submitted before the agent was shut down.

        Returns:
            List of batch IDs that were resumed
        """
        await self.initialize()

        resumed = []

        # Get active batches (still processing)
        active = await self._store.get_active_batches()
        for job in active:
            if job.id not in self._monitor_tasks:
                self._start_monitor(job.id)
                resumed.append(job.id)
                logger.info(f"Resumed monitoring batch {job.id}")

        # Process completed but unhandled batches
        completed = await self._store.get_completed_unprocessed()
        for job in completed:
            await self._process_completed_batch(job.id)
            resumed.append(job.id)
            logger.info(f"Processing completed batch {job.id}")

        return resumed

    async def list_pending(self) -> list[BatchJob]:
        """
        List all pending (unprocessed) batches.

        Returns:
            List of pending BatchJobs
        """
        await self.initialize()
        return await self._store.get_pending_batches()

    async def list_batches(
        self,
        limit: int = 50,
        status: BatchStatus | None = None,
        processed: bool | None = None,
    ) -> list[BatchJob]:
        """
        List batch jobs with optional filtering.

        Args:
            limit: Maximum number of results
            status: Optional filter by status
            processed: Optional filter by processed state

        Returns:
            List of BatchJobs
        """
        await self.initialize()
        return await self._store.list_batches(limit=limit, status=status, processed=processed)

    async def cleanup(self, days: int = 30) -> int:
        """
        Delete old processed batches.

        Args:
            days: Delete batches older than this many days

        Returns:
            Number of batches deleted
        """
        await self.initialize()
        return await self._store.cleanup_old_batches(days)

    async def close(self) -> None:
        """Close the manager and release resources."""
        # Cancel all monitor tasks
        for task in self._monitor_tasks.values():
            task.cancel()
        self._monitor_tasks.clear()

        await self._client.close()

    async def __aenter__(self) -> "BatchManager":
        await self.initialize()
        return self

    async def __aexit__(self, *args: object) -> None:
        await self.close()

    # =========================================================================
    # Private Methods
    # =========================================================================

    def _start_monitor(self, batch_id: str) -> None:
        """Start background monitoring for a batch."""
        if batch_id in self._monitor_tasks:
            return

        task = asyncio.create_task(self._monitor_batch(batch_id))
        self._monitor_tasks[batch_id] = task

    async def _monitor_batch(self, batch_id: str) -> None:
        """Background task to monitor batch until completion."""
        active_statuses = {
            BatchStatus.VALIDATING,
            BatchStatus.IN_PROGRESS,
            BatchStatus.FINALIZING,
        }

        try:
            while True:
                try:
                    job = await self._client.get_batch(batch_id)
                    await self._store.update_status(job)

                    if job.status not in active_statuses:
                        logger.info(f"Batch {batch_id} completed with status {job.status}")
                        await self._process_completed_batch(batch_id)
                        break

                    logger.debug(
                        f"Batch {batch_id}: {job.status.value} "
                        f"({job.request_counts.succeeded}/{job.request_counts.total})"
                    )

                except Exception as e:
                    logger.error(f"Error monitoring batch {batch_id}: {e}")

                await asyncio.sleep(self.poll_interval)

        except asyncio.CancelledError:
            logger.debug(f"Monitor task cancelled for batch {batch_id}")
            raise

        finally:
            # Clean up task reference
            self._monitor_tasks.pop(batch_id, None)

    async def _process_completed_batch(self, batch_id: str) -> None:
        """Process results for a completed batch."""
        try:
            # Get callback info
            callback_name, callback_context = await self._store.get_callback_info(batch_id)

            if callback_name and callback_name in self._callbacks:
                # Fetch results
                results = await self.get_results(batch_id)

                # Call the handler
                callback = self._callbacks[callback_name]
                await callback(results, callback_context or {})

                logger.info(f"Processed batch {batch_id} with callback {callback_name}")
            else:
                logger.debug(f"No callback for batch {batch_id}")

            # Mark as processed
            await self._store.mark_processed(batch_id)

        except Exception as e:
            logger.error(f"Failed to process batch {batch_id}: {e}")
            # Don't mark as processed so it can be retried
            raise

__init__(api_key, provider_type, model, db_path, *, base_url=None, poll_interval=60.0)

Initialize the batch manager.

Parameters:

Name	Type	Description	Default
api_key	str	API key for the provider	required
provider_type	ProviderType	OPENAI_COMPATIBLE or ANTHROPIC	required
model	str	Default model for batch requests	required
db_path	Path	Path to SQLite database for persistence	required
base_url	str | None	Optional custom base URL	None
poll_interval	float	Seconds between status checks (default 60)	60.0

Source code in src/nagents/batch/manager.py

def __init__(
    self,
    api_key: str,
    provider_type: ProviderType,
    model: str,
    db_path: Path,
    *,
    base_url: str | None = None,
    poll_interval: float = 60.0,
):
    """
    Initialize the batch manager.

    Args:
        api_key: API key for the provider
        provider_type: OPENAI_COMPATIBLE or ANTHROPIC
        model: Default model for batch requests
        db_path: Path to SQLite database for persistence
        base_url: Optional custom base URL
        poll_interval: Seconds between status checks (default 60)
    """
    self.api_key = api_key
    self.provider_type = provider_type
    self.model = model
    self.base_url = base_url
    self.poll_interval = poll_interval

    # Initialize components
    self._store = BatchStore(db_path)
    self._client = BatchClient(
        provider_type=provider_type,
        api_key=api_key,
        model=model,
        base_url=base_url,
    )

    # Callback registry
    self._callbacks: dict[str, BatchCallback] = {}

    # Background task tracking
    self._monitor_tasks: dict[str, asyncio.Task[None]] = {}
    self._initialized = False

initialize() async

Initialize the store and resume pending batches.

Source code in src/nagents/batch/manager.py

async def initialize(self) -> None:
    """Initialize the store and resume pending batches."""
    if self._initialized:
        return

    await self._store.initialize()
    self._initialized = True
    logger.info("BatchManager initialized")

register_callback(name)

Decorator to register a callback handler.

Parameters:

Name	Type	Description	Default
name	str	Unique name for this callback	required

Example

@manager.register_callback("process_summaries") async def handler(results: list[BatchResult], context: dict): for result in results: save_summary(result.custom_id, result.content)

Source code in src/nagents/batch/manager.py

def register_callback(
    self,
    name: str,
) -> Callable[[BatchCallback], BatchCallback]:
    """
    Decorator to register a callback handler.

    Args:
        name: Unique name for this callback

    Example:
        @manager.register_callback("process_summaries")
        async def handler(results: list[BatchResult], context: dict):
            for result in results:
                save_summary(result.custom_id, result.content)
    """

    def decorator(func: BatchCallback) -> BatchCallback:
        self._callbacks[name] = func
        logger.debug(f"Registered batch callback: {name}")
        return func

    return decorator

add_callback(name, callback)

Add a callback handler directly.

Parameters:

Name	Type	Description	Default
name	str	Unique name for this callback	required
callback	BatchCallback	Async function to handle results	required

Source code in src/nagents/batch/manager.py

def add_callback(self, name: str, callback: BatchCallback) -> None:
    """
    Add a callback handler directly.

    Args:
        name: Unique name for this callback
        callback: Async function to handle results
    """
    self._callbacks[name] = callback
    logger.debug(f"Added batch callback: {name}")

create_batch(requests, *, config=None, tools=None, generation_config=None, callback_name=None, callback_context=None, store_requests=True, auto_monitor=True) async

Create a batch job with persistence.

Parameters:

Name	Type	Description	Default
requests	list[BatchRequest]	List of BatchRequest objects	required
config	BatchConfig | None	Optional batch configuration	None
tools	list[ToolDefinition] | None	Optional default tools for all requests	None
generation_config	GenerationConfig | None	Optional default generation config	None
callback_name	str | None	Name of registered callback for results	None
callback_context	dict[str, Any] | None	Context dict passed to callback	None
store_requests	bool	Whether to store requests for retry (default True)	True
auto_monitor	bool	Whether to start background monitoring (default True)	True

Returns:

Type	Description
BatchJob	The created BatchJob

Source code in src/nagents/batch/manager.py

async def create_batch(
    self,
    requests: list[BatchRequest],
    *,
    config: BatchConfig | None = None,
    tools: list[ToolDefinition] | None = None,
    generation_config: GenerationConfig | None = None,
    callback_name: str | None = None,
    callback_context: dict[str, Any] | None = None,
    store_requests: bool = True,
    auto_monitor: bool = True,
) -> BatchJob:
    """
    Create a batch job with persistence.

    Args:
        requests: List of BatchRequest objects
        config: Optional batch configuration
        tools: Optional default tools for all requests
        generation_config: Optional default generation config
        callback_name: Name of registered callback for results
        callback_context: Context dict passed to callback
        store_requests: Whether to store requests for retry (default True)
        auto_monitor: Whether to start background monitoring (default True)

    Returns:
        The created BatchJob
    """
    await self.initialize()

    # Validate callback if provided
    if callback_name and callback_name not in self._callbacks:
        raise ValueError(f"Unknown callback: {callback_name}. Available: {list(self._callbacks.keys())}")

    # Create batch via API
    job = await self._client.create_batch(requests, config, tools, generation_config)

    # Persist to database
    await self._store.save_batch(
        job,
        requests=requests if store_requests else None,
        model=self.model,
        base_url=self.base_url or "",
        callback_name=callback_name,
        callback_context=callback_context,
    )

    logger.info(f"Created batch {job.id} with {len(requests)} requests")

    # Start background monitoring
    if auto_monitor:
        self._start_monitor(job.id)

    return job

resume_all() async

Resume monitoring all pending batches.

Call this on startup to resume any batches that were submitted before the agent was shut down.

Returns:

Type	Description
list[str]	List of batch IDs that were resumed

Source code in src/nagents/batch/manager.py

async def resume_all(self) -> list[str]:
    """
    Resume monitoring all pending batches.

    Call this on startup to resume any batches that were
    submitted before the agent was shut down.

    Returns:
        List of batch IDs that were resumed
    """
    await self.initialize()

    resumed = []

    # Get active batches (still processing)
    active = await self._store.get_active_batches()
    for job in active:
        if job.id not in self._monitor_tasks:
            self._start_monitor(job.id)
            resumed.append(job.id)
            logger.info(f"Resumed monitoring batch {job.id}")

    # Process completed but unhandled batches
    completed = await self._store.get_completed_unprocessed()
    for job in completed:
        await self._process_completed_batch(job.id)
        resumed.append(job.id)
        logger.info(f"Processing completed batch {job.id}")

    return resumed

close() async

Close the manager and release resources.

Source code in src/nagents/batch/manager.py

async def close(self) -> None:
    """Close the manager and release resources."""
    # Cancel all monitor tasks
    for task in self._monitor_tasks.values():
        task.cancel()
    self._monitor_tasks.clear()

    await self._client.close()

nagents.BatchClient

Unified batch processing client for OpenAI, Azure OpenAI, and Anthropic.

Example

async with BatchClient( provider_type=ProviderType.OPENAI_COMPATIBLE, api_key="sk-...", model="gpt-4o-mini", ) as client: # Create batch requests requests = [ BatchRequest( custom_id="req-1", messages=[Message(role="user", content="Hello!")], ), BatchRequest( custom_id="req-2", messages=[Message(role="user", content="How are you?")], ), ]

# Submit batch
job = await client.create_batch(requests)
print(f"Batch created: {job.id}")

# Poll for completion
while job.status in (BatchStatus.VALIDATING, BatchStatus.IN_PROGRESS):
    await asyncio.sleep(60)
    job = await client.get_batch(job.id)

# Get results
async for result in client.get_results(job):
    print(f"{result.custom_id}: {result.content}")

Source code in src/nagents/batch/client.py

class BatchClient:
    """
    Unified batch processing client for OpenAI, Azure OpenAI, and Anthropic.

    Example:
        async with BatchClient(
            provider_type=ProviderType.OPENAI_COMPATIBLE,
            api_key="sk-...",
            model="gpt-4o-mini",
        ) as client:
            # Create batch requests
            requests = [
                BatchRequest(
                    custom_id="req-1",
                    messages=[Message(role="user", content="Hello!")],
                ),
                BatchRequest(
                    custom_id="req-2",
                    messages=[Message(role="user", content="How are you?")],
                ),
            ]

            # Submit batch
            job = await client.create_batch(requests)
            print(f"Batch created: {job.id}")

            # Poll for completion
            while job.status in (BatchStatus.VALIDATING, BatchStatus.IN_PROGRESS):
                await asyncio.sleep(60)
                job = await client.get_batch(job.id)

            # Get results
            async for result in client.get_results(job):
                print(f"{result.custom_id}: {result.content}")
    """

    def __init__(
        self,
        provider_type: ProviderType,
        api_key: str,
        model: str,
        base_url: str | None = None,
        timeout: float = 120.0,
        logger: "HTTPLogger | None" = None,
    ):
        """
        Initialize the batch client.

        Args:
            provider_type: OPENAI_COMPATIBLE, AZURE_OPENAI_COMPATIBLE, AZURE_OPENAI_COMPATIBLE_V1, or ANTHROPIC
            api_key: API key for authentication
            model: Default model for batch requests
            base_url: Optional custom base URL
            timeout: Request timeout in seconds
            logger: Optional HTTP logger for request/response logging
        """
        if provider_type == ProviderType.GEMINI_NATIVE:
            raise ValueError("Batch processing is not supported for Gemini native API. Use Vertex AI instead.")

        self.provider_type = provider_type
        self.api_key = api_key
        self.model = model
        self._http = HTTPClient(timeout=timeout, logger=logger)

        # Set base URLs based on provider type
        if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
            # Azure V1: base_url includes /openai, need to add /v1 for batch API
            if base_url:
                self.base_url = f"{base_url.rstrip('/')}/v1"
            else:
                raise ValueError("base_url is required for AZURE_OPENAI_COMPATIBLE_V1")
        elif base_url:
            self.base_url = base_url.rstrip("/")
        elif provider_type == ProviderType.OPENAI_COMPATIBLE:
            self.base_url = "https://api.openai.com/v1"
        elif provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
            # Azure deployment-based: needs base_url
            raise ValueError("base_url is required for AZURE_OPENAI_COMPATIBLE")
        else:  # ANTHROPIC
            self.base_url = "https://api.anthropic.com/v1"

    def _is_azure(self) -> bool:
        """Check if this is an Azure provider."""
        return self.provider_type in (
            ProviderType.AZURE_OPENAI_COMPATIBLE,
            ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
        )

    def set_session_id(self, session_id: str | None) -> None:
        """Set the current session ID for logging."""
        self._http.set_session_id(session_id)

    def _log_file_request(self, method: str, url: str, headers: dict[str, str], filename: str) -> None:
        """Log a file upload/download request."""
        if self._http._logger:
            self._http._logger.log_request(method, url, headers, {"file": filename}, self._http._session_id)

    def _log_file_response(self, url: str, status: int, body: str, headers: dict[str, str] | None = None) -> None:
        """Log a file upload/download response."""
        if self._http._logger:
            self._http._logger.log_response(url, status, body, headers, self._http._session_id)

    def _get_headers(self, for_file_upload: bool = False) -> dict[str, str]:
        """Get headers for API requests."""
        if self._is_azure():
            return {
                "api-key": self.api_key,
                "Content-Type": "application/json",
            }
        elif self.provider_type == ProviderType.OPENAI_COMPATIBLE:
            return {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            }
        else:  # ANTHROPIC
            return {
                "x-api-key": self.api_key,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json",
            }

    def _get_file_upload_headers(self) -> dict[str, str]:
        """Get headers for file upload (no Content-Type for multipart)."""
        if self._is_azure():
            return {"api-key": self.api_key}
        elif self.provider_type == ProviderType.OPENAI_COMPATIBLE:
            return {"Authorization": f"Bearer {self.api_key}"}
        else:  # ANTHROPIC
            return {
                "x-api-key": self.api_key,
                "anthropic-version": "2023-06-01",
            }

    # =========================================================================
    # OpenAI Batch Implementation
    # =========================================================================

    async def _create_batch_openai(
        self,
        requests: list[BatchRequest],
        config: BatchConfig | None = None,
        tools: list[ToolDefinition] | None = None,
        generation_config: GenerationConfig | None = None,
    ) -> BatchJob:
        """Create a batch job using OpenAI's file-based API."""
        from ..adapters import openai as openai_adapter

        config = config or BatchConfig()

        # Azure uses /chat/completions, OpenAI uses /v1/chat/completions
        # Strip /v1 prefix for Azure if present
        file_endpoint = config.endpoint
        batch_endpoint = config.endpoint
        if self._is_azure() and file_endpoint.startswith("/v1/"):
            file_endpoint = file_endpoint[3:]  # Remove /v1 prefix
            batch_endpoint = file_endpoint

        # Build JSONL content
        lines = []
        for req in requests:
            # Format messages
            formatted_messages = openai_adapter.format_messages(req.messages)

            body: dict[str, Any] = {
                "model": req.model or self.model,
                "messages": formatted_messages,
            }

            # Add tools
            req_tools = req.tools or tools
            if req_tools:
                body["tools"] = openai_adapter.format_tools(req_tools)

            # Add generation config
            req_config = req.config or generation_config
            if req_config:
                if req_config.temperature is not None:
                    body["temperature"] = req_config.temperature
                if req_config.max_tokens is not None:
                    body["max_completion_tokens"] = req_config.max_tokens
                if req_config.top_p is not None:
                    body["top_p"] = req_config.top_p
                if req_config.stop:
                    body["stop"] = req_config.stop

            line = {
                "custom_id": req.custom_id,
                "method": "POST",
                "url": file_endpoint,
                "body": body,
            }
            lines.append(json.dumps(line))

        jsonl_content = "\n".join(lines)

        # Step 1: Upload the file
        upload_url = f"{self.base_url}/files"
        upload_headers = self._get_file_upload_headers()

        # Log file upload request
        self._log_file_request("POST", upload_url, upload_headers, "batch_input.jsonl")

        # Use multipart form data for file upload
        import aiohttp

        form = aiohttp.FormData()
        form.add_field("purpose", "batch")
        form.add_field(
            "file",
            io.BytesIO(jsonl_content.encode("utf-8")),
            filename="batch_input.jsonl",
            content_type="application/jsonl",
        )

        async with (
            aiohttp.ClientSession() as session,
            session.post(
                upload_url,
                headers=upload_headers,
                data=form,
            ) as resp,
        ):
            response_text = await resp.text()
            resp_headers = dict(resp.headers)
            # Log file upload response
            self._log_file_response(upload_url, resp.status, response_text, resp_headers)

            if resp.status >= 400:
                raise RuntimeError(f"Failed to upload batch file: {resp.status} {response_text}")
            file_response = json.loads(response_text)

        input_file_id = file_response["id"]
        logger.info(f"Uploaded batch input file: {input_file_id}")

        # Step 2: Create the batch
        batch_url = f"{self.base_url}/batches"
        batch_body = {
            "input_file_id": input_file_id,
            "endpoint": batch_endpoint,
            "completion_window": config.completion_window,
        }
        if config.metadata:
            batch_body["metadata"] = config.metadata

        response = await self._http.post_json(batch_url, batch_body, self._get_headers())

        return self._parse_openai_batch(response)

    def _parse_openai_batch(self, response: dict[str, Any]) -> BatchJob:
        """Parse OpenAI batch response into BatchJob."""

        status_map = {
            "validating": BatchStatus.VALIDATING,
            "failed": BatchStatus.FAILED,
            "in_progress": BatchStatus.IN_PROGRESS,
            "finalizing": BatchStatus.FINALIZING,
            "completed": BatchStatus.COMPLETED,
            "expired": BatchStatus.EXPIRED,
            "cancelling": BatchStatus.CANCELLING,
            "cancelled": BatchStatus.CANCELLED,
        }

        counts = response.get("request_counts", {})
        provider: Literal["openai", "azure", "anthropic"] = "azure" if self._is_azure() else "openai"

        return BatchJob(
            id=response["id"],
            status=status_map.get(response.get("status", ""), BatchStatus.IN_PROGRESS),
            provider=provider,
            request_counts=BatchRequestCounts(
                total=counts.get("total", 0),
                succeeded=counts.get("completed", 0),
                failed=counts.get("failed", 0),
            ),
            created_at=response.get("created_at"),
            expires_at=response.get("expires_at"),
            completed_at=response.get("completed_at"),
            input_file_id=response.get("input_file_id"),
            output_file_id=response.get("output_file_id"),
            error_file_id=response.get("error_file_id"),
            endpoint=response.get("endpoint"),
            metadata=response.get("metadata", {}),
            raw_response=response,
        )

    async def _get_batch_openai(self, batch_id: str) -> BatchJob:
        """Get OpenAI batch status."""
        url = f"{self.base_url}/batches/{batch_id}"
        response = await self._http.get_json(url, self._get_headers())
        return self._parse_openai_batch(response)

    async def _get_results_openai(self, job: BatchJob) -> AsyncIterator[BatchResult]:
        """Stream results from OpenAI batch."""
        if not job.output_file_id:
            if job.error_file_id:
                logger.warning(f"No output file, but error file exists: {job.error_file_id}")
                async for result in self._get_error_results(job):
                    yield result
            else:
                logger.warning("No output file ID or error file ID available")
            return

        url = f"{self.base_url}/files/{job.output_file_id}/content"
        headers = self._get_headers()

        # Log file download request
        self._log_file_request("GET", url, headers, f"results_{job.output_file_id}")

        # Download file content
        import aiohttp

        async with (
            aiohttp.ClientSession() as session,
            session.get(url, headers=headers) as resp,
        ):
            content = await resp.text()
            resp_headers = dict(resp.headers)
            # Log file download response
            self._log_file_response(url, resp.status, content, resp_headers)

            if resp.status >= 400:
                raise RuntimeError(f"Failed to download results: {resp.status} {content}")

        # Parse JSONL
        for line in content.strip().split("\n"):
            if not line:
                continue
            try:
                data = json.loads(line)
                yield self._parse_openai_result(data)
            except json.JSONDecodeError:
                logger.warning(f"Failed to parse result line: {line[:100]}")

    def _parse_openai_result(self, data: dict[str, Any]) -> BatchResult:
        """Parse OpenAI batch result."""
        custom_id = data.get("custom_id", "")
        error = data.get("error")
        response = data.get("response", {})

        if error:
            return BatchResult(
                custom_id=custom_id,
                result_type="errored",
                error_type=error.get("code", "unknown"),
                error_message=error.get("message", "Unknown error"),
                raw_response=data,
            )

        # Check response status
        status_code = response.get("status_code", 200)
        if status_code >= 400:
            return BatchResult(
                custom_id=custom_id,
                result_type="errored",
                error_type=f"http_{status_code}",
                error_message=str(response.get("body", {})),
                raw_response=data,
            )

        # Parse successful response
        body = response.get("body", {})
        choices = body.get("choices", [])

        content = None
        tool_calls = None
        finish_reason = FinishReason.UNKNOWN

        if choices:
            choice = choices[0]
            message = choice.get("message", {})
            content = message.get("content")
            if message.get("tool_calls"):
                tool_calls = message["tool_calls"]

            # Parse finish_reason
            fr = choice.get("finish_reason")
            if fr == "stop":
                finish_reason = FinishReason.STOP
            elif fr == "tool_calls":
                finish_reason = FinishReason.TOOL_CALLS
            elif fr == "length":
                finish_reason = FinishReason.LENGTH
            elif fr == "content_filter":
                finish_reason = FinishReason.CONTENT_FILTER

        # Parse detailed usage
        usage = None
        if body.get("usage"):
            usage_data = body["usage"]
            prompt_details = usage_data.get("prompt_tokens_details") or {}
            completion_details = usage_data.get("completion_tokens_details") or {}
            usage = UsageInfo(
                prompt_tokens=usage_data.get("prompt_tokens", 0),
                completion_tokens=usage_data.get("completion_tokens", 0),
                total_tokens=usage_data.get("total_tokens", 0),
                cached_tokens=prompt_details.get("cached_tokens", 0),
                audio_tokens=prompt_details.get("audio_tokens", 0) + completion_details.get("audio_tokens", 0),
                reasoning_tokens=completion_details.get("reasoning_tokens", 0),
            )

        return BatchResult(
            custom_id=custom_id,
            result_type="succeeded",
            content=content,
            tool_calls=tool_calls,
            usage=usage,
            finish_reason=finish_reason,
            raw_response=data,
        )

    async def _get_error_results(self, job: BatchJob) -> AsyncIterator[BatchResult]:
        """Stream error results from OpenAI batch error file."""
        if not job.error_file_id:
            return

        url = f"{self.base_url}/files/{job.error_file_id}/content"
        headers = self._get_headers()

        self._log_file_request("GET", url, headers, f"errors_{job.error_file_id}")

        import aiohttp

        async with (
            aiohttp.ClientSession() as session,
            session.get(url, headers=headers) as resp,
        ):
            content = await resp.text()
            resp_headers = dict(resp.headers)
            self._log_file_response(url, resp.status, content, resp_headers)

            if resp.status >= 400:
                raise RuntimeError(f"Failed to download error file: {resp.status} {content}")

        for line in content.strip().split("\n"):
            if not line:
                continue
            try:
                data = json.loads(line)
                custom_id = data.get("custom_id", "")
                error = data.get("error", {})
                yield BatchResult(
                    custom_id=custom_id,
                    result_type="errored",
                    error_type=error.get("code", "unknown"),
                    error_message=error.get("message", "Unknown error"),
                    raw_response=data,
                )
            except json.JSONDecodeError:
                logger.warning(f"Failed to parse error line: {line[:100]}")

    async def _cancel_batch_openai(self, batch_id: str) -> BatchJob:
        """Cancel OpenAI batch."""
        url = f"{self.base_url}/batches/{batch_id}/cancel"
        response = await self._http.post_json(url, {}, self._get_headers())
        return self._parse_openai_batch(response)

    async def _list_batches_openai(
        self, limit: int = 20, after: str | None = None
    ) -> tuple[list[BatchJob], str | None]:
        """List OpenAI batches."""
        url = f"{self.base_url}/batches?limit={limit}"
        if after:
            url += f"&after={after}"

        response = await self._http.get_json(url, self._get_headers())

        jobs = [self._parse_openai_batch(b) for b in response.get("data", [])]
        next_cursor = response.get("last_id") if response.get("has_more") else None

        return jobs, next_cursor

    # =========================================================================
    # Anthropic Batch Implementation
    # =========================================================================

    async def _create_batch_anthropic(
        self,
        requests: list[BatchRequest],
        config: BatchConfig | None = None,
        tools: list[ToolDefinition] | None = None,
        generation_config: GenerationConfig | None = None,
    ) -> BatchJob:
        """Create a batch job using Anthropic's Messages Batches API."""
        from ..adapters import anthropic as anthropic_adapter

        batch_requests = []
        for req in requests:
            # Format messages (extracts system prompt)
            system_prompt, formatted_messages = anthropic_adapter.format_messages(req.messages)

            # Use explicit system prompt if provided
            if req.system_prompt:
                system_prompt = req.system_prompt

            params: dict[str, Any] = {
                "model": req.model or self.model,
                "messages": formatted_messages,
                "max_tokens": 4096,  # Required by Anthropic
            }

            if system_prompt:
                params["system"] = system_prompt

            # Add tools
            req_tools = req.tools or tools
            if req_tools:
                params["tools"] = anthropic_adapter.format_tools(req_tools)

            # Add generation config
            req_config = req.config or generation_config
            if req_config:
                if req_config.max_tokens:
                    params["max_tokens"] = req_config.max_tokens
                if req_config.temperature is not None:
                    params["temperature"] = req_config.temperature
                if req_config.top_p is not None:
                    params["top_p"] = req_config.top_p
                if req_config.stop:
                    params["stop_sequences"] = req_config.stop

            batch_requests.append({"custom_id": req.custom_id, "params": params})

        # Create batch
        url = f"{self.base_url}/messages/batches"
        body = {"requests": batch_requests}

        response = await self._http.post_json(url, body, self._get_headers())

        return self._parse_anthropic_batch(response)

    def _parse_anthropic_batch(self, response: dict[str, Any]) -> BatchJob:
        """Parse Anthropic batch response into BatchJob."""
        status_map = {
            "in_progress": BatchStatus.IN_PROGRESS,
            "ended": BatchStatus.COMPLETED,
            "canceling": BatchStatus.CANCELLING,
        }

        counts = response.get("request_counts", {})

        # Determine final status
        processing_status = response.get("processing_status", "in_progress")
        status = status_map.get(processing_status, BatchStatus.IN_PROGRESS)

        # If ended, check if all succeeded or mixed
        if processing_status == "ended":
            if counts.get("expired", 0) > 0 and counts.get("succeeded", 0) == 0:
                status = BatchStatus.EXPIRED
            elif counts.get("canceled", 0) == counts.get("total", 0):
                status = BatchStatus.CANCELLED

        return BatchJob(
            id=response["id"],
            status=status,
            provider="anthropic",
            request_counts=BatchRequestCounts(
                total=counts.get("processing", 0)
                + counts.get("succeeded", 0)
                + counts.get("errored", 0)
                + counts.get("canceled", 0)
                + counts.get("expired", 0),
                processing=counts.get("processing", 0),
                succeeded=counts.get("succeeded", 0),
                failed=counts.get("errored", 0),
                cancelled=counts.get("canceled", 0),
                expired=counts.get("expired", 0),
            ),
            created_at=response.get("created_at"),
            expires_at=response.get("expires_at"),
            completed_at=response.get("ended_at"),
            cancelled_at=response.get("cancel_initiated_at"),
            results_url=response.get("results_url"),
            raw_response=response,
        )

    async def _get_batch_anthropic(self, batch_id: str) -> BatchJob:
        """Get Anthropic batch status."""
        url = f"{self.base_url}/messages/batches/{batch_id}"
        response = await self._http.get_json(url, self._get_headers())
        return self._parse_anthropic_batch(response)

    async def _get_results_anthropic(self, job: BatchJob) -> AsyncIterator[BatchResult]:
        """Stream results from Anthropic batch."""
        if not job.results_url:
            logger.warning("No results URL available")
            return

        headers = self._get_headers()

        # Stream results from URL
        import aiohttp

        async with aiohttp.ClientSession() as session:  # noqa: SIM117
            async with session.get(job.results_url, headers=headers) as resp:
                if resp.status >= 400:
                    error_text = await resp.text()
                    raise RuntimeError(f"Failed to download results: {resp.status} {error_text}")

                # Stream line by line
                async for line in resp.content:
                    line_str = line.decode("utf-8").strip()
                    if not line_str:
                        continue
                    try:
                        data = json.loads(line_str)
                        yield self._parse_anthropic_result(data)
                    except json.JSONDecodeError:
                        logger.warning(f"Failed to parse result line: {line_str[:100]}")

    def _parse_anthropic_result(self, data: dict[str, Any]) -> BatchResult:
        """Parse Anthropic batch result."""
        custom_id = data.get("custom_id", "")
        result = data.get("result", {})
        result_type = result.get("type", "errored")

        if result_type == "succeeded":
            message = result.get("message", {})
            content_blocks = message.get("content", [])

            # Extract text content
            text_parts = []
            tool_calls = []
            for block in content_blocks:
                if block.get("type") == "text":
                    text_parts.append(block.get("text", ""))
                elif block.get("type") == "tool_use":
                    tool_calls.append(block)

            return BatchResult(
                custom_id=custom_id,
                result_type="succeeded",
                content="\n".join(text_parts) if text_parts else None,
                tool_calls=tool_calls if tool_calls else None,
                usage=message.get("usage"),
                raw_response=data,
            )

        elif result_type == "errored":
            error = result.get("error", {})
            return BatchResult(
                custom_id=custom_id,
                result_type="errored",
                error_type=error.get("type", "unknown"),
                error_message=error.get("message", "Unknown error"),
                raw_response=data,
            )

        elif result_type == "canceled":
            return BatchResult(
                custom_id=custom_id,
                result_type="cancelled",
                raw_response=data,
            )

        else:  # expired
            return BatchResult(
                custom_id=custom_id,
                result_type="expired",
                raw_response=data,
            )

    async def _cancel_batch_anthropic(self, batch_id: str) -> BatchJob:
        """Cancel Anthropic batch."""
        url = f"{self.base_url}/messages/batches/{batch_id}/cancel"
        response = await self._http.post_json(url, {}, self._get_headers())
        return self._parse_anthropic_batch(response)

    async def _list_batches_anthropic(
        self, limit: int = 20, after: str | None = None
    ) -> tuple[list[BatchJob], str | None]:
        """List Anthropic batches."""
        url = f"{self.base_url}/messages/batches?limit={limit}"
        if after:
            url += f"&after_id={after}"

        response = await self._http.get_json(url, self._get_headers())

        jobs = [self._parse_anthropic_batch(b) for b in response.get("data", [])]
        next_cursor = response.get("last_id") if response.get("has_more") else None

        return jobs, next_cursor

    # =========================================================================
    # Public Interface
    # =========================================================================

    async def create_batch(
        self,
        requests: list[BatchRequest],
        config: BatchConfig | None = None,
        tools: list[ToolDefinition] | None = None,
        generation_config: GenerationConfig | None = None,
    ) -> BatchJob:
        """
        Create a batch processing job.

        Args:
            requests: List of BatchRequest objects
            config: Optional batch configuration
            tools: Optional default tools for all requests
            generation_config: Optional default generation config

        Returns:
            BatchJob with ID and initial status
        """
        if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
            return await self._create_batch_openai(requests, config, tools, generation_config)
        else:
            return await self._create_batch_anthropic(requests, config, tools, generation_config)

    async def get_batch(self, batch_id: str) -> BatchJob:
        """
        Get the current status of a batch job.

        Args:
            batch_id: The batch ID

        Returns:
            Updated BatchJob
        """
        if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
            return await self._get_batch_openai(batch_id)
        else:
            return await self._get_batch_anthropic(batch_id)

    async def get_results(self, job: BatchJob) -> AsyncIterator[BatchResult]:
        """
        Stream results from a completed batch job.

        Args:
            job: The BatchJob (must be completed)

        Yields:
            BatchResult for each request
        """
        if job.provider in ("openai", "azure"):
            async for result in self._get_results_openai(job):
                yield result
        else:
            async for result in self._get_results_anthropic(job):
                yield result

    async def cancel_batch(self, batch_id: str) -> BatchJob:
        """
        Cancel a batch job.

        Args:
            batch_id: The batch ID

        Returns:
            Updated BatchJob (status will be CANCELLING)
        """
        if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
            return await self._cancel_batch_openai(batch_id)
        else:
            return await self._cancel_batch_anthropic(batch_id)

    async def list_batches(self, limit: int = 20, after: str | None = None) -> tuple[list[BatchJob], str | None]:
        """
        List batch jobs.

        Args:
            limit: Maximum number of batches to return
            after: Cursor for pagination

        Returns:
            Tuple of (list of BatchJobs, next cursor or None)
        """
        if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
            return await self._list_batches_openai(limit, after)
        else:
            return await self._list_batches_anthropic(limit, after)

    async def wait_for_completion(
        self,
        batch_id: str,
        poll_interval: float = 60.0,
        max_wait: float | None = None,
    ) -> BatchJob:
        """
        Wait for a batch to complete.

        Args:
            batch_id: The batch ID
            poll_interval: Seconds between status checks
            max_wait: Maximum seconds to wait (None for no limit)

        Returns:
            Completed BatchJob

        Raises:
            TimeoutError: If max_wait is exceeded
        """
        import asyncio
        import time

        start = time.time()
        active_statuses = {
            BatchStatus.VALIDATING,
            BatchStatus.IN_PROGRESS,
            BatchStatus.FINALIZING,
        }

        poll_count = 0
        while True:
            poll_count += 1
            job = await self.get_batch(batch_id)
            elapsed = time.time() - start
            logger.info(f"Batch {batch_id} status check #{poll_count}: {job.status.value} (elapsed: {elapsed:.1f}s)")

            if job.status not in active_statuses:
                return job

            if max_wait and (time.time() - start) > max_wait:
                raise TimeoutError(f"Batch {batch_id} did not complete within {max_wait} seconds")

            logger.info(f"Waiting {poll_interval}s before next status check...")
            await asyncio.sleep(poll_interval)

    async def close(self) -> None:
        """Close the client and release resources."""
        await self._http.close()

    async def __aenter__(self) -> "BatchClient":
        return self

    async def __aexit__(self, *args: object) -> None:
        await self.close()

__init__(provider_type, api_key, model, base_url=None, timeout=120.0, logger=None)

Initialize the batch client.

Parameters:

Name	Type	Description	Default
provider_type	ProviderType	OPENAI_COMPATIBLE, AZURE_OPENAI_COMPATIBLE, AZURE_OPENAI_COMPATIBLE_V1, or ANTHROPIC	required
api_key	str	API key for authentication	required
model	str	Default model for batch requests	required
base_url	str | None	Optional custom base URL	None
timeout	float	Request timeout in seconds	120.0
logger	HTTPLogger | None	Optional HTTP logger for request/response logging	None

Source code in src/nagents/batch/client.py

def __init__(
    self,
    provider_type: ProviderType,
    api_key: str,
    model: str,
    base_url: str | None = None,
    timeout: float = 120.0,
    logger: "HTTPLogger | None" = None,
):
    """
    Initialize the batch client.

    Args:
        provider_type: OPENAI_COMPATIBLE, AZURE_OPENAI_COMPATIBLE, AZURE_OPENAI_COMPATIBLE_V1, or ANTHROPIC
        api_key: API key for authentication
        model: Default model for batch requests
        base_url: Optional custom base URL
        timeout: Request timeout in seconds
        logger: Optional HTTP logger for request/response logging
    """
    if provider_type == ProviderType.GEMINI_NATIVE:
        raise ValueError("Batch processing is not supported for Gemini native API. Use Vertex AI instead.")

    self.provider_type = provider_type
    self.api_key = api_key
    self.model = model
    self._http = HTTPClient(timeout=timeout, logger=logger)

    # Set base URLs based on provider type
    if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
        # Azure V1: base_url includes /openai, need to add /v1 for batch API
        if base_url:
            self.base_url = f"{base_url.rstrip('/')}/v1"
        else:
            raise ValueError("base_url is required for AZURE_OPENAI_COMPATIBLE_V1")
    elif base_url:
        self.base_url = base_url.rstrip("/")
    elif provider_type == ProviderType.OPENAI_COMPATIBLE:
        self.base_url = "https://api.openai.com/v1"
    elif provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
        # Azure deployment-based: needs base_url
        raise ValueError("base_url is required for AZURE_OPENAI_COMPATIBLE")
    else:  # ANTHROPIC
        self.base_url = "https://api.anthropic.com/v1"

create_batch(requests, config=None, tools=None, generation_config=None) async

Create a batch processing job.

Parameters:

Name	Type	Description	Default
requests	list[BatchRequest]	List of BatchRequest objects	required
config	BatchConfig | None	Optional batch configuration	None
tools	list[ToolDefinition] | None	Optional default tools for all requests	None
generation_config	GenerationConfig | None	Optional default generation config	None

Returns:

Type	Description
BatchJob	BatchJob with ID and initial status

Source code in src/nagents/batch/client.py

async def create_batch(
    self,
    requests: list[BatchRequest],
    config: BatchConfig | None = None,
    tools: list[ToolDefinition] | None = None,
    generation_config: GenerationConfig | None = None,
) -> BatchJob:
    """
    Create a batch processing job.

    Args:
        requests: List of BatchRequest objects
        config: Optional batch configuration
        tools: Optional default tools for all requests
        generation_config: Optional default generation config

    Returns:
        BatchJob with ID and initial status
    """
    if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
        return await self._create_batch_openai(requests, config, tools, generation_config)
    else:
        return await self._create_batch_anthropic(requests, config, tools, generation_config)

get_batch(batch_id) async

Get the current status of a batch job.

Parameters:

Name	Type	Description	Default
batch_id	str	The batch ID	required

Returns:

Type	Description
BatchJob	Updated BatchJob

Source code in src/nagents/batch/client.py

async def get_batch(self, batch_id: str) -> BatchJob:
    """
    Get the current status of a batch job.

    Args:
        batch_id: The batch ID

    Returns:
        Updated BatchJob
    """
    if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
        return await self._get_batch_openai(batch_id)
    else:
        return await self._get_batch_anthropic(batch_id)

get_results(job) async

Stream results from a completed batch job.

Parameters:

Name	Type	Description	Default
job	BatchJob	The BatchJob (must be completed)	required

Yields:

Type	Description
AsyncIterator[BatchResult]	BatchResult for each request

Source code in src/nagents/batch/client.py

async def get_results(self, job: BatchJob) -> AsyncIterator[BatchResult]:
    """
    Stream results from a completed batch job.

    Args:
        job: The BatchJob (must be completed)

    Yields:
        BatchResult for each request
    """
    if job.provider in ("openai", "azure"):
        async for result in self._get_results_openai(job):
            yield result
    else:
        async for result in self._get_results_anthropic(job):
            yield result

cancel_batch(batch_id) async

Cancel a batch job.

Parameters:

Name	Type	Description	Default
batch_id	str	The batch ID	required

Returns:

Type	Description
BatchJob	Updated BatchJob (status will be CANCELLING)

Source code in src/nagents/batch/client.py

async def cancel_batch(self, batch_id: str) -> BatchJob:
    """
    Cancel a batch job.

    Args:
        batch_id: The batch ID

    Returns:
        Updated BatchJob (status will be CANCELLING)
    """
    if self._is_azure() or self.provider_type == ProviderType.OPENAI_COMPATIBLE:
        return await self._cancel_batch_openai(batch_id)
    else:
        return await self._cancel_batch_anthropic(batch_id)

close() async

Close the client and release resources.

Source code in src/nagents/batch/client.py

async def close(self) -> None:
    """Close the client and release resources."""
    await self._http.close()

nagents.BatchStore

SQLite-based persistence for batch jobs.

Tracks batch jobs across agent restarts, enabling: - Resume pending batch monitoring after restart - Store original requests for retry on failure - Track callback handlers for result processing - Query batch history and status

Example

store = BatchStore(Path("batches.db")) await store.initialize()

Save a new batch

await store.save_batch(batch_job, requests, callback_name="process_summaries")

On restart, get pending batches

pending = await store.get_pending_batches() for job in pending: # Resume monitoring... pass

Mark as processed

await store.mark_processed(job.id)

Source code in src/nagents/batch/store.py

class BatchStore:
    """
    SQLite-based persistence for batch jobs.

    Tracks batch jobs across agent restarts, enabling:
    - Resume pending batch monitoring after restart
    - Store original requests for retry on failure
    - Track callback handlers for result processing
    - Query batch history and status

    Example:
        store = BatchStore(Path("batches.db"))
        await store.initialize()

        # Save a new batch
        await store.save_batch(batch_job, requests, callback_name="process_summaries")

        # On restart, get pending batches
        pending = await store.get_pending_batches()
        for job in pending:
            # Resume monitoring...
            pass

        # Mark as processed
        await store.mark_processed(job.id)
    """

    def __init__(self, db_path: Path):
        """
        Initialize the batch store.

        Args:
            db_path: Path to the SQLite database file
        """
        self.db_path = db_path
        self._initialized = False
        self._migration_manager: MigrationManager | None = None

    async def initialize(self) -> None:
        """Initialize database and run migrations."""
        if self._initialized:
            return

        # Ensure parent directory exists
        self.db_path.parent.mkdir(parents=True, exist_ok=True)

        # Create migration manager with batch-specific migrations
        self._migration_manager = MigrationManager(self.db_path, db_name="batch", migrations=list(batch_migrations))
        await self._migration_manager.initialize()

        self._initialized = True
        logger.debug(f"Batch store initialized at {self.db_path}")

    async def get_schema_version(self) -> int:
        """Get current database schema version.

        Returns:
            Current version number, or 0 if no migrations applied
        """
        await self.initialize()
        if self._migration_manager is None:
            return 0
        return await self._migration_manager.get_version()

    async def get_migration_history(self) -> list[dict[str, Any]]:
        """Get history of applied migrations.

        Returns:
            List of migration info dicts
        """
        await self.initialize()
        if self._migration_manager is None:
            return []
        return await self._migration_manager.get_applied_migrations()

    async def rollback_migration(self, steps: int = 1) -> None:
        """Rollback the last N migrations.

        Args:
            steps: Number of migrations to rollback (default: 1)

        Raises:
            ValueError: If not enough migrations to rollback
        """
        await self.initialize()
        if self._migration_manager is None:
            return
        await self._migration_manager.rollback(steps)

    async def migrate_to_version(self, version: int) -> None:
        """Migrate database to a specific version.

        Args:
            version: Target schema version

        Raises:
            ValueError: If target version is invalid
        """
        await self.initialize()
        if self._migration_manager is None:
            return
        await self._migration_manager.migrate_to(version)

    async def save_batch(
        self,
        job: BatchJob,
        requests: list[BatchRequest] | None = None,
        *,
        model: str = "",
        base_url: str = "",
        callback_name: str | None = None,
        callback_context: dict[str, Any] | None = None,
    ) -> None:
        """
        Save or update a batch job.

        Args:
            job: The BatchJob to save
            requests: Optional list of original requests (for retry)
            model: Model used for the batch
            base_url: Base URL for API (for resuming)
            callback_name: Name of callback handler for results
            callback_context: Context dict passed to callback handler
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            # Upsert batch job
            await db.execute(
                """INSERT INTO v2_batch_jobs (
                    id, provider, status, model,
                    total_requests, succeeded, failed, cancelled, expired,
                    created_at, expires_at, completed_at,
                    input_file_id, output_file_id, error_file_id, results_url,
                    base_url, callback_name, callback_context, metadata,
                    db_updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
                ON CONFLICT(id) DO UPDATE SET
                    status = excluded.status,
                    total_requests = excluded.total_requests,
                    succeeded = excluded.succeeded,
                    failed = excluded.failed,
                    cancelled = excluded.cancelled,
                    expired = excluded.expired,
                    completed_at = excluded.completed_at,
                    output_file_id = excluded.output_file_id,
                    error_file_id = excluded.error_file_id,
                    results_url = excluded.results_url,
                    db_updated_at = datetime('now')
                """,
                (
                    job.id,
                    job.provider,
                    job.status.value,
                    model,
                    job.request_counts.total,
                    job.request_counts.succeeded,
                    job.request_counts.failed,
                    job.request_counts.cancelled,
                    job.request_counts.expired,
                    job.created_at,
                    job.expires_at,
                    job.completed_at,
                    job.input_file_id,
                    job.output_file_id,
                    job.error_file_id,
                    job.results_url,
                    base_url,
                    callback_name,
                    json.dumps(callback_context) if callback_context else None,
                    json.dumps(job.metadata) if job.metadata else None,
                ),
            )

            # Save requests if provided
            if requests:
                for req in requests:
                    messages_json = json.dumps([asdict(m) for m in req.messages])
                    tools_json = json.dumps([asdict(t) for t in req.tools]) if req.tools else None
                    config_json = json.dumps(asdict(req.config)) if req.config else None

                    await db.execute(
                        """INSERT OR IGNORE INTO v2_batch_requests (
                            batch_id, custom_id, messages, model, tools, config, system_prompt
                        ) VALUES (?, ?, ?, ?, ?, ?, ?)""",
                        (
                            job.id,
                            req.custom_id,
                            messages_json,
                            req.model,
                            tools_json,
                            config_json,
                            req.system_prompt,
                        ),
                    )

            await db.commit()
            logger.debug(f"Saved batch job: {job.id} (status={job.status.value})")

    async def update_status(self, job: BatchJob) -> None:
        """
        Update batch job status and counts.

        Args:
            job: BatchJob with updated status
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            await db.execute(
                """UPDATE v2_batch_jobs SET
                    status = ?,
                    total_requests = ?,
                    succeeded = ?,
                    failed = ?,
                    cancelled = ?,
                    expired = ?,
                    completed_at = ?,
                    output_file_id = ?,
                    error_file_id = ?,
                    results_url = ?,
                    db_updated_at = datetime('now')
                WHERE id = ?""",
                (
                    job.status.value,
                    job.request_counts.total,
                    job.request_counts.succeeded,
                    job.request_counts.failed,
                    job.request_counts.cancelled,
                    job.request_counts.expired,
                    job.completed_at,
                    job.output_file_id,
                    job.error_file_id,
                    job.results_url,
                    job.id,
                ),
            )
            await db.commit()

    async def get_batch(self, batch_id: str) -> BatchJob | None:
        """
        Get a batch job by ID.

        Args:
            batch_id: The batch ID

        Returns:
            BatchJob or None if not found
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                "SELECT * FROM v2_batch_jobs WHERE id = ?",
                (batch_id,),
            )
            row = await cursor.fetchone()
            return self._row_to_batch_job(row) if row else None

    async def get_pending_batches(
        self,
        provider: Literal["openai", "anthropic"] | None = None,
    ) -> list[BatchJob]:
        """
        Get all pending (unprocessed) batch jobs.

        These are jobs that were submitted but not yet processed
        (either still running or completed but results not handled).

        Args:
            provider: Optional filter by provider

        Returns:
            List of pending BatchJobs
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row

            if provider:
                cursor = await db.execute(
                    """SELECT * FROM v2_batch_jobs
                    WHERE processed = 0 AND provider = ?
                    ORDER BY db_created_at ASC""",
                    (provider,),
                )
            else:
                cursor = await db.execute(
                    """SELECT * FROM v2_batch_jobs
                    WHERE processed = 0
                    ORDER BY db_created_at ASC"""
                )

            rows = await cursor.fetchall()
            return [self._row_to_batch_job(row) for row in rows]

    async def get_active_batches(self) -> list[BatchJob]:
        """
        Get batches that are still in progress.

        Returns:
            List of active BatchJobs
        """
        await self.initialize()

        active_statuses = [
            BatchStatus.VALIDATING.value,
            BatchStatus.IN_PROGRESS.value,
            BatchStatus.FINALIZING.value,
            BatchStatus.CANCELLING.value,
        ]

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row
            placeholders = ",".join(["?"] * len(active_statuses))
            cursor = await db.execute(
                f"""SELECT * FROM v2_batch_jobs
                WHERE status IN ({placeholders})
                ORDER BY db_created_at ASC""",
                active_statuses,
            )

            rows = await cursor.fetchall()
            return [self._row_to_batch_job(row) for row in rows]

    async def get_completed_unprocessed(self) -> list[BatchJob]:
        """
        Get completed batches that haven't been processed yet.

        These are batches where the API has finished but we haven't
        retrieved and handled the results yet.

        Returns:
            List of completed but unprocessed BatchJobs
        """
        await self.initialize()

        completed_statuses = [
            BatchStatus.COMPLETED.value,
            BatchStatus.ENDED.value,
            BatchStatus.FAILED.value,
            BatchStatus.EXPIRED.value,
            BatchStatus.CANCELLED.value,
        ]

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row
            placeholders = ",".join(["?"] * len(completed_statuses))
            cursor = await db.execute(
                f"""SELECT * FROM v2_batch_jobs
                WHERE status IN ({placeholders}) AND processed = 0
                ORDER BY db_created_at ASC""",
                completed_statuses,
            )

            rows = await cursor.fetchall()
            return [self._row_to_batch_job(row) for row in rows]

    async def mark_processed(
        self,
        batch_id: str,
        *,
        error: str | None = None,
    ) -> None:
        """
        Mark a batch as processed (results have been handled).

        Args:
            batch_id: The batch ID
            error: Optional error message if processing failed
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            now = datetime.now(UTC).isoformat()
            await db.execute(
                """UPDATE v2_batch_jobs SET
                    processed = 1,
                    processed_at = ?,
                    db_updated_at = datetime('now')
                WHERE id = ?""",
                (now, batch_id),
            )
            await db.commit()
            logger.debug(f"Marked batch as processed: {batch_id}")

    async def save_request_result(
        self,
        batch_id: str,
        custom_id: str,
        result_type: str,
        content: str | None = None,
        error: str | None = None,
    ) -> None:
        """
        Save result for an individual request.

        Args:
            batch_id: The batch ID
            custom_id: The request's custom_id
            result_type: succeeded, errored, cancelled, expired
            content: Response content (for succeeded)
            error: Error message (for errored)
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            await db.execute(
                """UPDATE v2_batch_requests SET
                    result_type = ?,
                    result_content = ?,
                    result_error = ?
                WHERE batch_id = ? AND custom_id = ?""",
                (result_type, content, error, batch_id, custom_id),
            )
            await db.commit()

    async def get_callback_info(self, batch_id: str) -> tuple[str | None, dict[str, Any] | None]:
        """
        Get callback info for a batch.

        Args:
            batch_id: The batch ID

        Returns:
            Tuple of (callback_name, callback_context)
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                "SELECT callback_name, callback_context FROM v2_batch_jobs WHERE id = ?",
                (batch_id,),
            )
            row = await cursor.fetchone()
            if not row:
                return None, None

            callback_name = row[0]
            callback_context = json.loads(row[1]) if row[1] else None
            return callback_name, callback_context

    async def get_connection_info(self, batch_id: str) -> tuple[str | None, str | None, str | None]:
        """
        Get connection info needed to resume batch monitoring.

        Args:
            batch_id: The batch ID

        Returns:
            Tuple of (provider, model, base_url)
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                "SELECT provider, model, base_url FROM v2_batch_jobs WHERE id = ?",
                (batch_id,),
            )
            row = await cursor.fetchone()
            if not row:
                return None, None, None
            return row[0], row[1], row[2]

    async def get_requests(self, batch_id: str) -> list[BatchRequest]:
        """
        Get original requests for a batch (for retry).

        Args:
            batch_id: The batch ID

        Returns:
            List of BatchRequest objects
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                "SELECT * FROM v2_batch_requests WHERE batch_id = ?",
                (batch_id,),
            )

            rows = await cursor.fetchall()
            requests = []
            for row in rows:
                from ..types import Message

                messages_data = json.loads(row["messages"])
                messages = [Message(**m) for m in messages_data]

                tools = None
                if row["tools"]:
                    from ..types import ToolDefinition

                    tools_data = json.loads(row["tools"])
                    tools = []
                    for t in tools_data:
                        tools.append(
                            ToolDefinition(
                                name=t["name"],
                                description=t["description"],
                                parameters=t.get("parameters", {}),
                            )
                        )

                config = None
                if row["config"]:
                    from ..types import GenerationConfig

                    config = GenerationConfig(**json.loads(row["config"]))

                requests.append(
                    BatchRequest(
                        custom_id=row["custom_id"],
                        messages=messages,
                        model=row["model"],
                        tools=tools,
                        config=config,
                        system_prompt=row["system_prompt"],
                    )
                )

            return requests

    async def list_batches(
        self,
        limit: int = 50,
        offset: int = 0,
        status: BatchStatus | None = None,
        processed: bool | None = None,
    ) -> list[BatchJob]:
        """
        List batch jobs with optional filtering.

        Args:
            limit: Maximum number of results
            offset: Offset for pagination
            status: Optional filter by status
            processed: Optional filter by processed state

        Returns:
            List of BatchJobs
        """
        await self.initialize()

        conditions = []
        params: list[Any] = []

        if status is not None:
            conditions.append("status = ?")
            params.append(status.value)

        if processed is not None:
            conditions.append("processed = ?")
            params.append(1 if processed else 0)

        where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
        params.extend([limit, offset])

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                f"""SELECT * FROM v2_batch_jobs
                {where_clause}
                ORDER BY db_created_at DESC
                LIMIT ? OFFSET ?""",
                params,
            )

            rows = await cursor.fetchall()
            return [self._row_to_batch_job(row) for row in rows]

    async def delete_batch(self, batch_id: str) -> bool:
        """
        Delete a batch job and its requests from the database.

        Args:
            batch_id: The batch ID

        Returns:
            True if deleted, False if not found
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            # Delete requests first (foreign key)
            await db.execute(
                "DELETE FROM v2_batch_requests WHERE batch_id = ?",
                (batch_id,),
            )
            cursor = await db.execute(
                "DELETE FROM v2_batch_jobs WHERE id = ?",
                (batch_id,),
            )
            await db.commit()

            deleted: bool = cursor.rowcount is not None and cursor.rowcount > 0
            if deleted:
                logger.debug(f"Deleted batch job: {batch_id}")
            return deleted

    async def cleanup_old_batches(self, days: int = 30) -> int:
        """
        Delete processed batches older than specified days.

        Args:
            days: Delete batches older than this many days

        Returns:
            Number of batches deleted
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            # Get batch IDs to delete
            cursor = await db.execute(
                """SELECT id FROM v2_batch_jobs
                WHERE processed = 1
                AND db_created_at < datetime('now', ?)""",
                (f"-{days} days",),
            )
            rows = await cursor.fetchall()
            batch_ids = [row[0] for row in rows]

            if not batch_ids:
                return 0

            # Delete requests
            placeholders = ",".join(["?"] * len(batch_ids))
            await db.execute(
                f"DELETE FROM v2_batch_requests WHERE batch_id IN ({placeholders})",
                batch_ids,
            )

            # Delete batches
            await db.execute(
                f"DELETE FROM v2_batch_jobs WHERE id IN ({placeholders})",
                batch_ids,
            )

            await db.commit()
            logger.info(f"Cleaned up {len(batch_ids)} old batch jobs")
            return len(batch_ids)

    def _row_to_batch_job(self, row: aiosqlite.Row) -> BatchJob:
        """Convert database row to BatchJob."""
        import contextlib

        status = BatchStatus(row["status"])

        metadata = {}
        if row["metadata"]:
            with contextlib.suppress(json.JSONDecodeError):
                metadata = json.loads(row["metadata"])

        return BatchJob(
            id=row["id"],
            status=status,
            provider=row["provider"],
            request_counts=BatchRequestCounts(
                total=row["total_requests"],
                succeeded=row["succeeded"],
                failed=row["failed"],
                cancelled=row["cancelled"],
                expired=row["expired"],
            ),
            created_at=row["created_at"],
            expires_at=row["expires_at"],
            completed_at=row["completed_at"],
            input_file_id=row["input_file_id"],
            output_file_id=row["output_file_id"],
            error_file_id=row["error_file_id"],
            results_url=row["results_url"],
            metadata=metadata,
        )

__init__(db_path)

Initialize the batch store.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the SQLite database file	required

Source code in src/nagents/batch/store.py

def __init__(self, db_path: Path):
    """
    Initialize the batch store.

    Args:
        db_path: Path to the SQLite database file
    """
    self.db_path = db_path
    self._initialized = False
    self._migration_manager: MigrationManager | None = None

initialize() async

Initialize database and run migrations.

Source code in src/nagents/batch/store.py

async def initialize(self) -> None:
    """Initialize database and run migrations."""
    if self._initialized:
        return

    # Ensure parent directory exists
    self.db_path.parent.mkdir(parents=True, exist_ok=True)

    # Create migration manager with batch-specific migrations
    self._migration_manager = MigrationManager(self.db_path, db_name="batch", migrations=list(batch_migrations))
    await self._migration_manager.initialize()

    self._initialized = True
    logger.debug(f"Batch store initialized at {self.db_path}")

nagents.BatchRequest dataclass

A single request in a batch.

This is the common format used to build batch requests. Adapters convert this to provider-specific formats.

Source code in src/nagents/batch/types.py

@dataclass
class BatchRequest:
    """
    A single request in a batch.

    This is the common format used to build batch requests.
    Adapters convert this to provider-specific formats.
    """

    custom_id: str  # Unique ID to match request with result
    messages: list[Message]
    model: str | None = None  # Optional override (uses batch default if not set)
    tools: list[ToolDefinition] | None = None
    config: GenerationConfig | None = None
    system_prompt: str | None = None  # Convenience for system message

nagents.BatchConfig dataclass

Configuration for batch processing.

OpenAI: - endpoint: /v1/chat/completions, /v1/embeddings, etc. - completion_window: Only "24h" supported currently

Azure OpenAI: - endpoint: /chat/completions (no /v1 prefix) - completion_window: Only "24h" supported currently

Anthropic: - No additional config needed (uses Messages API format)

Source code in src/nagents/batch/types.py

@dataclass
class BatchConfig:
    """
    Configuration for batch processing.

    OpenAI:
    - endpoint: /v1/chat/completions, /v1/embeddings, etc.
    - completion_window: Only "24h" supported currently

    Azure OpenAI:
    - endpoint: /chat/completions (no /v1 prefix)
    - completion_window: Only "24h" supported currently

    Anthropic:
    - No additional config needed (uses Messages API format)
    """

    endpoint: str = "/v1/chat/completions"
    completion_window: str = "24h"

    # Common
    metadata: dict[str, str] | None = None

nagents.BatchJob dataclass

Represents a batch processing job.

Contains metadata about the batch and its current status.

Source code in src/nagents/batch/types.py

@dataclass
class BatchJob:
    """
    Represents a batch processing job.

    Contains metadata about the batch and its current status.
    """

    id: str
    status: BatchStatus
    provider: Literal["openai", "azure", "anthropic"]

    # Request counts
    request_counts: BatchRequestCounts = field(default_factory=BatchRequestCounts)

    # Timestamps (ISO format)
    created_at: str | None = None
    expires_at: str | None = None
    completed_at: str | None = None
    cancelled_at: str | None = None

    # OpenAI-specific
    input_file_id: str | None = None
    output_file_id: str | None = None
    error_file_id: str | None = None
    endpoint: str | None = None

    # Anthropic-specific
    results_url: str | None = None

    # Metadata
    metadata: dict[str, str] = field(default_factory=dict)

    # Raw response for debugging
    raw_response: dict[str, Any] | None = None

nagents.BatchRequestCounts dataclass

Count of requests in various states.

Source code in src/nagents/batch/types.py

@dataclass
class BatchRequestCounts:
    """Count of requests in various states."""

    total: int = 0
    processing: int = 0  # Anthropic
    succeeded: int = 0
    failed: int = 0
    cancelled: int = 0
    expired: int = 0

nagents.BatchResult dataclass

Result for a single request in a batch.

The result can be one of: - succeeded: Contains the response message/content - errored: Contains error information - cancelled: Request was cancelled before processing - expired: Batch expired before request was processed

Source code in src/nagents/batch/types.py

@dataclass
class BatchResult:
    """
    Result for a single request in a batch.

    The result can be one of:
    - succeeded: Contains the response message/content
    - errored: Contains error information
    - cancelled: Request was cancelled before processing
    - expired: Batch expired before request was processed
    """

    custom_id: str
    result_type: Literal["succeeded", "errored", "cancelled", "expired"]

    # For succeeded results
    content: str | None = None
    tool_calls: list[BatchToolCall] | None = None
    usage: UsageInfo | None = None
    finish_reason: FinishReason = FinishReason.UNKNOWN

    # For errored results
    error_type: str | None = None
    error_message: str | None = None

    # Full response for advanced use cases
    raw_response: dict[str, Any] | None = None

nagents.BatchStatus

Bases: Enum

Status of a batch job.

Source code in src/nagents/batch/types.py

class BatchStatus(Enum):
    """Status of a batch job."""

    # Common statuses
    VALIDATING = "validating"  # Input being validated
    IN_PROGRESS = "in_progress"  # Batch is processing
    COMPLETED = "completed"  # All requests completed
    FAILED = "failed"  # Batch failed validation or processing
    EXPIRED = "expired"  # Batch didn't complete in time window
    CANCELLING = "cancelling"  # Cancellation in progress
    CANCELLED = "cancelled"  # Batch was cancelled

    # OpenAI-specific
    FINALIZING = "finalizing"  # Results being prepared

    # Anthropic-specific (mapped to common)
    ENDED = "ended"  # Processing ended (success or partial)