diff --git a/.env.example b/.env.example
index 9ce5d9e..d60a99b 100644
--- a/.env.example
+++ b/.env.example
@@ -53,6 +53,10 @@ DEV_PAGES_USE_VITE=true
 DEV_PAGES_VITE_URL=http://localhost:5173
 DEV_PAGES_MANIFEST_PATH=frontend/dev-pages/dist/.vite/manifest.json
 
+# Execute enabled acquisition rules on this interval (seconds). Set a value that
+# respects each indexer's API limits.
+ACQUISITION_AUTOMATION_INTERVAL_SECONDS=300
+
 # Prefer file-based PowerSync keys for local development.
 # Generate them with: ./scripts/generate_dev_powersync_keys.sh
 POWERSYNC_JWT_PRIVATE_KEY_FILE=.local/powersync/private.pem
diff --git a/alembic/versions/d4e5f6a7b8c9_add_acquisition.py b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
new file mode 100644
index 0000000..9a785c2
--- /dev/null
+++ b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
@@ -0,0 +1,79 @@
+"""add acquisition endpoints, rules and jobs
+
+Revision ID: d4e5f6a7b8c9
+Revises: c3f8b2a9d1e4
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "d4e5f6a7b8c9"
+down_revision: str | Sequence[str] | None = "c3f8b2a9d1e4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create private acquisition integration tables."""
+    op.create_table(
+        "acquisition_endpoints",
+        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(120), nullable=False),
+        sa.Column("kind", sa.String(32), nullable=False),
+        sa.Column("base_url", sa.String(2048), nullable=False),
+        sa.Column("credentials", postgresql.JSONB(), nullable=True),
+        sa.Column("settings", postgresql.JSONB(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("endpoint_id"),
+    )
+    op.create_index("ix_acquisition_endpoints_owner_user_id", "acquisition_endpoints", ["owner_user_id"])
+    op.create_table(
+        "acquisition_rules",
+        sa.Column("rule_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(120), nullable=False),
+        sa.Column("query", sa.String(500), nullable=False),
+        sa.Column("endpoint_ids", postgresql.JSONB(), nullable=True),
+        sa.Column("download_client_id", sa.Uuid(), nullable=True),
+        sa.Column("filters", postgresql.JSONB(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["download_client_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("rule_id"),
+    )
+    op.create_index("ix_acquisition_rules_owner_user_id", "acquisition_rules", ["owner_user_id"])
+    op.create_table(
+        "acquisition_jobs",
+        sa.Column("job_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("rule_id", sa.Uuid(), nullable=True),
+        sa.Column("title", sa.String(500), nullable=False),
+        sa.Column("download_url", sa.Text(), nullable=False),
+        sa.Column("status", sa.String(32), server_default=sa.text("'queued'"), nullable=False),
+        sa.Column("client_reference", sa.String(255), nullable=True),
+        sa.Column("error", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["endpoint_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["rule_id"], ["acquisition_rules.rule_id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("job_id"),
+    )
+    op.create_index("ix_acquisition_jobs_owner_user_id", "acquisition_jobs", ["owner_user_id"])
+
+
+def downgrade() -> None:
+    """Drop acquisition tables."""
+    op.drop_table("acquisition_jobs")
+    op.drop_table("acquisition_rules")
+    op.drop_table("acquisition_endpoints")
diff --git a/papyrus/api/routes/__init__.py b/papyrus/api/routes/__init__.py
index 7835593..19772d8 100644
--- a/papyrus/api/routes/__init__.py
+++ b/papyrus/api/routes/__init__.py
@@ -3,6 +3,7 @@
 from fastapi import APIRouter, FastAPI
 
 from papyrus.api.routes import (
+    acquisition,
     annotations,
     auth,
     bookmarks,
@@ -26,6 +27,7 @@
 
 api_router = APIRouter()
 
+api_router.include_router(acquisition.router, prefix="/acquisition", tags=["Acquisition"])
 api_router.include_router(auth.router, prefix="/auth", tags=["Auth"])
 api_router.include_router(users.router, prefix="/users", tags=["Users"])
 api_router.include_router(books.router, prefix="/books", tags=["Books"])
diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py
new file mode 100644
index 0000000..bcbbadf
--- /dev/null
+++ b/papyrus/api/routes/acquisition.py
@@ -0,0 +1,251 @@
+"""Private torrent and indexer acquisition endpoints."""
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.api.deps import CurrentUserId
+from papyrus.core.database import get_db
+from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload
+from papyrus.models.acquisition import AcquisitionEndpoint as AcquisitionEndpointModel
+from papyrus.models.acquisition import AcquisitionJob as AcquisitionJobModel
+from papyrus.models.acquisition import AcquisitionRule as AcquisitionRuleModel
+from papyrus.schemas.acquisition import (
+    AcquisitionCapabilities,
+    AcquisitionEndpoint,
+    AcquisitionEndpointCreate,
+    AcquisitionEndpointUpdate,
+    AcquisitionJob,
+    AcquisitionRule,
+    AcquisitionRuleCreate,
+    ArrCommandRequest,
+    Release,
+    SearchRequest,
+    SubmitRequest,
+)
+from papyrus.services.acquisition import (
+    dispatch_arr_command,
+    owned_endpoint,
+    run_rule,
+    search_endpoint,
+    submit_to_client,
+)
+
+router = APIRouter()
+DbSession = Annotated[AsyncSession, Depends(get_db)]
+
+
+def _credentials(api_key: str | None, username: str | None, password: str | None) -> dict[str, str] | None:
+    values = {
+        key: value for key, value in {"api_key": api_key, "username": username, "password": password}.items() if value
+    }
+    if not values:
+        return None
+    return {"encrypted": encrypt_secret_payload(values)}
+
+
+def _merge_credentials(endpoint: AcquisitionEndpointModel, replacement: dict[str, str] | None) -> None:
+    if replacement is None:
+        return
+    current = endpoint.credentials or {}
+    if encrypted := current.get("encrypted"):
+        current = decrypt_secret_payload(encrypted)
+    endpoint.credentials = {
+        "encrypted": encrypt_secret_payload({**current, **decrypt_secret_payload(replacement["encrypted"])})
+    }
+
+
+@router.get("/capabilities", response_model=AcquisitionCapabilities)
+async def acquisition_capabilities() -> AcquisitionCapabilities:
+    return AcquisitionCapabilities(
+        endpoint_kinds=[
+            "qbittorrent",
+            "transmission",
+            "deluge",
+            "prowlarr",
+            "torznab",
+            "readarr",
+            "sonarr",
+            "radarr",
+            "lidarr",
+            "whisparr",
+        ],
+        indexer_kinds=["prowlarr", "torznab"],
+        download_client_kinds=["qbittorrent", "transmission", "deluge"],
+        arr_kinds=["readarr", "sonarr", "radarr", "lidarr", "whisparr"],
+        arr_commands={
+            "readarr": ["AuthorSearch", "BookSearch"],
+            "sonarr": ["SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"],
+            "radarr": ["MoviesSearch", "MissingMoviesSearch"],
+            "lidarr": ["ArtistSearch", "AlbumSearch", "MissingAlbumSearch"],
+            "whisparr": ["SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"],
+        },
+    )
+
+
+@router.get("/endpoints", response_model=list[AcquisitionEndpoint])
+async def list_endpoints(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionEndpointModel]:
+    result = await db.execute(
+        select(AcquisitionEndpointModel)
+        .where(AcquisitionEndpointModel.owner_user_id == user_id)
+        .order_by(AcquisitionEndpointModel.name)
+    )
+    return list(result.scalars())
+
+
+@router.post("/endpoints", response_model=AcquisitionEndpoint, status_code=status.HTTP_201_CREATED)
+async def create_endpoint(
+    user_id: CurrentUserId, request: AcquisitionEndpointCreate, db: DbSession
+) -> AcquisitionEndpointModel:
+    endpoint = AcquisitionEndpointModel(
+        owner_user_id=user_id,
+        name=request.name,
+        kind=request.kind.value,
+        base_url=str(request.base_url),
+        credentials=_credentials(
+            request.api_key.get_secret_value() if request.api_key else None,
+            request.username.get_secret_value() if request.username else None,
+            request.password.get_secret_value() if request.password else None,
+        ),
+        settings=request.settings,
+    )
+    db.add(endpoint)
+    await db.commit()
+    await db.refresh(endpoint)
+    return endpoint
+
+
+@router.patch("/endpoints/{endpoint_id}", response_model=AcquisitionEndpoint)
+async def update_endpoint(
+    user_id: CurrentUserId, endpoint_id: UUID, request: AcquisitionEndpointUpdate, db: DbSession
+) -> AcquisitionEndpointModel:
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    updates = request.model_dump(exclude_unset=True, exclude={"api_key", "username", "password"})
+    for field, value in updates.items():
+        setattr(endpoint, field, str(value) if field == "base_url" else value)
+    replacement = _credentials(
+        request.api_key.get_secret_value() if request.api_key else None,
+        request.username.get_secret_value() if request.username else None,
+        request.password.get_secret_value() if request.password else None,
+    )
+    _merge_credentials(endpoint, replacement)
+    await db.commit()
+    await db.refresh(endpoint)
+    return endpoint
+
+
+@router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSession) -> None:
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    await db.delete(endpoint)
+    await db.commit()
+
+
+@router.post("/search", response_model=list[Release])
+async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: DbSession) -> list[Release]:
+    statement = select(AcquisitionEndpointModel).where(
+        AcquisitionEndpointModel.owner_user_id == user_id,
+        AcquisitionEndpointModel.enabled.is_(True),
+        AcquisitionEndpointModel.kind.in_(("prowlarr", "torznab")),
+    )
+    if request.endpoint_ids:
+        statement = statement.where(AcquisitionEndpointModel.endpoint_id.in_(request.endpoint_ids))
+    result = await db.execute(statement)
+    releases: list[Release] = []
+    for endpoint in result.scalars():
+        releases.extend(await search_endpoint(endpoint, request.query))
+    return releases
+
+
+@router.post("/submissions", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+async def submit_release(user_id: CurrentUserId, request: SubmitRequest, db: DbSession) -> AcquisitionJobModel:
+    endpoint = await owned_endpoint(db, user_id, request.endpoint_id)
+    job = AcquisitionJobModel(
+        owner_user_id=user_id, endpoint_id=endpoint.endpoint_id, title=request.title, download_url=request.download_url
+    )
+    db.add(job)
+    try:
+        job.client_reference = await submit_to_client(
+            endpoint, request.download_url, request.category, request.save_path
+        )
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    await db.commit()
+    await db.refresh(job)
+    return job
+
+
+@router.post("/arr/{endpoint_id}/commands", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+async def run_arr_command(
+    user_id: CurrentUserId, endpoint_id: UUID, request: ArrCommandRequest, db: DbSession
+) -> AcquisitionJobModel:
+    """Delegate a managed search/acquisition to Readarr or another Arr app."""
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    job = AcquisitionJobModel(
+        owner_user_id=user_id,
+        endpoint_id=endpoint.endpoint_id,
+        title=request.command,
+        download_url=f"arr-command:{request.command}",
+    )
+    db.add(job)
+    try:
+        job.client_reference = await dispatch_arr_command(endpoint, request.command, request.ids)
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    await db.commit()
+    await db.refresh(job)
+    return job
+
+
+@router.get("/jobs", response_model=list[AcquisitionJob])
+async def list_jobs(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionJobModel]:
+    result = await db.execute(
+        select(AcquisitionJobModel)
+        .where(AcquisitionJobModel.owner_user_id == user_id)
+        .order_by(AcquisitionJobModel.created_at.desc())
+    )
+    return list(result.scalars())
+
+
+@router.get("/rules", response_model=list[AcquisitionRule])
+async def list_rules(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionRuleModel]:
+    result = await db.execute(select(AcquisitionRuleModel).where(AcquisitionRuleModel.owner_user_id == user_id))
+    return list(result.scalars())
+
+
+@router.post("/rules", response_model=AcquisitionRule, status_code=status.HTTP_201_CREATED)
+async def create_rule(user_id: CurrentUserId, request: AcquisitionRuleCreate, db: DbSession) -> AcquisitionRuleModel:
+    await owned_endpoint(db, user_id, request.download_client_id)
+    rule = AcquisitionRuleModel(
+        owner_user_id=user_id,
+        name=request.name,
+        query=request.query,
+        endpoint_ids=[str(value) for value in request.endpoint_ids] if request.endpoint_ids else None,
+        download_client_id=request.download_client_id,
+        filters=request.filters,
+        enabled=request.enabled,
+    )
+    db.add(rule)
+    await db.commit()
+    await db.refresh(rule)
+    return rule
+
+
+@router.post("/rules/{rule_id}/run", response_model=list[AcquisitionJob])
+async def run_acquisition_rule(user_id: CurrentUserId, rule_id: UUID, db: DbSession) -> list[AcquisitionJobModel]:
+    result = await db.execute(
+        select(AcquisitionRuleModel).where(
+            AcquisitionRuleModel.rule_id == rule_id, AcquisitionRuleModel.owner_user_id == user_id
+        )
+    )
+    rule = result.scalar_one_or_none()
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Acquisition rule not found")
+    return await run_rule(db, rule)
diff --git a/papyrus/config.py b/papyrus/config.py
index 4208038..b6aa29d 100644
--- a/papyrus/config.py
+++ b/papyrus/config.py
@@ -71,6 +71,7 @@ class Settings(BaseSettings):
     dev_pages_use_vite: bool = False
     dev_pages_vite_url: str = "http://localhost:5173"
     dev_pages_manifest_path: str = "frontend/dev-pages/dist/.vite/manifest.json"
+    acquisition_automation_interval_seconds: int = 300
 
     @field_validator("debug", mode="before")
     @classmethod
diff --git a/papyrus/core/security.py b/papyrus/core/security.py
index 8334630..b35cd48 100644
--- a/papyrus/core/security.py
+++ b/papyrus/core/security.py
@@ -1,13 +1,17 @@
 """Authentication and token security helpers."""
 
+from base64 import urlsafe_b64encode
 from datetime import UTC, datetime, timedelta
 from functools import lru_cache
 from hashlib import sha256
+from json import dumps as json_dumps
+from json import loads as json_loads
 from pathlib import Path
 from secrets import token_urlsafe
 from typing import Any
 
 import jwt
+from cryptography.fernet import Fernet, InvalidToken
 from cryptography.hazmat.primitives import serialization
 from jwt.algorithms import RSAAlgorithm
 from pwdlib import PasswordHash
@@ -125,6 +129,29 @@ def generate_opaque_token() -> str:
     return token_urlsafe(48)
 
 
+@lru_cache
+def _get_credentials_cipher() -> Fernet:
+    key = urlsafe_b64encode(sha256(get_settings().secret_key.encode("utf-8")).digest())
+    return Fernet(key)
+
+
+def encrypt_secret_payload(payload: dict[str, str]) -> str:
+    return _get_credentials_cipher().encrypt(json_dumps(payload).encode("utf-8")).decode("utf-8")
+
+
+def decrypt_secret_payload(token: str) -> dict[str, str]:
+    try:
+        decoded = _get_credentials_cipher().decrypt(token.encode("utf-8")).decode("utf-8")
+    except InvalidToken as exc:
+        raise ValueError("Encrypted credential payload is invalid") from exc
+    payload = json_loads(decoded)
+    if not isinstance(payload, dict) or not all(
+        isinstance(key, str) and isinstance(value, str) for key, value in payload.items()
+    ):
+        raise ValueError("Encrypted credential payload has an invalid shape")
+    return payload
+
+
 def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str:
     settings = get_settings()
     ttl = expires_delta or timedelta(minutes=settings.access_token_expire_minutes)
diff --git a/papyrus/main.py b/papyrus/main.py
index f3db84d..9bec526 100644
--- a/papyrus/main.py
+++ b/papyrus/main.py
@@ -1,8 +1,9 @@
 """FastAPI application factory and configuration."""
 
+import asyncio
 import logging
 from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 
 import structlog
 from fastapi import FastAPI, Request, status
@@ -16,9 +17,11 @@
 
 from papyrus.api.routes import api_router, include_debug_routers
 from papyrus.config import get_settings
+from papyrus.core.database import async_session_maker
 from papyrus.core.dev_pages import DEV_PAGES_DIST_DIR, DEV_PAGES_STATIC_URL
 from papyrus.core.exceptions import AppError
 from papyrus.core.rate_limit import limiter
+from papyrus.services.acquisition import run_enabled_rules
 
 settings = get_settings()
 
@@ -40,9 +43,20 @@ def configure_logging() -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan events."""
-    # Startup
-    yield
-    # Shutdown
+
+    async def acquisition_worker() -> None:
+        while True:
+            async with async_session_maker() as session:
+                await run_enabled_rules(session)
+            await asyncio.sleep(settings.acquisition_automation_interval_seconds)
+
+    worker = asyncio.create_task(acquisition_worker())
+    try:
+        yield
+    finally:
+        worker.cancel()
+        with suppress(asyncio.CancelledError):
+            await worker
 
 
 def create_app() -> FastAPI:
diff --git a/papyrus/models/__init__.py b/papyrus/models/__init__.py
index be17f15..a94957f 100644
--- a/papyrus/models/__init__.py
+++ b/papyrus/models/__init__.py
@@ -1,4 +1,5 @@
 from papyrus.core.database import Base
+from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
 from papyrus.models.auth import AuthExchangeCode, AuthSession, EmailActionToken, PasswordCredential, UserIdentity
 from papyrus.models.media import MediaAsset
 from papyrus.models.powersync_demo import PowerSyncDemoItem
@@ -6,6 +7,9 @@
 from papyrus.models.user import User
 
 __all__ = [
+    "AcquisitionEndpoint",
+    "AcquisitionJob",
+    "AcquisitionRule",
     "AuthExchangeCode",
     "AuthSession",
     "Base",
diff --git a/papyrus/models/acquisition.py b/papyrus/models/acquisition.py
new file mode 100644
index 0000000..66e2e9d
--- /dev/null
+++ b/papyrus/models/acquisition.py
@@ -0,0 +1,63 @@
+"""Acquisition sources, download clients and automatic acquisition rules."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, Uuid, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from papyrus.core.database import Base
+
+
+class AcquisitionEndpoint(Base):
+    """A private indexer, Prowlarr instance, or download client connection."""
+
+    __tablename__ = "acquisition_endpoints"
+
+    endpoint_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
+    owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
+    name: Mapped[str] = mapped_column(String(120), nullable=False)
+    kind: Mapped[str] = mapped_column(String(32), nullable=False)
+    base_url: Mapped[str] = mapped_column(String(2048), nullable=False)
+    credentials: Mapped[dict[str, str] | None] = mapped_column(JSONB, nullable=True)
+    settings: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+
+class AcquisitionRule(Base):
+    """A saved search that can automatically submit matching releases."""
+
+    __tablename__ = "acquisition_rules"
+
+    rule_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
+    owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
+    name: Mapped[str] = mapped_column(String(120), nullable=False)
+    query: Mapped[str] = mapped_column(String(500), nullable=False)
+    endpoint_ids: Mapped[list[str] | None] = mapped_column(JSONB, nullable=True)
+    download_client_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_endpoints.endpoint_id"))
+    filters: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+
+class AcquisitionJob(Base):
+    """An auditable submission to a download client."""
+
+    __tablename__ = "acquisition_jobs"
+
+    job_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
+    owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
+    endpoint_id: Mapped[UUID] = mapped_column(Uuid, ForeignKey("acquisition_endpoints.endpoint_id"), nullable=False)
+    rule_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_rules.rule_id", ondelete="SET NULL"))
+    title: Mapped[str] = mapped_column(String(500), nullable=False)
+    download_url: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", server_default="queued")
+    client_reference: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
diff --git a/papyrus/schemas/acquisition.py b/papyrus/schemas/acquisition.py
new file mode 100644
index 0000000..850f118
--- /dev/null
+++ b/papyrus/schemas/acquisition.py
@@ -0,0 +1,153 @@
+"""HTTP schemas for private acquisition integrations."""
+
+from datetime import datetime
+from enum import StrEnum
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator
+
+
+class EndpointKind(StrEnum):
+    QBITTORRENT = "qbittorrent"
+    TRANSMISSION = "transmission"
+    DELUGE = "deluge"
+    PROWLARR = "prowlarr"
+    TORZNAB = "torznab"
+    READARR = "readarr"
+    SONARR = "sonarr"
+    RADARR = "radarr"
+    LIDARR = "lidarr"
+    WHISPARR = "whisparr"
+
+
+class AcquisitionCapabilities(BaseModel):
+    enabled: bool = True
+    endpoint_kinds: list[EndpointKind]
+    indexer_kinds: list[EndpointKind]
+    download_client_kinds: list[EndpointKind]
+    arr_kinds: list[EndpointKind]
+    arr_commands: dict[EndpointKind, list[str]]
+
+
+class AcquisitionEndpointCreate(BaseModel):
+    name: str = Field(min_length=1, max_length=120)
+    kind: EndpointKind
+    base_url: HttpUrl
+    api_key: SecretStr | None = None
+    username: SecretStr | None = None
+    password: SecretStr | None = None
+    settings: dict[str, object] | None = None
+
+    @field_validator("base_url")
+    @classmethod
+    def validate_endpoint_url(cls, value: HttpUrl) -> HttpUrl:
+        return validate_endpoint_url(value)
+
+
+class AcquisitionEndpointUpdate(BaseModel):
+    name: str | None = Field(None, min_length=1, max_length=120)
+    base_url: HttpUrl | None = None
+    api_key: SecretStr | None = None
+    username: SecretStr | None = None
+    password: SecretStr | None = None
+    settings: dict[str, object] | None = None
+    enabled: bool | None = None
+
+    @field_validator("base_url")
+    @classmethod
+    def validate_endpoint_url(cls, value: HttpUrl | None) -> HttpUrl | None:
+        if value is None:
+            return None
+        return validate_endpoint_url(value)
+
+
+class AcquisitionEndpoint(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    endpoint_id: UUID
+    name: str
+    kind: EndpointKind
+    base_url: str
+    enabled: bool
+    settings: dict[str, object] | None = None
+    created_at: datetime | None = None
+
+
+class Release(BaseModel):
+    title: str
+    download_url: str
+    protocol: str
+    indexer: str
+    size_bytes: int | None = None
+    seeders: int | None = None
+    publish_date: datetime | None = None
+
+
+class SearchRequest(BaseModel):
+    query: str = Field(min_length=1, max_length=500)
+    endpoint_ids: list[UUID] | None = None
+
+
+class SubmitRequest(BaseModel):
+    endpoint_id: UUID
+    title: str = Field(min_length=1, max_length=500)
+    download_url: str = Field(min_length=1, max_length=4096)
+    category: str | None = Field(None, max_length=100)
+    save_path: str | None = Field(None, max_length=1024)
+
+    @field_validator("download_url")
+    @classmethod
+    def validate_torrent_download_url(cls, value: str) -> str:
+        if value.startswith("magnet:"):
+            return value
+        if value.startswith(("http://", "https://")):
+            return value
+        raise ValueError("download_url must be a magnet, HTTP, or HTTPS torrent URL")
+
+
+class ArrCommandRequest(BaseModel):
+    """A Servarr command and the IDs scoped to that application's library."""
+
+    command: str = Field(min_length=1, max_length=80)
+    ids: list[int] = Field(default_factory=list, max_length=100)
+
+
+class AcquisitionRuleCreate(BaseModel):
+    name: str = Field(min_length=1, max_length=120)
+    query: str = Field(min_length=1, max_length=500)
+    endpoint_ids: list[UUID] | None = None
+    download_client_id: UUID
+    filters: dict[str, object] | None = None
+    enabled: bool = True
+
+
+class AcquisitionRule(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    rule_id: UUID
+    name: str
+    query: str
+    endpoint_ids: list[UUID] | None = None
+    download_client_id: UUID | None
+    filters: dict[str, object] | None = None
+    enabled: bool
+    last_run_at: datetime | None = None
+
+
+class AcquisitionJob(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    job_id: UUID
+    endpoint_id: UUID
+    rule_id: UUID | None
+    title: str
+    download_url: str
+    status: str
+    client_reference: str | None = None
+    error: str | None = None
+    created_at: datetime | None = None
+
+
+def validate_endpoint_url(value: HttpUrl) -> HttpUrl:
+    if value.username or value.password:
+        raise ValueError("Endpoint URL must not include credentials")
+    if value.scheme not in {"http", "https"}:
+        raise ValueError("Endpoint URL must use HTTP or HTTPS")
+    return value
diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py
new file mode 100644
index 0000000..e6a3cfd
--- /dev/null
+++ b/papyrus/services/acquisition.py
@@ -0,0 +1,336 @@
+"""Adapters for indexer protocols and self-hosted download clients."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+from datetime import UTC, datetime
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode, urljoin
+from urllib.request import Request, urlopen
+from xml.etree import ElementTree
+
+from fastapi import HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.core.security import decrypt_secret_payload
+from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
+from papyrus.schemas.acquisition import Release
+
+
+def _url(endpoint: AcquisitionEndpoint, path: str) -> str:
+    return urljoin(endpoint.base_url.rstrip("/") + "/", path.lstrip("/"))
+
+
+async def _request(
+    url: str, *, method: str = "GET", headers: dict[str, str] | None = None, body: bytes | None = None
+) -> tuple[int, dict[str, str], bytes]:
+    """Perform a bounded blocking HTTP request off the event loop."""
+
+    def send() -> tuple[int, dict[str, str], bytes]:
+        request = Request(url, data=body, headers=headers or {}, method=method)
+        try:
+            with urlopen(request, timeout=15) as response:  # noqa: S310 - user-owned self-hosted integrations
+                return response.status, dict(response.headers.items()), response.read(5_000_000)
+        except HTTPError as exc:
+            return exc.code, dict(exc.headers.items()), exc.read(1_000_000)
+        except URLError as exc:
+            raise HTTPException(status_code=502, detail=f"Integration request failed: {exc.reason}") from exc
+
+    return await asyncio.to_thread(send)
+
+
+def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]:
+    credentials = endpoint.credentials or {}
+    encrypted = credentials.get("encrypted")
+    if encrypted is None:
+        return credentials
+    try:
+        return decrypt_secret_payload(encrypted)
+    except ValueError as exc:
+        raise HTTPException(status_code=500, detail="Stored integration credentials are invalid") from exc
+
+
+async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Release]:
+    """Search Prowlarr or a Torznab-compatible torrent indexer."""
+    credentials = _credentials(endpoint)
+    if endpoint.kind == "prowlarr":
+        request_url = _url(endpoint, f"api/v1/search?{urlencode({'query': query})}")
+        response_status, _, payload = await _request(request_url, headers={"X-Api-Key": credentials.get("api_key", "")})
+        if response_status >= 400:
+            raise HTTPException(status_code=502, detail="Prowlarr search failed")
+        data = json.loads(payload)
+        return [
+            Release(
+                title=item.get("title", "Untitled"),
+                download_url=item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid", ""),
+                protocol="torrent",
+                indexer=item.get("indexer", "Prowlarr"),
+                size_bytes=item.get("size"),
+                seeders=item.get("seeders"),
+            )
+            for item in data
+            if (item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid"))
+            and item.get("protocol", "torrent") == "torrent"
+        ]
+
+    params = urlencode({"t": "search", "q": query, "apikey": credentials.get("api_key", "")})
+    response_status, _, payload = await _request(_url(endpoint, f"api?{params}"))
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} search failed")
+    return _parse_torznab(payload, endpoint.name)
+
+
+def _parse_torznab(payload: bytes, indexer: str) -> list[Release]:
+    try:
+        root = ElementTree.fromstring(payload)
+    except ElementTree.ParseError as exc:
+        raise HTTPException(status_code=502, detail="Indexer returned invalid XML") from exc
+    releases: list[Release] = []
+    for item in root.findall(".//item"):
+        enclosure = item.find("enclosure")
+        link = (enclosure.get("url") if enclosure is not None else None) or item.findtext("link")
+        if not link:
+            continue
+        attrs = {child.attrib.get("name"): child.attrib.get("value") for child in item if child.tag.endswith("attr")}
+        size = attrs.get("size")
+        seeders = attrs.get("seeders")
+        releases.append(
+            Release(
+                title=item.findtext("title") or "Untitled",
+                download_url=link,
+                protocol="torrent",
+                indexer=indexer,
+                size_bytes=int(size) if size is not None and size.isdigit() else None,
+                seeders=int(seeders) if seeders is not None and seeders.isdigit() else None,
+            )
+        )
+    return releases
+
+
+async def submit_to_client(
+    endpoint: AcquisitionEndpoint, download_url: str, category: str | None, save_path: str | None
+) -> str | None:
+    """Submit a magnet or torrent URL to a supported BitTorrent client."""
+    if endpoint.kind == "qbittorrent":
+        return await _submit_qbittorrent(endpoint, download_url, category, save_path)
+    if endpoint.kind == "transmission":
+        return await _submit_transmission(endpoint, download_url, save_path)
+    if endpoint.kind == "deluge":
+        return await _submit_deluge(endpoint, download_url, save_path)
+    raise HTTPException(status_code=422, detail="Endpoint is not a download client")
+
+
+async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids: list[int]) -> str | None:
+    """Start an acquisition/search command in a Servarr application.
+
+    The Arr apps own their library and release-grab decisions, so they must be
+    driven through their command API rather than handed an arbitrary magnet.
+    """
+    if endpoint.kind not in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}:
+        raise HTTPException(status_code=422, detail="Endpoint is not a Servarr application")
+
+    allowed_commands = {
+        "readarr": {"AuthorSearch", "BookSearch"},
+        "sonarr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"},
+        "radarr": {"MoviesSearch", "MissingMoviesSearch"},
+        "lidarr": {"ArtistSearch", "AlbumSearch", "MissingAlbumSearch"},
+        "whisparr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"},
+    }
+    if command not in allowed_commands[endpoint.kind]:
+        raise HTTPException(status_code=422, detail="Command is not supported by this Servarr application")
+
+    id_field = {
+        "AuthorSearch": "authorIds",
+        "BookSearch": "bookIds",
+        "SeriesSearch": "seriesId",
+        "EpisodeSearch": "episodeIds",
+        "MissingEpisodeSearch": "seriesId",
+        "MoviesSearch": "movieIds",
+        "MissingMoviesSearch": "movieIds",
+        "ArtistSearch": "artistIds",
+        "AlbumSearch": "albumIds",
+        "MissingAlbumSearch": "artistIds",
+    }[command]
+    payload: dict[str, object] = {"name": command}
+    if ids:
+        payload[id_field] = ids[0] if id_field in {"seriesId"} else ids
+
+    response_status, _, response_payload = await _request(
+        _url(endpoint, "api/v1/command"),
+        method="POST",
+        headers={"Content-Type": "application/json", "X-Api-Key": _credentials(endpoint).get("api_key", "")},
+        body=json.dumps(payload).encode(),
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} command failed")
+    response = json.loads(response_payload)
+    return str(response.get("id")) if response.get("id") is not None else None
+
+
+async def _submit_qbittorrent(
+    endpoint: AcquisitionEndpoint, download_url: str, category: str | None, save_path: str | None
+) -> str | None:
+    credentials = _credentials(endpoint)
+    login = urlencode(
+        {"username": credentials.get("username", ""), "password": credentials.get("password", "")}
+    ).encode()
+    response_status, headers, _ = await _request(
+        _url(endpoint, "api/v2/auth/login"),
+        method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        body=login,
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="qBittorrent authentication failed")
+    payload = {"urls": download_url}
+    if category:
+        payload["category"] = category
+    if save_path:
+        payload["savepath"] = save_path
+    response_status, _, _ = await _request(
+        _url(endpoint, "api/v2/torrents/add"),
+        method="POST",
+        headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Cookie": headers.get("Set-Cookie", "").split(";", 1)[0],
+        },
+        body=urlencode(payload).encode(),
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="qBittorrent rejected the release")
+    return None
+
+
+async def _submit_transmission(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None:
+    credentials = _credentials(endpoint)
+    arguments: dict[str, str] = {"filename": download_url}
+    if save_path:
+        arguments["download-dir"] = save_path
+    body = json.dumps({"method": "torrent-add", "arguments": arguments}).encode()
+    headers = {"Content-Type": "application/json"}
+    if credentials.get("username"):
+        token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode()
+        headers["Authorization"] = f"Basic {token}"
+    response_status, response_headers, payload = await _request(
+        _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body
+    )
+    if response_status == 409:
+        headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "")
+        response_status, _, payload = await _request(
+            _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body
+        )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="Transmission rejected the release")
+    return str(json.loads(payload).get("arguments", {}).get("torrent-added", {}).get("hashString") or "") or None
+
+
+async def _submit_deluge(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None:
+    credentials = _credentials(endpoint)
+    headers = {"Content-Type": "application/json"}
+    login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode()
+    response_status, response_headers, _ = await _request(
+        _url(endpoint, "json"), method="POST", headers=headers, body=login
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="Deluge authentication failed")
+    options = {"download_location": save_path} if save_path else {}
+    body = json.dumps({"method": "core.add_torrent_magnet", "params": [download_url, options], "id": 2}).encode()
+    headers["Cookie"] = response_headers.get("Set-Cookie", "").split(";", 1)[0]
+    response_status, _, payload = await _request(_url(endpoint, "json"), method="POST", headers=headers, body=body)
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="Deluge rejected the release")
+    result = json.loads(payload).get("result")
+    return str(result) if result is not None else None
+
+
+async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> AcquisitionEndpoint:
+    result = await session.execute(
+        select(AcquisitionEndpoint).where(
+            AcquisitionEndpoint.endpoint_id == endpoint_id, AcquisitionEndpoint.owner_user_id == owner_user_id
+        )
+    )
+    endpoint = result.scalar_one_or_none()
+    if endpoint is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition endpoint not found")
+    return endpoint
+
+
+async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]:
+    """Run one rule once; callers may schedule this from their worker/cron service."""
+    client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id)
+    if client.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}:
+        filters = rule.filters or {}
+        command = filters.get("arr_command")
+        ids = filters.get("arr_ids", [])
+        if (
+            not isinstance(command, str)
+            or not isinstance(ids, list)
+            or not all(isinstance(value, int) for value in ids)
+        ):
+            raise HTTPException(
+                status_code=422,
+                detail="Arr rules require filters.arr_command and filters.arr_ids",
+            )
+        job = AcquisitionJob(
+            owner_user_id=rule.owner_user_id,
+            endpoint_id=client.endpoint_id,
+            rule_id=rule.rule_id,
+            title=command,
+            download_url=f"arr-command:{command}",
+        )
+        session.add(job)
+        try:
+            job.client_reference = await dispatch_arr_command(client, command, ids)
+            job.status = "submitted"
+        except HTTPException as exc:
+            job.status = "failed"
+            job.error = str(exc.detail)
+        rule.last_run_at = datetime.now(UTC)
+        await session.commit()
+        return [job]
+
+    endpoint_ids = rule.endpoint_ids or []
+    result = await session.execute(
+        select(AcquisitionEndpoint).where(
+            AcquisitionEndpoint.owner_user_id == rule.owner_user_id,
+            AcquisitionEndpoint.endpoint_id.in_(endpoint_ids),
+            AcquisitionEndpoint.enabled.is_(True),
+        )
+    )
+    releases = [release for endpoint in result.scalars() for release in await search_endpoint(endpoint, rule.query)]
+    if not releases:
+        rule.last_run_at = datetime.now(UTC)
+        await session.commit()
+        return []
+    selected = sorted(releases, key=lambda release: (release.seeders or 0, release.size_bytes or 0), reverse=True)[0]
+    job = AcquisitionJob(
+        owner_user_id=rule.owner_user_id,
+        endpoint_id=client.endpoint_id,
+        rule_id=rule.rule_id,
+        title=selected.title,
+        download_url=selected.download_url,
+    )
+    session.add(job)
+    try:
+        job.client_reference = await submit_to_client(client, selected.download_url, None, None)
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    rule.last_run_at = datetime.now(UTC)
+    await session.commit()
+    return [job]
+
+
+async def run_enabled_rules(session: AsyncSession) -> None:
+    """Run enabled rules, isolating a failed remote integration from the others."""
+    result = await session.execute(select(AcquisitionRule).where(AcquisitionRule.enabled.is_(True)))
+    for rule in result.scalars():
+        try:
+            await run_rule(session, rule)
+        except HTTPException:
+            await session.rollback()
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
new file mode 100644
index 0000000..d471034
--- /dev/null
+++ b/tests/api/routes/test_acquisition.py
@@ -0,0 +1,116 @@
+"""Tests for private BitTorrent acquisition configuration."""
+
+from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.core.security import decrypt_secret_payload
+from papyrus.models.acquisition import AcquisitionEndpoint
+
+
+async def test_capabilities_advertise_torrent_only_scope(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    response = await client.get("/v1/acquisition/capabilities", headers=auth_headers)
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["indexer_kinds"] == ["prowlarr", "torznab"]
+    assert body["download_client_kinds"] == ["qbittorrent", "transmission", "deluge"]
+    assert "newznab" not in body["endpoint_kinds"]
+    assert body["arr_commands"]["readarr"] == ["AuthorSearch", "BookSearch"]
+
+
+async def test_create_and_list_endpoint_hides_and_encrypts_credentials(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    db_session: AsyncSession,
+) -> None:
+    """Credentials can be configured but must never be returned to a client."""
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Home qBittorrent",
+            "kind": "qbittorrent",
+            "base_url": "http://127.0.0.1:8080",
+            "username": "admin",
+            "password": "secret",
+        },
+    )
+
+    assert response.status_code == 201
+    assert "password" not in response.text
+    assert "username" not in response.text
+    endpoint = response.json()
+
+    response = await client.get("/v1/acquisition/endpoints", headers=auth_headers)
+    assert response.status_code == 200
+    assert response.json() == [endpoint]
+
+    stored = (await db_session.execute(select(AcquisitionEndpoint))).scalar_one()
+    assert stored.credentials is not None
+    assert "password" not in stored.credentials
+    assert decrypt_secret_payload(stored.credentials["encrypted"]) == {
+        "username": "admin",
+        "password": "secret",
+    }
+
+
+async def test_create_rule_requires_owned_download_client(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    """Rules cannot target a client belonging to another user or a missing client."""
+    response = await client.post(
+        "/v1/acquisition/rules",
+        headers=auth_headers,
+        json={
+            "name": "Monthly reading",
+            "query": "example book",
+            "download_client_id": "00000000-0000-0000-0000-000000000001",
+        },
+    )
+
+    assert response.status_code == 404
+
+
+async def test_readarr_endpoint_is_a_supported_integration(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    """Readarr uses its own acquisition workflow and is accepted as an endpoint."""
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Reading automation",
+            "kind": "readarr",
+            "base_url": "http://readarr.local:8787",
+            "api_key": "private-key",
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.json()["kind"] == "readarr"
+
+
+async def test_newznab_endpoint_is_rejected(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Usenet",
+            "kind": "newznab",
+            "base_url": "http://newznab.local",
+            "api_key": "private-key",
+        },
+    )
+
+    assert response.status_code == 422
+
+
+async def test_endpoint_url_rejects_embedded_credentials(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Bad URL",
+            "kind": "qbittorrent",
+            "base_url": "http://user:secret@127.0.0.1:8080",
+        },
+    )
+
+    assert response.status_code == 422