wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Trying to fix the server

Browse Source
This commit is contained in:
Filipi Fuchter
2025-11-28 12:06:43 -03:00
parent 6e88619e0d
commit 9d5acb12c7
1 changed files with 34 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
examples/aws-agentcore/server/server.py
View File

@@ -72,81 +72,53 @@ async def offer(request: Request):
raise HTTPException(500, "Bedrock response stream missing") raise HTTPException(500, "Bedrock response stream missing")
answer_sdp = None answer_sdp = None
buffer = "" # Storage for incomplete lines across chunks
# Flag to break the outer chunk loop once the answer is found # --- Start Streaming Loop using iter_lines() ---
answer_found = False for line in streaming_body.iter_lines():
print(f"Not decoded line: {line}")
# --- Start Streaming Loop --- # iter_lines yields bytes, so we must decode it to a string.
for chunk in streaming_body.iter_chunks(): try:
if answer_found: decoded_line = line.decode("utf-8")
# If we found the answer, break the chunk loop immediately. except UnicodeDecodeError:
break print("Warning: Could not decode line from stream.")
continue
decoded_chunk = chunk.decode("utf-8") # 1. Clean up and strip whitespace
buffer += decoded_chunk processed_line = decoded_line.strip()
# Use splitlines(keepends=True) for reliable line-by-line processing print(f"Processing line: {processed_line}")
lines = buffer.splitlines(keepends=True)
buffer = "" # Reset buffer
# Process all complete lines if not processed_line:
for i, line in enumerate(lines): continue
print(f"Processing line {i}: {line}")
# Check for incomplete line (last item in lines list without a newline) # 2. Check for the SSE 'data:' prefix (now comparing string to string)
if i == len(lines) - 1 and not line.endswith("\n"): if processed_line.startswith("data:"):
buffer = line # Extract the content *after* "data: "
continue sse_data = processed_line[len("data:") :].strip()
print("Extracted SSE Data payload:", sse_data)
# 1. Clean up and strip whitespace # 3. Parse the extracted payload as JSON
processed_line = line.strip() try:
event = json.loads(sse_data)
print("Received event:", event)
if not processed_line: # 4. Check for the 'answer' key
continue if "answer" in event:
payload = event["answer"]
print("Raw line received:", processed_line) if payload.get("type") == "answer":
answer_sdp = payload
print("WebRTC answer found. Stopping stream processing.")
# Break the line loop immediately
break
# 2. Check for the SSE 'data:' prefix except json.JSONDecodeError:
if processed_line.startswith("data:"): print(f"Failed to parse extracted SSE payload as JSON: {sse_data}")
# Extract the content *after* "data: " pass
sse_data = processed_line[len("data:") :].strip()
print("Extracted SSE Data payload:", sse_data)
# 3. Parse the extracted payload as JSON
try:
event = json.loads(sse_data)
print("Received event:", event)
# 4. Check for the 'answer' key
if "answer" in event:
payload = event["answer"]
if payload.get("type") == "answer":
answer_sdp = payload
answer_found = True # Set flag
print("WebRTC answer found. Stopping stream processing.")
# Break the inner line loop
break
except json.JSONDecodeError:
print(f"Failed to parse extracted SSE payload as JSON: {sse_data}")
pass
# --- End Streaming Loop --- # --- End Streaming Loop ---
# After the loop, the `StreamingBody` should be exhausted.
if answer_sdp is None:
# Check buffer only if answer wasn't found (for maximum safety)
if buffer.strip().startswith("data:"):
try:
sse_data = buffer.strip()[len("data:") :].strip()
event = json.loads(sse_data)
if "answer" in event and event["answer"].get("type") == "answer":
answer_sdp = event["answer"]
except Exception:
pass
if answer_sdp is None: if answer_sdp is None:
raise HTTPException(500, "Did not find WebRTC answer in agent output") raise HTTPException(500, "Did not find WebRTC answer in agent output")