Testing Connectors¶

Testing ensures your connector descriptor is valid, credentials work, and operations return expected data. Manifest Platform provides testing at every layer: descriptor validation, connection verification, operation execution, and ongoing health monitoring.

Testing Workflow¶

graph TD
    V["1. Validate Descriptor"] --> T["2. Test Connection"]
    T --> O["3. Test Operations"]
    O --> SD["Schema Discovery<br/>(optional)"]
    O --> DDL["DDL Execution<br/>(database connectors)"]
    O --> H["4. Monitor Health"]
    H --> HC["On-Demand Health Check"]
    H --> BHC["Bulk Health Check"]

1. Descriptor Validation¶

Before registering a connector, validate the descriptor to catch structural issues.

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

result = client.connectors.validate(descriptor={
    "metadata": {
        "name": "My API",
        "slug": "my-api",
        "version": "1.0.0",
        "description": "Integration with My API service",
        "publisher": "Team"
    },
    "authentication": [...],
    "operations": [...],
    "capabilities": {"sync_modes": ["full_refresh"]},
    "sla": {"availability_percentage": 99.0}
})

if result.valid:
    print("Descriptor is valid")
    # result.normalized_descriptor contains the cleaned-up version
else:
    for error in result.errors:
        print(f"Error: {error}")

What Validation Checks¶

Check	Description
Required fields	All mandatory fields present with correct types
Auth profile uniqueness	No duplicate authentication type entries
Operation ID uniqueness	Every operation has a unique `operation_id`
Auth references	Each operation's `auth_type_ref` points to a defined auth profile
MCP mappings	MCP route mappings reference existing operations
MCP tool name uniqueness	No duplicate tool names in MCP configuration
Field constraints	String lengths, regex patterns, numeric ranges

2. Connection Testing¶

After creating an instance and storing credentials, test the connection:

Coming Soon

The Python SDK for local development is not yet publicly available.

result = client.connector_instances.test_connection(
    instance_id="uuid-of-instance",
    scope_type="organization",
    scope_id="uuid-of-org",
)

print(f"Status: {result.status}")
print(f"Message: {result.message}")
print(f"Details: {result.details}")
print(f"Tested at: {result.tested_at}")

How Connection Testing Works¶

The test method depends on the authentication type:

Auth Type	Test Method
`none`	No credentials required; skips credential validation
`api_key`	Calls the `test_endpoint` URL with credentials injected into headers
`basic`	Calls the `test_endpoint` URL with HTTP Basic auth header
`service_account`	Calls the `test_endpoint` URL with service account credentials
`oauth2_client_credentials`	Performs token exchange and verifies success
`oauth2_authorization_code`	Verifies stored tokens are valid (refreshes if needed)
Credential chain	Executes all chain steps in order and verifies the final credential is obtained

Test Result Fields¶

Field	Description
`status`	`success`, `failed`, or `pending`
`message`	Human-readable summary of the result
`details`	Additional diagnostic detail (e.g., raw error from the external API)
`tested_at`	ISO 8601 timestamp of when the test was run

Status	Meaning
`success`	Credentials are valid, external system is reachable
`failed`	Authentication failed, system unreachable, or unexpected response
`pending`	Test is in progress (async)

Test after every credential change

Always run a connection test after storing or rotating credentials. A common issue is copying credentials with trailing whitespace or newline characters.

3. Testing Operations¶

Test individual connector operations to verify they return expected data.

Dataset Source Test¶

If the connector instance is linked as a dataset source, use the source test endpoint:

Coming Soon

The Python SDK for local development is not yet publicly available.

result = client.datasets.test_source(
    dataset_id="uuid-of-dataset",
    source_id="uuid-of-source",
    parameters={"status": "open"},
    limit=5,
)

if result.success:
    print(f"Returned {result.record_count} rows in {result.execution_time_ms}ms")
    print(f"Inferred schema: {result.response_schema}")
    for row in result.sample_rows:
        print(row)
else:
    print(f"Error: {result.error}")
    print(f"Status code: {result.status_code}")

The test response includes:

Field	Description
`success`	Whether the operation completed without errors
`status_code`	HTTP status code from the external API
`response_schema`	Schema inferred from the response data
`sample_rows`	Up to `limit` rows of data
`record_count`	Total rows returned
`error`	Error message if the operation failed
`execution_time_ms`	How long the operation took

Operation Input Validation¶

The platform validates operation parameters before execution:

# The platform automatically validates:
# - Required parameters are present
# - Parameter types match the declared field_type
# - Extra parameters are silently ignored

Validation issues return structured errors:

{
  "valid": false,
  "issues": [
    {"path": "jql", "expected": "required", "actual": "missing"},
    {"path": "maxResults", "expected": "integer", "actual": "string"}
  ]
}

Operation Output Validation¶

If an operation defines a response_schema, the platform validates the external API's response:

Top-level type check (object, array, string, etc.)
Required property check for object responses
Property-level type checks

Output validation failures are logged as warnings but do not block the response. This prevents brittle behavior when external APIs add new fields.

4. Health Monitoring¶

On-Demand Health Checks¶

Run a lightweight health check on any instance:

Coming Soon

The Python SDK for local development is not yet publicly available.

health = client.connector_instances.health_check(
    instance_id="uuid-of-instance",
    scope_type="organization",
    scope_id="uuid-of-org",
    timeout_seconds=10.0,
)

print(f"Status: {health.status}")       # healthy or unhealthy
print(f"Latency: {health.latency_ms}ms")
if health.error:
    print(f"Error: {health.error}")

Health Check Result Fields¶

Field	Description
`status`	`healthy` or `unhealthy`
`latency_ms`	Round-trip time to the external system in milliseconds
`error`	Error message if the check failed, otherwise `null`

Bulk Health Checks¶

Check multiple instances at once:

Coming Soon

The Python SDK for local development is not yet publicly available.

results = client.connector_instances.bulk_operation(
    instance_ids=["uuid-1", "uuid-2", "uuid-3"],
    action="health_check",
)

for r in results:
    status = "OK" if r.status == "success" else "FAIL"
    print(f"  {r.instance_id}: {status}")

Expiring Credentials¶

Monitor credentials approaching expiration:

Coming Soon

The Python SDK for local development is not yet publicly available.

expiring = client.connector_instances.list_expiring_credentials(
    days_ahead=14,
)

for cred in expiring:
    if cred.days_remaining < 0:
        print(f"EXPIRED: {cred.connector_name} (expired {abs(cred.days_remaining):.0f} days ago)")
    else:
        print(f"EXPIRING: {cred.connector_name} ({cred.days_remaining:.0f} days remaining)")

Debugging Common Issues¶

Authentication Failures¶

Symptom	Likely Cause	Fix
`401 Unauthorized`	Invalid or expired credentials	Re-store credentials and test connection
`403 Forbidden`	Credentials lack required permissions	Check API scopes/roles on the external system
Connection timeout	Network or firewall issue	Verify the external system is reachable from the platform
`test_endpoint` returns unexpected status	Wrong test URL or changed API	Update the `test_endpoint` in the connector descriptor

Operation Failures¶

Symptom	Likely Cause	Fix
Missing required parameter	Parameter not provided in request	Check operation parameter definitions
Type mismatch	String sent where integer expected	Verify `field_type` in parameter spec matches actual API
Empty response	Operation returns no data	Verify the query/filter parameters are correct
Rate limit exceeded	Too many requests in interval	Check `rate_limits` in descriptor and reduce request frequency

MCP Tool Issues¶

Symptom	Likely Cause	Fix
Tool not appearing for agent	MCP not enabled on connector	Set `mcp_config.enabled: true` in descriptor
Tool call fails silently	Credential scope mismatch	Ensure agent has credentials at the correct scope
Stale tool behavior	Cached MCP server	Update the connector descriptor to invalidate cache

Enable logging for debugging

Connector operations are logged in the platform's observability system. Check the execution traces for detailed request/response data, including headers, status codes, and timing.