Python SDK
The official Python client for CTWiseAPI.
Installation
pip install ctwise
Requirements:
- Python 3.8+
requestslibrary (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
Requirements Search
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")
Get Trending 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)
| Endpoint | Description |
|---|---|
GET /v1/catalog/sources | List all available regulatory sources |
Protected Endpoints (API Key Required)
Regulatory Rules
| Endpoint | Description |
|---|---|
GET /v1/rules | Search rules by keyword, source, category |
GET /v1/rules/{id} | Get specific rule details |
POST /v1/semantic-search | AI-powered natural language search |
GET /v1/rules/search | Alternative GET-based semantic search |
FDA 483 Intelligence
| Endpoint | Description |
|---|---|
POST /v1/483/observations/search | Semantic search for 483 observations |
GET /v1/483/facilities/{fei} | Get facility profile |
GET /v1/483/facilities/{fei}/citations | Get facility citations |
GET /v1/483/risk-scores/{fei} | Get facility risk score |
GET /v1/483/cfr-references | List CFR references |
GET /v1/483/cfr-references/{cfr} | Get CFR details |
GET /v1/483/analytics/summary | Get analytics summary |
GET /v1/483/citations/trending | Get trending citations |
GET /v1/483/regulatory-mapping | List regulatory mappings |
GET /v1/483/regulatory-mapping/{cfr} | Get CFR to ICH mapping |
GET /v1/483/analytics/benchmarks | Get 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 Code | HTTP Status | Cause | Solution |
|---|---|---|---|
INVALID_API_KEY | 401 | Invalid or expired API key | Verify your API key is correct and active |
MISSING_API_KEY | 401 | No API key provided | Include X-Api-Key header in requests |
RATE_LIMIT_EXCEEDED | 429 | Too many requests | Wait and retry with exponential backoff |
INSUFFICIENT_TIER | 403 | Feature requires higher tier | Upgrade your subscription |
INVALID_REQUEST | 400 | Malformed request body | Check request parameters |
NOT_FOUND | 404 | Resource doesn't exist | Verify the rule ID or endpoint path |
Debugging Tips
-
Enable verbose logging to see request/response details:
import logging
logging.basicConfig(level=logging.DEBUG) -
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 -
Verify API key format: Keys should start with
ctw_prefix -
Check tier access: Some sources (ICH, EMA, WHO) require Starter tier or higher
Support
- PyPI: pypi.org/project/ctwise
- GitHub: github.com/ctwise/python-sdk
- Issues: github.com/ctwise/python-sdk/issues