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

fixing format

Browse Source
This commit is contained in:
Filipi Fuchter
2025-11-27 18:04:45 -03:00
parent 8ae27020a8
commit 94a7ffcbcf
1 changed files with 30 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
examples/aws-agentcore/bot_hellow.py
View File

@@ -1,7 +1,7 @@
import socket
import asyncio
import aioice
import socket
import aioice
from bedrock_agentcore import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@@ -10,7 +10,7 @@ app = BedrockAgentCoreApp()
def test_udp():
"""Test UDP connectivity using STUN server"""
stun_server = ("stun.l.google.com", 19302)
msg = b'\x00\x01\x00\x00' + b'\x21\x12\xa4\x42' + b'\x00' * 12
msg = b"\x00\x01\x00\x00" + b"\x21\x12\xa4\x42" + b"\x00" * 12
sock = None
try:
@@ -65,6 +65,7 @@ async def _async_turn_test(turn_server, turn_port, username, password, turn_tran
finally:
await connection.close()
def test_turn_with_auth(server, port, username, password, transport):
"""Sync wrapper for aioice TURN test."""
return asyncio.run(_async_turn_test(server, port, username, password, transport))
@@ -75,27 +76,38 @@ def comprehensive_network_test():
results = {}
# Test basic UDP connectivity
results['udp_stun'] = test_udp()
results["udp_stun"] = test_udp()
# TURN test list
turn_servers = [
(
"turn.cloudflare.com", # cleaned
3478,
"g001676bf83775dd93ddd77c69c80da1d13027af179ef51b4af7e3567a5028fd",
"a77444990e4fe3b82fd0fbe795f7499409edfbc3b170cea4b28155021ab6623c",
"udp"
"g0c069ee067cb5585f7a4799950312a355f2cb9c20e23d21bd1ecb1f93c1e8b0",
"d40a0992b4d08175af51f32f46a3732725a69045422f96b388420c98dd13ced3",
"udp",
),
(
"turn.cloudflare.com", # cleaned
5349,
"g0c069ee067cb5585f7a4799950312a355f2cb9c20e23d21bd1ecb1f93c1e8b0",
"d40a0992b4d08175af51f32f46a3732725a69045422f96b388420c98dd13ced3",
"tcp",
),
(
"turn.cloudflare.com", # cleaned
443,
"g0c069ee067cb5585f7a4799950312a355f2cb9c20e23d21bd1ecb1f93c1e8b0",
"d40a0992b4d08175af51f32f46a3732725a69045422f96b388420c98dd13ced3",
"tcp",
),
]
results['turn_tests'] = []
results["turn_tests"] = []
for host, port, username, password, transport in turn_servers:
success = test_turn_with_auth(host, port, username, password, transport)
results['turn_tests'].append({
"server": f"{host}:{port}",
"success": success
})
results["turn_tests"].append({"server": f"{host}:{port}", "success": success})
return results
@@ -124,9 +136,9 @@ def test_tcp_connectivity(host, port):
def my_agent(payload):
network_results = comprehensive_network_test()
udp_ok = network_results.get('udp_stun', False)
turn_ok = any(t['success'] for t in network_results['turn_tests'])
tcp_ok = network_results.get('tcp_test', False)
udp_ok = network_results.get("udp_stun", False)
turn_ok = any(t["success"] for t in network_results["turn_tests"])
tcp_ok = network_results.get("tcp_test", False)
connectivity_status = []
if udp_ok:
@@ -139,7 +151,9 @@ def my_agent(payload):
return {
"result": f"Hello {payload.get('name', 'World')}!",
"network_test_results": network_results,
"connectivity_status": ", ".join(connectivity_status) if connectivity_status else "No connectivity",
"connectivity_status": ", ".join(connectivity_status)
if connectivity_status
else "No connectivity",
"webrtc_feasible": udp_ok or turn_ok,
"turn_available": turn_ok,
}