Skip to main content

Python SDK

The official Python client for CTWiseAPI.

Installation

pip install ctwise

Requirements:

  • Python 3.8+
  • requests library (installed automatically)

Quick Start

from ctwise import CTWiseClient

# Initialize with your API key
client = CTWiseClient(api_key="ctw_YOUR_API_KEY")

# Search for regulatory requirements
results = client.requirements.search(
therapeutic_area="Oncology",
authorities=["FDA", "EMA"],
phase="III"
)

# Print results
for req in results.requirements:
print(f"[{req.authority}] {req.title}")
print(f" Type: {req.requirement_type}")
print(f" Reference: {req.regulatory_reference.document}")
print()

Configuration

Basic Configuration

from ctwise import CTWiseClient

client = CTWiseClient(
api_key="ctw_YOUR_API_KEY",
base_url="https://api.ctwise.ai", # Default
timeout=30, # Request timeout in seconds
max_retries=3, # Number of retries on failure
)

Environment Variables

export CTWISE_API_KEY="ctw_YOUR_API_KEY"
from ctwise import CTWiseClient

# API key is read from CTWISE_API_KEY environment variable
client = CTWiseClient()

API Reference

Search for regulatory requirements across multiple authorities.

results = client.requirements.search(
# Search criteria
therapeutic_area="Oncology",
requirement_types=["endpoint_definition", "safety_monitoring"],
authorities=["FDA", "ICH", "EMA"],
phase="III",
keywords=["primary endpoint", "overall survival"],

# Options
include_guidance=True,
include_examples=False,
max_results=100,
)

# Access results
print(f"Found {results.total_count} requirements")

for req in results.requirements:
print(req.requirement_id)
print(req.title)
print(req.description)
print(req.implementation_guidance) # Starter tier only

Amendment Patterns

Get historical amendment patterns for a therapeutic area.

patterns = client.patterns.get(
therapeutic_area="Oncology",
time_period="5_years"
)

for pattern in patterns.patterns:
print(f"Pattern: {pattern.pattern_type}")
print(f"Frequency: {pattern.frequency}%")
print(f"Risk Level: {pattern.risk_level}")

Prevention Strategies

Get prevention strategies to avoid common protocol issues.

strategies = client.prevention.get_strategies(
therapeutic_area="Oncology",
phase="III"
)

for strategy in strategies.strategies:
print(f"Strategy: {strategy.title}")
print(f"Effectiveness: {strategy.effectiveness}%")
for action in strategy.actions:
print(f" - {action}")

Protocol Validation

Validate a protocol against regulatory requirements.

validation = client.rules.validate(
protocol={
"title": "Phase III Oncology Study",
"therapeutic_area": "Oncology",
"phase": "III",
"primary_endpoint": "Overall survival",
"sample_size": 500,
}
)

print(f"Compliance Score: {validation.compliance_score}%")
print(f"Status: {validation.status}")

for issue in validation.issues:
print(f"[{issue.severity}] {issue.description}")
print(f" Recommendation: {issue.recommendation}")

Error Handling

from ctwise import CTWiseClient
from ctwise.exceptions import (
CTWiseError,
AuthenticationError,
RateLimitError,
ValidationError,
NotFoundError,
InsufficientTierError,
)

client = CTWiseClient(api_key="ctw_YOUR_API_KEY")

try:
results = client.requirements.search(therapeutic_area="Oncology")
except AuthenticationError:
print("Invalid API key")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds")
except InsufficientTierError:
print("This feature requires a higher tier")
except ValidationError as e:
print(f"Invalid request: {e.message}")
except CTWiseError as e:
print(f"API error: {e.message}")

Pagination

For large result sets, use pagination:

# Get first page
results = client.requirements.search(
therapeutic_area="Oncology",
max_results=50,
)

# Iterate through all pages
all_requirements = []
for page in client.requirements.search_iter(therapeutic_area="Oncology"):
all_requirements.extend(page.requirements)

print(f"Total requirements: {len(all_requirements)}")

Async Support

For async/await applications:

import asyncio
from ctwise import AsyncCTWiseClient

async def main():
async with AsyncCTWiseClient(api_key="ctw_YOUR_API_KEY") as client:
results = await client.requirements.search(
therapeutic_area="Oncology"
)

for req in results.requirements:
print(req.title)

asyncio.run(main())

Type Hints

The SDK includes full type hints for IDE support:

from ctwise import CTWiseClient
from ctwise.types import RequirementsSearchResponse, Requirement

client = CTWiseClient(api_key="xxx")
results: RequirementsSearchResponse = client.requirements.search(
therapeutic_area="Oncology"
)

req: Requirement = results.requirements[0]
print(req.title) # IDE autocomplete works!

Logging

Enable debug logging:

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("ctwise")
logger.setLevel(logging.DEBUG)

# Now API calls will log request/response details
client = CTWiseClient(api_key="xxx")

Best Practices

1. Reuse Client Instances

# Good - reuse client
client = CTWiseClient(api_key="xxx")
results1 = client.requirements.search(therapeutic_area="Oncology")
results2 = client.requirements.search(therapeutic_area="Psychiatry")

# Bad - create new client each time
results1 = CTWiseClient(api_key="xxx").requirements.search(...)
results2 = CTWiseClient(api_key="xxx").requirements.search(...)

2. Handle Rate Limits

from ctwise.exceptions import RateLimitError
import time

def search_with_backoff(client, **kwargs):
for attempt in range(3):
try:
return client.requirements.search(**kwargs)
except RateLimitError as e:
time.sleep(e.retry_after)
raise Exception("Max retries exceeded")

3. Use Environment Variables for Keys

import os
from ctwise import CTWiseClient

# Never hardcode API keys
client = CTWiseClient(api_key=os.getenv("CTWISE_API_KEY"))

Examples

Integration with Pandas

import pandas as pd
from ctwise import CTWiseClient

client = CTWiseClient()

results = client.requirements.search(
therapeutic_area="Oncology",
max_results=100
)

# Convert to DataFrame
df = pd.DataFrame([
{
"id": r.requirement_id,
"title": r.title,
"authority": r.authority,
"type": r.requirement_type,
"effective_date": r.effective_date,
}
for r in results.requirements
])

print(df.head())

Flask Integration

from flask import Flask, jsonify
from ctwise import CTWiseClient

app = Flask(__name__)
client = CTWiseClient()

@app.route("/search")
def search():
results = client.requirements.search(
therapeutic_area="Oncology"
)
return jsonify({
"count": results.total_count,
"requirements": [r.dict() for r in results.requirements]
})

FDA 483 Inspection Intelligence

CTWiseAPI includes comprehensive FDA 483 inspection data with semantic search, facility risk scoring, and regulatory cross-references.

Search 483 Observations

Search FDA 483 inspection observations using natural language:

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.ctwise.ai/v1"

# Semantic search for 483 observations
response = requests.post(
f"{BASE_URL}/483/observations/search",
headers={
"X-Api-Key": API_KEY,
"Content-Type": "application/json"
},
json={
"query": "data integrity failures in sterile manufacturing",
"top_k": 10,
"filters": {
"category": "Data Integrity",
"program_area": "Human Drugs"
}
}
)

data = response.json()
print(f"Found {data['total']} results\n")

for obs in data["results"]:
print(f"Facility: {obs['legal_name']} (FEI: {obs['fei_number']})")
print(f"CFR: {obs['act_cfr_number']}")
print(f"Description: {obs['short_description'][:100]}...")
print(f"Similarity: {obs['similarity_score']:.2f}")
print(f"Date: {obs['inspection_end_date']}")
print()

Get Facility Profile

Retrieve a facility's inspection history and classification breakdown:

# Get facility profile by FEI number
fei_number = "3016004437"
response = requests.get(
f"{BASE_URL}/483/facilities/{fei_number}",
headers={"X-Api-Key": API_KEY}
)

facility = response.json()
print(f"Facility: {facility['legal_name']}")
print(f"Location: {facility['city']}, {facility['state']} {facility['zip_code']}")
print(f"Total Inspections: {facility['total_inspections']}")
print(f"\nClassifications:")
print(f" OAI (Official Action Indicated): {facility['classification_breakdown']['oai_count']}")
print(f" VAI (Voluntary Action Indicated): {facility['classification_breakdown']['vai_count']}")
print(f" NAI (No Action Indicated): {facility['classification_breakdown']['nai_count']}")
print(f"\nRisk Score: {facility['risk_assessment']['risk_score']:.1f}/100")
print(f"Risk Level: {facility['risk_assessment']['risk_level']}")

Get Facility Citations

Retrieve all citations for a specific facility:

# Get citations for a facility
response = requests.get(
f"{BASE_URL}/483/facilities/{fei_number}/citations",
headers={"X-Api-Key": API_KEY},
params={"limit": 20}
)

citations = response.json()
print(f"Total Citations: {citations['total']}\n")

for citation in citations["results"]:
print(f"Date: {citation['inspection_end_date']}")
print(f"CFR: {citation['act_cfr_number']}")
print(f"Description: {citation['short_description'][:80]}...")
print(f"Category: {citation['category']}")
print()

Calculate Risk Scores

Get a comprehensive risk assessment with 5-factor breakdown:

# Get risk score for a facility
response = requests.get(
f"{BASE_URL}/483/risk-scores/{fei_number}",
headers={"X-Api-Key": API_KEY}
)

risk = response.json()
print(f"Composite Risk Score: {risk['risk_score']:.1f}/100")
print(f"Risk Level: {risk['risk_level']}")
print(f"Methodology: {risk['methodology_version']}")
print(f"\nFactor Breakdown:")
print(f" OAI Ratio: {risk['factors']['oai_ratio']:.1f}")
print(f" Citation Frequency: {risk['factors']['citation_frequency']:.1f}")
print(f" Recency: {risk['factors']['recency']:.1f}")
print(f" Severity: {risk['factors']['severity']:.1f}")
print(f" Peer Benchmark: {risk['factors']['peer_benchmark']:.1f}")

Browse CFR References

See which CFR sections are most frequently cited:

# Get most cited CFR sections
response = requests.get(
f"{BASE_URL}/483/cfr-references",
headers={"X-Api-Key": API_KEY},
params={
"sort_order": "desc",
"limit": 10
}
)

cfr_data = response.json()
print(f"Top {len(cfr_data['results'])} most cited CFR sections:\n")

for cfr in cfr_data["results"]:
print(f"{cfr['act_cfr_number']}: {cfr['occurrence_count']} citations")
print(f" {cfr['short_description']}")
print()

# Get details for a specific CFR section
cfr_section = "21 CFR 211.68"
response = requests.get(
f"{BASE_URL}/483/cfr-references/{cfr_section}",
headers={"X-Api-Key": API_KEY},
params={"cfr": cfr_section}
)

cfr_detail = response.json()
print(f"\nCFR: {cfr_detail['act_cfr_number']}")
print(f"Title: {cfr_detail['short_description']}")
print(f"Total Citations: {cfr_detail['occurrence_count']}")

Get Analytics Summary

Retrieve aggregate statistics across all 483 data:

# Get analytics summary
response = requests.get(
f"{BASE_URL}/483/analytics/summary",
headers={"X-Api-Key": API_KEY}
)

summary = response.json()
print(f"Total Citations: {summary['total_citations']:,}")
print(f"Total Facilities: {summary['total_facilities']:,}")
print(f"Total CFR References: {summary['total_cfr_references']:,}")
print(f"\nRisk Distribution:")
for level, count in summary["risk_distribution"].items():
print(f" {level.capitalize()}: {count}")

print(f"\nTop 5 Most Cited CFRs:")
for cfr in summary["top_cited_cfr"][:5]:
print(f" {cfr['act_cfr_number']}: {cfr['occurrence_count']} citations")

Find CFR sections with increasing citation frequency:

# Get trending citations
response = requests.get(
f"{BASE_URL}/483/citations/trending",
headers={"X-Api-Key": API_KEY},
params={
"days": 365,
"limit": 10
}
)

trending = response.json()
print(f"Trending citations in the last {trending['days']} days:\n")

for item in trending["results"]:
print(f"{item['act_cfr_number']}")
print(f" Recent: {item['recent_count']} citations")
print(f" Previous: {item['previous_count']} citations")
print(f" Trend: {item['trend_direction']} ({item['trend_percentage']:+.1f}%)")
print()

Regulatory Mapping (ICH E6 R3)

Cross-reference CFR citations with ICH E6(R3) requirements:

# Get regulatory mapping for a CFR section
response = requests.get(
f"{BASE_URL}/483/regulatory-mapping/21 CFR 211.68",
headers={"X-Api-Key": API_KEY}
)

mapping = response.json()
print(f"CFR: {mapping['cfr_section']}")
print(f"Title: {mapping['cfr_title']}")
print(f"483 Citations: {mapping['occurrence_count']}")
print(f"\nICH E6(R3) Mappings:")

for ich in mapping["ich_mappings"]:
print(f"\n {ich['ich_section']}")
print(f" Requirement: {ich['matched_requirement']}")
print(f" Relevance: {ich['relevance_score']:.2f}")
print(f" Rationale: {ich['mapping_rationale']}")

Get Analytics Benchmarks

Compare facility risk scores against peer groups:

# Get benchmarks by product type
response = requests.get(
f"{BASE_URL}/483/analytics/benchmarks",
headers={"X-Api-Key": API_KEY},
params={"product_type": "drug"}
)

benchmarks = response.json()
print(f"Product Type: {benchmarks['product_type']}")
print(f"Facility Count: {benchmarks['facility_count']}")
print(f"\nRisk Score Statistics:")
print(f" Mean: {benchmarks['risk_score_stats']['mean']:.1f}")
print(f" Median: {benchmarks['risk_score_stats']['median']:.1f}")
print(f" P75: {benchmarks['risk_score_stats']['p75']:.1f}")
print(f" P90: {benchmarks['risk_score_stats']['p90']:.1f}")

REST API Reference

If you prefer to call the REST API directly without the SDK, here are the available endpoints:

Public Endpoints (No Authentication Required)

EndpointDescription
GET /v1/catalog/sourcesList all available regulatory sources

Protected Endpoints (API Key Required)

Regulatory Rules

EndpointDescription
GET /v1/rulesSearch rules by keyword, source, category
GET /v1/rules/{id}Get specific rule details
POST /v1/semantic-searchAI-powered natural language search
GET /v1/rules/searchAlternative GET-based semantic search

FDA 483 Intelligence

EndpointDescription
POST /v1/483/observations/searchSemantic search for 483 observations
GET /v1/483/facilities/{fei}Get facility profile
GET /v1/483/facilities/{fei}/citationsGet facility citations
GET /v1/483/risk-scores/{fei}Get facility risk score
GET /v1/483/cfr-referencesList CFR references
GET /v1/483/cfr-references/{cfr}Get CFR details
GET /v1/483/analytics/summaryGet analytics summary
GET /v1/483/citations/trendingGet trending citations
GET /v1/483/regulatory-mappingList regulatory mappings
GET /v1/483/regulatory-mapping/{cfr}Get CFR to ICH mapping
GET /v1/483/analytics/benchmarksGet peer benchmarks

Semantic Search Endpoints

The SDK's search() method uses these endpoints under the hood:

# SDK method
results = client.rules.search(query="informed consent requirements")

# Equivalent REST API call (POST method - recommended)
import requests
response = requests.post(
"https://api.ctwise.ai/v1/semantic-search",
headers={"X-Api-Key": "YOUR_API_KEY"},
json={"query": "informed consent requirements", "limit": 10}
)

# Equivalent REST API call (GET method - alternative)
response = requests.get(
"https://api.ctwise.ai/v1/rules/search",
headers={"X-Api-Key": "YOUR_API_KEY"},
params={"q": "informed consent requirements", "limit": 10}
)

Troubleshooting

Common Errors

Error CodeHTTP StatusCauseSolution
INVALID_API_KEY401Invalid or expired API keyVerify your API key is correct and active
MISSING_API_KEY401No API key providedInclude X-Api-Key header in requests
RATE_LIMIT_EXCEEDED429Too many requestsWait and retry with exponential backoff
INSUFFICIENT_TIER403Feature requires higher tierUpgrade your subscription
INVALID_REQUEST400Malformed request bodyCheck request parameters
NOT_FOUND404Resource doesn't existVerify the rule ID or endpoint path

Debugging Tips

  1. Enable verbose logging to see request/response details:

    import logging
    logging.basicConfig(level=logging.DEBUG)
  2. Test connectivity with the catalog endpoint (no auth required):

    response = requests.get("https://api.ctwise.ai/v1/catalog/sources")
    print(response.status_code) # Should be 200
  3. Verify API key format: Keys should start with ctw_ prefix

  4. Check tier access: Some sources (ICH, EMA, WHO) require Starter tier or higher

Support