Create source connection check

Test a source connection by creating a connection check for the specified source connector.

POST /api/v1/sources/{source_id}/connection-check

cURL

curl --request POST \
  --url "${UNSTRUCTURED_API_URL}/api/v1/sources/a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d/connection-check" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateConnectionCheckSourcesRequest

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL"),
)

response = client.sources.create_source_connection_check(
    request=CreateConnectionCheckSourcesRequest(
        source_id="a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d"
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateConnectionCheckSourcesRequest

async def create_source_connection_check():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.sources.create_source_connection_check_async(
        request=CreateConnectionCheckSourcesRequest(
            source_id="a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d"
        )
    )
    print(response)

asyncio.run(create_source_connection_check())

Response

{
  "id": "c2b3a4d5-6e7f-8a9b-0c1d-2e3f4a5b6c7d",
  "status": "SCHEDULED",
  "reason": null,
  "created_at": "2026-01-01T00:00:00Z",
  "reported_at": null
}

Path parameters

source_id · string · required

The unique identifier of the source connector.

Response

id · string · required

Unique identifier for the connection check.

status · string · required

Connection check status. One of: SCHEDULED, SUCCESS, FAILURE.

reason · string

Failure reason, if applicable.

created_at · string · required

ISO 8601 timestamp when the check was initiated.

reported_at · string

ISO 8601 timestamp when the check result was reported.

cURL

curl --request POST \
  --url "${UNSTRUCTURED_API_URL}/api/v1/sources/a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d/connection-check" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateConnectionCheckSourcesRequest

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL"),
)

response = client.sources.create_source_connection_check(
    request=CreateConnectionCheckSourcesRequest(
        source_id="a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d"
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateConnectionCheckSourcesRequest

async def create_source_connection_check():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.sources.create_source_connection_check_async(
        request=CreateConnectionCheckSourcesRequest(
            source_id="a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d"
        )
    )
    print(response)

asyncio.run(create_source_connection_check())

Response

{
  "id": "c2b3a4d5-6e7f-8a9b-0c1d-2e3f4a5b6c7d",
  "status": "SCHEDULED",
  "reason": null,
  "created_at": "2026-01-01T00:00:00Z",
  "reported_at": null
}