Explore/ NPI registry - Providers Search
Public workflow
NPI registry - Providers Search
By Fullstacktics · used 0 times · updated 2026-10-19
How it runs
Distilled · v2 Includes a distilled script, so your copy runs it from its first run. If a page has changed, it falls back to the AI agent for that run.
Instructions
Open npiregistry.cms.hhs.gov Search for providers named $last_name in state $state. Return each match: npi, name, credential, primary_taxonomy, practice_address, phone.
You provide “last_name”, “state” each time you run it.
Review the distilled script
This Python runs in your browser session when you use the workflow. Review it before running.
from playwright.sync_api import expectimport re def run(page, context, browser): def visible(locator, scope): try: expect(locator).to_be_visible(timeout=10000) except AssertionError: scope.deopt( reason="The next operation's target did not become visible.", evidence={"locator": str(locator)}, ) def exact_count(locator, count, scope): try: expect(locator).to_have_count(count, timeout=10000) except AssertionError: scope.deopt( reason="The target did not resolve with the required cardinality.", evidence={"expected": count, "actual": locator.count()}, ) def query_rows(result, scope): if ( not isinstance(result, dict) or result.get("truncated") is not False or not isinstance(result.get("rows"), list) ): scope.deopt( reason="The store query was not an untruncated row result.", evidence={"query_result": result}, ) return result["rows"] def checked_batch(batch, scope): if not isinstance(batch, list): scope.deopt( reason="The registry search did not return an array.", evidence={"response": batch}, ) for record in batch: if ( not isinstance(record, dict) or not isinstance(record.get("number"), str) or not record["number"].strip() ): scope.deopt( reason="A registry result lacks a usable NPI.", evidence={"record": record}, ) return batch def current_heading(scope): locator = page.get_by_role("heading").filter(has_text="Results:") visible(locator, scope) exact_count(locator, 1, scope) text = locator.inner_text() match = re.search(r"Results:\s*([\d,]+)\s*-\s*([\d,]+)", text) if not match: scope.deopt( reason="The live results heading has no usable pagination range.", evidence={"heading": text}, ) return { "text": text, "start": int(match.group(1).replace(",", "")), "end": int(match.group(2).replace(",", "")), } object_schema = {"type": "object"} with browser.scope( "open-registry", goal="Open the NPI Registry search page.", returns=None, ) as opened: page.goto("https://npiregistry.cms.hhs.gov", wait_until="domcontentloaded") page.wait_for_load_state("domcontentloaded") opened.value = None with browser.scope( "configure-provider-search", goal="Resolve the live state option, fill the provider surname, enable exact matching, and submit once.", returns={ "type": "object", "properties": { "last_name": {"type": "string"}, "state": {"type": "string"}, "state_value": {"type": "string"}, "initial_count": {"type": "integer"}, "empty_evidence": {"type": ["string", "null"]}, }, "required": [ "last_name", "state", "state_value", "initial_count", "empty_evidence", ], }, ) as configured: last_name = browser.input["promptInputs"]["last_name"] state = browser.input["promptInputs"]["state"] if not isinstance(last_name, str) or not isinstance(state, str): configured.deopt( reason="Search inputs must be strings.", evidence={"last_name": last_name, "state": state}, ) surname = page.locator("#lastName") visible(surname, configured) expect(surname).to_be_editable(timeout=10000) surname.fill(last_name) state_control = page.locator("#state") visible(state_control, configured) options = page.locator("#state option").evaluate_all( "(els)=>els.map(e=>({label:e.textContent.trim(),value:e.value}))" ) matches = [ option for option in options if option["label"].casefold() == state.strip().casefold() ] if len(matches) != 1 or not matches[0]["value"]: configured.deopt( reason="The requested state does not uniquely match a live state option.", evidence={"requested_state": state, "options": options}, ) state_value = matches[0]["value"] expect(state_control).to_be_enabled(timeout=10000) state_control.select_option(state_value) expect(state_control).to_have_value(state_value, timeout=10000) exact = page.locator("#exactMatch") visible(exact, configured) expect(exact).to_be_enabled(timeout=10000) exact.check() expect(exact).to_be_checked(timeout=10000) search = page.get_by_role("button", name="Search", exact=True) visible(search, configured) expect(search).to_be_enabled(timeout=10000) with page.expect_response( lambda response: "/RegistryBack/search" in response.url ) as submitted: search.click() response = submitted.value if not response.ok: configured.deopt( reason="The submitted registry search failed.", evidence={"status": response.status, "url": response.url}, ) initial = checked_batch(response.json(), configured) empty_evidence = None if initial: current_heading(configured) else: expect(surname).to_be_visible(timeout=10000) empty_evidence = page.locator("body").inner_text() configured.value = { "last_name": last_name, "state": state, "state_value": state_value, "initial_count": len(initial), "empty_evidence": empty_evidence, } with browser.scope( "collect-registry-results", goal="Collect fresh search pages, deduplicate by NPI, and respect the registry result ceiling.", returns={ "type": "object", "properties": { "records": {"type": "array", "items": {"type": "object"}}, "pages": {"type": "array", "items": {"type": "object"}}, "ui_page": {"type": "integer"}, "truncated": {"type": "boolean"}, "limit": {"type": "integer"}, }, "required": ["records", "pages", "ui_page", "truncated", "limit"], }, ) as collected: payload = { "firstName": None, "lastName": configured.value["last_name"], "organizationName": None, "aoFirstName": None, "aoLastName": None, "skip": 0, "enumerationType": None, "number": None, "city": None, "state": configured.value["state_value"], "country": None, "taxonomyDescription": None, "postalCode": None, "exactMatch": True, "addressType": None, } batch = checked_batch( page.evaluate( """async (payload)=>{ const r=await fetch('/RegistryBack/search',{ method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(payload) }); if(!r.ok)throw new Error(r.status); return await r.json() }""", payload, ), collected, ) if bool(batch) != bool(configured.value["initial_count"]): collected.deopt( reason="The submitted search and fresh collection disagree about whether matches exist.", evidence={ "submitted_count": configured.value["initial_count"], "fresh_count": len(batch), }, ) def registry_batch(offset): request_payload = dict(payload, skip=offset) result = page.evaluate( """async (payload) => { const response = await fetch('/RegistryBack/search', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(payload)}); if (!response.ok) throw new Error('Registry HTTP '+response.status); return await response.json(); }""", request_payload, ) return checked_batch(result, collected) limit = 2100 pages = [] records = {} skip = 0 ui_page = 0 truncated = False more_clicked = False while True: if len(batch) > 101: collected.deopt( reason="The registry returned an unexpected page size.", evidence={"skip": skip, "count": len(batch)}, ) pages.append({"skip": skip, "response_count": len(batch)}) for record in batch[:min(100, limit - skip)]: records[record["number"]] = record if len(batch) <= 100: break if skip + 100 >= limit: truncated = True break next_skip = skip + 100 if skip == 0: before = current_heading(collected) if before["start"] != skip + 1: collected.deopt( reason="The live UI is not on the initial collected page before More.", evidence={"heading": before, "skip": skip}, ) more = page.get_by_text("More", exact=True) visible(more, collected) expect(more).to_be_enabled(timeout=10000) with page.expect_response( lambda response: "/RegistryBack/search" in response.url ) as next_response: more.click() response = next_response.value if not response.ok: collected.deopt( reason="The registry's next-page response failed.", evidence={"status": response.status}, ) batch = checked_batch(response.json(), collected) more_clicked = True else: batch = registry_batch(next_skip) skip = next_skip if records: live_heading = current_heading(collected) live_skip = live_heading["start"] - 1 live_pages = [info for info in pages if info["skip"] == live_skip] if ( live_skip % 100 != 0 or len(live_pages) != 1 or live_skip > (100 if more_clicked else 0) ): collected.deopt( reason="The actual UI position does not correspond to a collected page reachable by the acknowledged clicks.", evidence={"heading": live_heading, "pages": pages, "more_clicked": more_clicked}, ) expected_end = live_skip + min(live_pages[0]["response_count"], 100) if live_heading["end"] != expected_end: collected.deopt( reason="The actual UI range does not match its collected response page.", evidence={"heading": live_heading, "page": live_pages[0]}, ) ui_page = live_skip // 100 collected.value = { "records": list(records.values()), "pages": pages, "ui_page": ui_page, "truncated": truncated, "limit": limit, } with browser.scope( "format-requested-fields", goal="Normalize every fresh registry match into the six requested fields.", returns={ "type": "array", "items": { "type": "object", "required": [ "npi", "name", "credential", "primary_taxonomy", "practice_address", "phone", ], "properties": { "npi": {"type": "string"}, "name": {"type": "string"}, "credential": {"type": ["string", "null"]}, "primary_taxonomy": {"type": ["string", "null"]}, "practice_address": {"type": ["string", "null"]}, "phone": {"type": ["string", "null"]}, }, }, }, ) as formatted: results = [] for record in collected.value["records"]: address_data = record.get("primaryAddress") or {} basic = record.get("basic") or {} taxonomy_data = record.get("primaryTaxonomy") or {} if not all( isinstance(value, dict) for value in (address_data, basic, taxonomy_data) ): formatted.deopt( reason="A provider's requested source fields have an unexpected shape.", evidence={"record": record}, ) postal = address_data.get("postalCode") or "" postal = postal[:5] + "-" + postal[5:] if len(postal) == 9 else postal address = ", ".join( value for value in [ address_data.get("addressLine1"), address_data.get("addressLine2"), ", ".join( value for value in [ address_data.get("city"), " ".join( value for value in [ address_data.get("state"), postal ] if value ), ] if value ), address_data.get("countryCode") if address_data.get("countryCode") != "US" else None, ] if value ) if record.get("enumerationType") == "NPI-1": name = " ".join( basic[key].strip() for key in ["firstName", "middleName", "lastName", "nameSuffix"] if basic.get(key) and basic[key] != "--" ) else: name = basic.get("orgName", basic.get("name", "")) if not isinstance(name, str) or not name.strip(): formatted.deopt( reason="A registry match has no usable provider name.", evidence={"record": record}, ) specialization = (taxonomy_data.get("specialization") or "").strip() taxonomy = (taxonomy_data.get("desc") or "") + ( " — " + specialization if specialization else "" ) results.append({ "npi": record["number"], "name": name, "credential": basic.get("credential") or None, "primary_taxonomy": taxonomy or None, "practice_address": address or None, "phone": address_data.get("teleNumber") or None, }) formatted.value = results with browser.scope( "save-workflow-results", goal="Save current normalized matches idempotently and account for each destination record.", returns={ "type": "object", "properties": { "outcomes": {"type": "array", "items": {"type": "object"}}, "table_count": {"type": "integer"}, }, "required": ["outcomes", "table_count"], }, ) as saved: browser.store.define_table( table="npi_matches", columns={ "npi": {"type": "text"}, "name": {"type": "text"}, "credential": {"type": "text", "nullable": True}, "primary_taxonomy": {"type": "text", "nullable": True}, "practice_address": {"type": "text", "nullable": True}, "phone": {"type": "text", "nullable": True}, }, primaryKey=["npi"], ) outcomes = [] for row in formatted.value: escaped_npi = row["npi"].replace("'", "''") existing = query_rows( browser.store.query( sql=( "select npi, name, credential, primary_taxonomy, " "practice_address, phone from npi_matches where npi = '" + escaped_npi + "'" ) ), saved, ) if len(existing) > 1: saved.deopt( reason="The destination NPI identity is not unique.", evidence={"npi": row["npi"], "rows": existing}, ) if existing: stored = existing[0] if not all(key in stored for key in row) or any( stored[key] != value for key, value in row.items() ): saved.deopt( reason="An existing destination record differs from the current extraction.", evidence={"current": row, "stored": stored}, ) reason = f"Verified existing saved registry match {row['npi']}" else: receipt = browser.store.insert( table="npi_matches", row=row, idempotencyKey="npi:" + row["npi"], ) if ( not isinstance(receipt, dict) or receipt.get("committed") is not True or receipt.get("key", {}).get("npi") != row["npi"] ): saved.deopt( reason="The store did not acknowledge the requested NPI save.", evidence={"npi": row["npi"], "receipt": receipt}, ) reason = f"Saved registry match {row['npi']} with requested fields to workflow store" browser.record.outcome( collection="npi_matches", recordKey=row["npi"], status="saved", reason=reason[:1000], ) outcomes.append({ "collection": "npi_matches", "recordKey": row["npi"], "status": "saved", }) count_rows = query_rows( browser.store.query(sql="select count(*) as count from npi_matches"), saved, ) if len(count_rows) != 1 or "count" not in count_rows[0]: saved.deopt( reason="The destination count query lacks its required column.", evidence={"rows": count_rows}, ) table_count = int(count_rows[0]["count"]) if table_count < len(outcomes): saved.deopt( reason="The destination contains fewer records than acknowledged saves.", evidence={"table_count": table_count, "saved_count": len(outcomes)}, ) saved.value = {"outcomes": outcomes, "table_count": table_count} with browser.scope( "verify-and-report", goal="Verify live pagination exhaustion and destination accounting, then report current matches.", returns=object_schema, ) as verified: final_heading = None more_count = 0 if collected.value["records"]: current_heading(verified) for page_info in collected.value["pages"][ collected.value["ui_page"] + 1: ]: more = page.get_by_role("button", name="More", exact=True) visible(more, verified) expect(more).to_be_enabled(timeout=10000) with page.expect_response( lambda response: "/RegistryBack/search" in response.url ) as advancing: more.click() response = advancing.value if not response.ok: verified.deopt( reason="The live pagination verification request failed.", evidence={"status": response.status}, ) response_batch = checked_batch(response.json(), verified) if len(response_batch) != page_info["response_count"]: verified.deopt( reason="Live pagination differs from the collected page.", evidence={ "page": page_info, "live_count": len(response_batch), }, ) expected_range = ( f"{page_info['skip'] + 1} - " f"{page_info['skip'] + min(len(response_batch), 100)}" ) page.get_by_role("heading").filter(has_text="Results:").wait_for( state="visible", timeout=10000 ) page.wait_for_function( '(s)=>document.body.innerText.includes("Results: "+s)', arg=expected_range, timeout=10000, ) final_heading = current_heading(verified) more = page.get_by_role("button", name="More", exact=True) exact_count(more, 0, verified) more_count = more.count() last_page = collected.value["pages"][-1] expected_end = last_page["skip"] + min( last_page["response_count"], 100 ) if final_heading["end"] != expected_end: verified.deopt( reason="The final displayed range does not match collection exhaustion.", evidence={ "heading": final_heading, "last_page": last_page, }, ) verification_rows = query_rows( browser.store.query( sql="select count(*) as matches, count(credential) as with_credentials from npi_matches" ), verified, ) if ( len(verification_rows) != 1 or not {"matches", "with_credentials"}.issubset(verification_rows[0]) ): verified.deopt( reason="The verification query is missing required aggregate columns.", evidence={"rows": verification_rows}, ) table_matches = int(verification_rows[0]["matches"]) if table_matches != saved.value["table_count"]: verified.deopt( reason="The destination count changed during verification.", evidence={ "before": saved.value["table_count"], "after": table_matches, }, ) verified.value = { "search": { "last_name": configured.value["last_name"], "state": configured.value["state"], "resolved_state": configured.value["state_value"], "exact_match": True, }, "source_url": page.url, "match_count": len(formatted.value), "results": formatted.value, "saved_count": len(saved.value["outcomes"]), "record_outcomes": saved.value["outcomes"], "evidence": { "final_results_heading": ( final_heading["text"] if final_heading else None ), "more_button_count": more_count, "empty_search_page": configured.value["empty_evidence"], "response_pages": collected.value["pages"], "store_table_count": table_matches, "store_with_credentials": int( verification_rows[0]["with_credentials"] ), }, "registry_limit": collected.value["limit"], "truncated_at_registry_limit": collected.value["truncated"], "notes": [ "All collected registry-returned matches are included without filtering their displayed current names or addresses.", "Missing credentials are represented as null.", ], } browser.done(result=verified.value)