How to Build a HIPAA-Compliant AI Medical Scribe App Using AWS Bedrock and Python
Why this architecture
An AI medical scribe ingests an audio (or live-streamed) clinician-patient encounter, produces a timestamped transcript, and drafts a structured clinical note (SOAP: Subjective, Objective, Assessment, Plan) for the clinician to review and sign. Because the input and output both contain Protected Health Information (PHI), every service in the pipeline must sit inside AWS's HIPAA-eligible services list and be covered by your AWS Business Associate Addendum (BAA). This tutorial builds that pipeline with Amazon Transcribe Medical for speech-to-text and Amazon Bedrock (Anthropic Claude models) for note drafting, glued together with Python running on AWS Lambda and Step Functions.
This is engineering guidance, not legal advice — a HIPAA-eligible service does not by itself make your application compliant. You are responsible for signing the AWS BAA, enabling it for every service you use, running a risk assessment, and getting your own compliance/legal sign-off before handling real patient data.
1. Technical architecture
┌─────────────┐ presigned PUT ┌──────────────────┐
│ Clinician │ ───────────────────▶ │ S3 (encrypted, │
│ client app │ │ raw-audio/) │
└─────────────┘ └─────────┬─────────┘
│ S3:ObjectCreated
▼
┌────────────────────┐
│ EventBridge rule │
└─────────┬───────────┘
▼
┌────────────────────┐
│ Step Functions │
│ state machine │
└─────────┬───────────┘
┌────────────────────────────┼─────────────────────────────┐
▼ ▼ ▼
┌────────────────────┐ ┌─────────────────────┐ ┌──────────────────────┐
│ Lambda: start │ │ Lambda: poll │ │ Lambda: draft note │
│ Transcribe Medical │ ──▶ │ transcription job │ ──▶ │ (Bedrock / Claude) │
│ job │ │ (retry w/ backoff) │ │ structured SOAP JSON │
└────────────────────┘ └─────────────────────┘ └──────────┬───────────┘
▼
┌──────────────────────┐
│ DynamoDB: encounter │
│ notes table (KMS CMK) │
└──────────┬───────────┘
▼
┌──────────────────────┐
│ Clinician review UI │
│ (sign-off required) │
└──────────────────────┘
Key controls baked into the diagram:
- No PHI ever touches a non-BAA-covered service. S3, Transcribe Medical, Bedrock, Lambda, Step Functions, EventBridge, and DynamoDB are all on AWS's HIPAA-eligible services list, and all must be explicitly enabled under your BAA.
- Encryption at rest via S3 SSE-KMS and DynamoDB with a customer-managed KMS key (CMK), and encryption in transit via TLS everywhere (enforced with bucket policies).
- Human-in-the-loop: the LLM draft is never auto-filed into the EHR. A clinician must review and explicitly sign off (Section 4 covers why this matters for hallucination risk).
- Audit trail: every state transition is logged to CloudTrail and the Step Functions execution history, satisfying the HIPAA Security Rule's audit control requirement (§164.312(b)).
2. Complete code implementation
2.1 Infrastructure (AWS CDK, Python)
# infra/scribe_stack.py
from aws_cdk import (
Stack,
Duration,
RemovalPolicy,
aws_s3 as s3,
aws_kms as kms,
aws_dynamodb as dynamodb,
aws_lambda as _lambda,
aws_stepfunctions as sfn,
aws_stepfunctions_tasks as tasks,
aws_events as events,
aws_events_targets as targets,
aws_iam as iam,
)
from constructs import Construct
class MedicalScribeStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Customer-managed KMS key — required so we control key policy and
# rotation independently of the AWS-managed S3/DynamoDB defaults.
phi_key = kms.Key(
self, "PhiCmk",
enable_key_rotation=True,
removal_policy=RemovalPolicy.RETAIN,
)
raw_audio_bucket = s3.Bucket(
self, "RawAudioBucket",
encryption=s3.BucketEncryption.KMS,
encryption_key=phi_key,
enforce_ssl=True,
block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
lifecycle_rules=[
s3.LifecycleRule(expiration=Duration.days(30)) # retention per your policy
],
versioned=True,
)
encounters_table = dynamodb.Table(
self, "EncounterNotes",
partition_key=dynamodb.Attribute(name="encounter_id", type=dynamodb.AttributeType.STRING),
sort_key=dynamodb.Attribute(name="version", type=dynamodb.AttributeType.NUMBER),
encryption=dynamodb.TableEncryption.CUSTOMER_MANAGED,
encryption_key=phi_key,
point_in_time_recovery=True,
removal_policy=RemovalPolicy.RETAIN,
)
start_transcribe_fn = _lambda.Function(
self, "StartTranscribeFn",
runtime=_lambda.Runtime.PYTHON_3_12,
handler="start_transcribe.handler",
code=_lambda.Code.from_asset("lambdas"),
timeout=Duration.seconds(30),
environment={"OUTPUT_BUCKET": raw_audio_bucket.bucket_name},
)
raw_audio_bucket.grant_read(start_transcribe_fn)
raw_audio_bucket.grant_write(start_transcribe_fn)
start_transcribe_fn.add_to_role_policy(iam.PolicyStatement(
actions=["transcribe:StartMedicalTranscriptionJob"],
resources=["*"],
))
poll_transcribe_fn = _lambda.Function(
self, "PollTranscribeFn",
runtime=_lambda.Runtime.PYTHON_3_12,
handler="poll_transcribe.handler",
code=_lambda.Code.from_asset("lambdas"),
timeout=Duration.seconds(30),
)
poll_transcribe_fn.add_to_role_policy(iam.PolicyStatement(
actions=["transcribe:GetMedicalTranscriptionJob"],
resources=["*"],
))
draft_note_fn = _lambda.Function(
self, "DraftNoteFn",
runtime=_lambda.Runtime.PYTHON_3_12,
handler="draft_note.handler",
code=_lambda.Code.from_asset("lambdas"),
timeout=Duration.seconds(60),
environment={
"TABLE_NAME": encounters_table.table_name,
"BEDROCK_MODEL_ID": "anthropic.claude-sonnet-4-5-20250929-v1:0",
},
)
encounters_table.grant_write_data(draft_note_fn)
draft_note_fn.add_to_role_policy(iam.PolicyStatement(
actions=["bedrock:InvokeModel"],
resources=["*"], # scope to the specific model ARN in production
))
raw_audio_bucket.grant_read(draft_note_fn)
# Step Functions: start -> poll (retry) -> draft note
start_task = tasks.LambdaInvoke(self, "StartJob", lambda_function=start_transcribe_fn,
output_path="$.Payload")
poll_task = tasks.LambdaInvoke(self, "PollJob", lambda_function=poll_transcribe_fn,
output_path="$.Payload")
wait = sfn.Wait(self, "WaitBeforePoll", time=sfn.WaitTime.duration(Duration.seconds(15)))
draft_task = tasks.LambdaInvoke(self, "DraftNote", lambda_function=draft_note_fn,
output_path="$.Payload")
is_complete = sfn.Choice(self, "JobComplete?")
definition = start_task.next(wait).next(poll_task).next(
is_complete
.when(sfn.Condition.string_equals("$.status", "COMPLETED"), draft_task)
.when(sfn.Condition.string_equals("$.status", "FAILED"),
sfn.Fail(self, "TranscriptionFailed", cause="Transcribe Medical job failed"))
.otherwise(wait)
)
state_machine = sfn.StateMachine(
self, "ScribeStateMachine",
definition_body=sfn.DefinitionBody.from_chainable(definition),
timeout=Duration.minutes(30),
)
rule = events.Rule(
self, "OnAudioUploaded",
event_pattern=events.EventPattern(
source=["aws.s3"],
detail_type=["Object Created"],
detail={"bucket": {"name": [raw_audio_bucket.bucket_name]}, "object": {"key": [{"prefix": "raw-audio/"}]}},
),
)
rule.add_target(targets.SfnStateMachine(state_machine))
2.2 Lambda: start the Transcribe Medical job
# lambdas/start_transcribe.py
import os
import time
import boto3
from botocore.exceptions import ClientError
transcribe = boto3.client("transcribe")
OUTPUT_BUCKET = os.environ["OUTPUT_BUCKET"]
def handler(event, context):
"""Triggered by an EventBridge rule on S3 ObjectCreated for raw-audio/*.
event["detail"] contains the S3 bucket/key of the newly uploaded audio.
"""
detail = event["detail"]
bucket = detail["bucket"]["name"]
key = detail["object"]["key"]
encounter_id = key.split("/")[-1].split(".")[0]
job_name = f"scribe-{encounter_id}-{int(time.time())}"
try:
transcribe.start_medical_transcription_job(
MedicalTranscriptionJobName=job_name,
LanguageCode="en-US",
MediaFormat="wav",
Media={"MediaFileUri": f"s3://{bucket}/{key}"},
OutputBucketName=OUTPUT_BUCKET,
OutputKey=f"transcripts/{encounter_id}.json",
Specialty="PRIMARYCARE",
Type="CONVERSATION",
Settings={
"ShowSpeakerLabels": True,
"MaxSpeakerLabels": 2, # clinician + patient
"ChannelIdentification": False,
},
)
except ClientError as exc:
code = exc.response["Error"]["Code"]
if code == "ConflictException":
# Job name collision — extremely unlikely given the timestamp suffix,
# but treat as retryable rather than a hard failure.
raise
if code == "BadRequestException":
# Usually an unsupported audio codec/sample rate. Fail fast — retrying
# will not fix a malformed input file.
return {"status": "FAILED", "encounter_id": encounter_id, "reason": str(exc)}
raise
return {"status": "IN_PROGRESS", "job_name": job_name, "encounter_id": encounter_id}
2.3 Lambda: poll job status
# lambdas/poll_transcribe.py
import boto3
transcribe = boto3.client("transcribe")
def handler(event, context):
job_name = event["job_name"]
resp = transcribe.get_medical_transcription_job(MedicalTranscriptionJobName=job_name)
job = resp["MedicalTranscriptionJob"]
status = job["TranscriptionJobStatus"] # IN_PROGRESS | COMPLETED | FAILED
result = {"status": status, "job_name": job_name, "encounter_id": event["encounter_id"]}
if status == "COMPLETED":
result["transcript_uri"] = job["Transcript"]["TranscriptFileUri"]
if status == "FAILED":
result["reason"] = job.get("FailureReason", "unknown")
return result
2.4 Lambda: draft the SOAP note with Bedrock
# lambdas/draft_note.py
import json
import os
import time
import boto3
import urllib.request
from datetime import datetime, timezone
from botocore.exceptions import ClientError
bedrock = boto3.client("bedrock-runtime")
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
MODEL_ID = os.environ["BEDROCK_MODEL_ID"]
SYSTEM_PROMPT = """You are a clinical documentation assistant. You will be given a
diarized transcript of a patient encounter. Produce a SOAP note as strict JSON with
keys: subjective, objective, assessment, plan. Only use information present in the
transcript. If a section has no supporting content, set its value to "Not discussed
during this encounter" rather than inferring or inventing clinical findings. Do not
suggest diagnoses that are not explicitly stated by the clinician in the transcript."""
def _fetch_transcript(transcript_uri: str) -> str:
with urllib.request.urlopen(transcript_uri) as resp:
payload = json.loads(resp.read())
items = payload["results"]["transcripts"]
return " ".join(t["transcript"] for t in items)
def _invoke_bedrock_with_retry(transcript_text: str, max_attempts: int = 4) -> dict:
body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 1500,
"system": SYSTEM_PROMPT,
"messages": [{"role": "user", "content": f"Transcript:\n\n{transcript_text}"}],
}
attempt = 0
while True:
try:
resp = bedrock.invoke_model(modelId=MODEL_ID, body=json.dumps(body))
payload = json.loads(resp["body"].read())
text = payload["content"][0]["text"]
return json.loads(text)
except ClientError as exc:
code = exc.response["Error"]["Code"]
attempt += 1
if code == "ThrottlingException" and attempt < max_attempts:
time.sleep(2 ** attempt) # exponential backoff: 2s, 4s, 8s
continue
raise
except (json.JSONDecodeError, KeyError, IndexError) as exc:
# Model returned non-JSON or an unexpected shape. Treat as a
# hallucination/formatting failure — do not retry blindly, since a
# malformed prompt will fail the same way every time. Surface it.
raise ValueError(f"Bedrock returned unparseable note structure: {exc}") from exc
def handler(event, context):
encounter_id = event["encounter_id"]
transcript_text = _fetch_transcript(event["transcript_uri"])
try:
soap = _invoke_bedrock_with_retry(transcript_text)
except ValueError as exc:
# Persist a FAILED draft so the clinician UI shows "note generation
# failed, transcript available" instead of silently losing the encounter.
table.put_item(Item={
"encounter_id": encounter_id,
"version": int(time.time()),
"status": "DRAFT_FAILED",
"error": str(exc),
"transcript_text": transcript_text,
})
return {"status": "DRAFT_FAILED", "encounter_id": encounter_id}
table.put_item(Item={
"encounter_id": encounter_id,
"version": int(time.time()),
"status": "PENDING_CLINICIAN_REVIEW",
"soap_note": soap,
"transcript_text": transcript_text,
"generated_at": datetime.now(timezone.utc).isoformat(),
"model_id": MODEL_ID,
})
return {"status": "DRAFTED", "encounter_id": encounter_id}
3. Step-by-step configuration guide
- Sign and scope the BAA. In AWS Organizations/Account settings, accept the AWS Business Associate Addendum, then confirm in the AWS Artifact console that Transcribe Medical, Bedrock, S3, Lambda, Step Functions, EventBridge, and DynamoDB are all listed as HIPAA-eligible services covered by your BAA.
- Request Bedrock model access. In the Bedrock console → Model access, request access to the Anthropic Claude model you intend to use. Approval is typically immediate for on-demand access but can take longer for provisioned throughput.
- Create the CMK and stack.
pip install aws-cdk-lib constructs boto3 --break-system-packages cdk bootstrap aws://<ACCOUNT_ID>/<REGION> cdk deploy MedicalScribeStack --require-approval never - Set environment variables on the
DraftNoteFnLambda (already wired via CDK above, but if configuring manually via console):TABLE_NAME,BEDROCK_MODEL_ID. - Enable S3 EventBridge notifications on the raw-audio bucket:
aws s3api put-bucket-notification-configuration --bucket <bucket> --notification-configuration '{"EventBridgeConfiguration": {}}'. - Test end-to-end by uploading a short, synthetic (non-PHI) WAV file to
raw-audio/test-encounter-001.wavand watching the Step Functions execution in the console — each state (StartJob,WaitBeforePoll,PollJob,DraftNote) should transition green. - Lock down IAM. Replace the wildcard
resources=["*"]in the CDK snippet above with the exact Bedrock model ARN and Transcribe job ARN pattern before going to production — this tutorial leaves it broad only for initial testing.
4. Error handling and edge cases
- Transcription job failures (
FAILEDstatus from Transcribe Medical) are usually caused by unsupported audio codecs, corrupted files, or audio under 1 second. The Step FunctionsChoicestate above routes these to an explicitFailstate rather than looping forever — surface this to the clinician as "please re-record." - Bedrock throttling (
ThrottlingException) is handled with exponential backoff and a capped retry count in_invoke_bedrock_with_retry. Pastmax_attempts, the exception propagates so Step Functions marks the execution failed and it shows up in your dashboards rather than being silently swallowed. - Model hallucination / malformed output is the highest-risk failure mode in a clinical setting. Three mitigations are in the code above: (1) the system prompt explicitly forbids inferring findings not present in the transcript, (2) the Lambda validates that the response is parseable JSON with the expected keys and treats a parse failure as a hard error rather than passing a broken note downstream, and (3) no note is ever auto-filed — every draft lands in
PENDING_CLINICIAN_REVIEWand a human must sign off before it can be exported to the EHR. - Long encounters exceeding model context — for visits over roughly 30 minutes, chunk the transcript by speaker turn and summarize incrementally (map-reduce style) rather than sending the full transcript in one call; a single oversized request will hit Bedrock's input token limit and return a
ValidationException. - DynamoDB write contention is avoided by using
encounter_id+version(epoch timestamp) as a composite key, so retried Step Functions executions never overwrite a prior draft — you get a full version history per encounter instead of a race condition. - Presigned URL expiry for the transcript output —
TranscriptFileUriURLs from Transcribe Medical are time-limited; thedraft_noteLambda fetches the transcript immediately within the same execution rather than caching the URL for later use.
