Python Examples¶

import os

API_KEY = os.environ.get("PROSPECTIVE_API_KEY", "sk_live_your_key_here")
BASE_URL = "https://api.bedrockbilling.com/api/v1"

HEADERS = {
    "Authorization": API_KEY,
    "Content-Type": "application/json"
}

import requests

def search_fee_schedules(code: str, fee_type: str = "OPPS"):
    """Search for fee schedule rates by HCPCS code."""
    response = requests.get(
        f"{BASE_URL}/fee-schedules/search",
        headers=HEADERS,
        params={
            "code": code,
            "fee_type": fee_type,
            "limit": 10
        }
    )
    response.raise_for_status()
    return response.json()

# Example usage
results = search_fee_schedules("99213")
for item in results["data"]:
    print(f"{item['code']}: ${item['rate']}")

def get_fee_schedule_by_code(code: str, fee_type: str, effective_date: str):
    """Get specific fee schedule for a code."""
    response = requests.get(
        f"{BASE_URL}/fee-schedules/code/{code}",
        headers=HEADERS,
        params={
            "fee_type": fee_type,
            "effective_date": effective_date
        }
    )
    response.raise_for_status()
    return response.json()

# Example
rate = get_fee_schedule_by_code("99213", "OPPS", "2025-01-01")
print(f"Rate: ${rate['rate']}")

def process_claim(claim_data: dict):
    """Process a claim through the pricing engine."""
    response = requests.post(
        f"{BASE_URL}/claims/process",
        headers=HEADERS,
        json=claim_data
    )
    response.raise_for_status()
    return response.json()

# Example OPPS claim
claim = {
    "bill_type": "131",
    "from_date": "2025-01-15",
    "thru_date": "2025-01-15",
    "patient_status": "01",
    "billing_provider": {
        "npi": "1234567890"
    },
    "line_items": [
        {
            "hcpcs": "99213",
            "rev_code": "0510",
            "units": 1,
            "charges": 150.00
        }
    ],
    "diagnosis_codes": ["Z00.00"]
}

result = process_claim(claim)
print(f"Total Payment: ${result['total_payment']}")

def get_mue(code: str, mue_type: str = "Practitioner"):
    """Get MUE limit for a code."""
    response = requests.get(
        f"{BASE_URL}/mue-data/code/{code}",
        headers=HEADERS,
        params={"mue_type": mue_type}
    )
    response.raise_for_status()
    return response.json()

# Example
mue = get_mue("99213", "Practitioner")
print(f"MUE Value: {mue['mue_value']} units")
print(f"MAI: {mue['mue_adjudication_indicator']}")

import time

def make_request_with_retry(url: str, params: dict = None, max_retries: int = 3):
    """Make a request with automatic retry on rate limit."""
    for attempt in range(max_retries):
        response = requests.get(url, headers=HEADERS, params=params)

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            print(f"Rate limited. Waiting {retry_after} seconds...")
            time.sleep(retry_after)
            continue

        response.raise_for_status()
        return response.json()

    raise Exception("Max retries exceeded")

def get_all_results(endpoint: str, params: dict = None):
    """Fetch all pages of a paginated endpoint."""
    params = params or {}
    params["skip"] = 0
    params["limit"] = 100

    all_results = []

    while True:
        response = requests.get(
            f"{BASE_URL}/{endpoint}",
            headers=HEADERS,
            params=params
        )
        response.raise_for_status()
        data = response.json()

        all_results.extend(data["data"])

        if len(data["data"]) < params["limit"]:
            break

        params["skip"] += params["limit"]

    return all_results

# Example: Get all fee schedules for a code prefix
all_opps = get_all_results("fee-schedules/search", {"code": "992", "fee_type": "OPPS"})
print(f"Found {len(all_opps)} records")

class ProspectiveAPIError(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail
        super().__init__(f"API Error {status_code}: {detail}")

def api_request(method: str, endpoint: str, **kwargs):
    """Make an API request with proper error handling."""
    url = f"{BASE_URL}/{endpoint}"

    response = requests.request(method, url, headers=HEADERS, **kwargs)

    if response.status_code == 401:
        raise ProspectiveAPIError(401, "Invalid API key")
    elif response.status_code == 404:
        raise ProspectiveAPIError(404, "Resource not found")
    elif response.status_code == 429:
        retry_after = response.headers.get("Retry-After", "60")
        raise ProspectiveAPIError(429, f"Rate limited. Retry after {retry_after}s")
    elif response.status_code >= 400:
        detail = response.json().get("detail", "Unknown error")
        raise ProspectiveAPIError(response.status_code, detail)

    return response.json()

# Usage
try:
    data = api_request("GET", "fee-schedules/code/99213", params={"fee_type": "OPPS"})
except ProspectiveAPIError as e:
    print(f"Error: {e}")