-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmulti_endpoint_agent.py
More file actions
271 lines (236 loc) · 10.3 KB
/
Copy pathmulti_endpoint_agent.py
File metadata and controls
271 lines (236 loc) · 10.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
#!/usr/bin/env python3
"""
Multi-Endpoint Agent Example
This agent demonstrates serving multiple endpoints:
- /swml - Voice AI SWML endpoint
- /swaig - SWAIG webhook callbacks (automatically at /swml/swaig)
- / - Web UI with hello world
- /api - JSON API endpoint
- /static/file.txt - Static file serving
"""
import os
from pathlib import Path
from datetime import datetime
from fastapi import FastAPI, Request, Response, APIRouter
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
from signalwire import AgentBase
from signalwire.core.function_result import FunctionResult
class MultiEndpointAgent(AgentBase):
"""Agent that serves both SWML for voice and web UI endpoints"""
def __init__(self):
# Initialize with route=/swml for the voice AI endpoint
super().__init__(
name="multi-endpoint",
route="/swml", # This sets the SWML endpoint at /swml
host="0.0.0.0",
port=8080
)
# Configure the voice AI agent
self.prompt_add_section("Role", "You are a helpful voice assistant.")
self.prompt_add_section("Instructions", bullets=[
"Greet callers warmly",
"Be concise in your responses",
"Use the available functions when appropriate"
])
# Add a simple greeting skill
self.add_language("English", "en-US", "inworld.Mark")
# Set up static files directory
self.static_dir = Path(__file__).parent / "static_files"
self.static_dir.mkdir(exist_ok=True)
# Create a simple text file to serve
file_txt = self.static_dir / "file.txt"
file_txt.write_text("Hello World from static file!")
@AgentBase.tool(
name="get_time",
description="Get the current time",
parameters={}
)
def get_time(self, args, raw_data):
"""Simple SWAIG function for voice interactions"""
now = datetime.now().strftime("%I:%M %p")
return FunctionResult(f"The current time is {now}")
def _register_routes(self, router: APIRouter):
"""
Override route registration to add custom endpoints
Note: The parent class already registers /swml routes,
we're adding additional endpoints here
"""
# First, register the parent SWML routes
super()._register_routes(router)
# Now add our custom UI and API endpoints
# These will be available at the root level when we customize the app
def get_app(self):
"""
Override get_app to add custom root-level routes
"""
if self._app is None:
# Create the FastAPI app
app = FastAPI(
title="Multi-Endpoint Agent",
description="Agent with SWML, UI, and API endpoints"
)
# Add CORS middleware
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Register health check endpoints
@app.get("/health")
async def health_check():
return {"status": "healthy", "agent": self.get_name()}
# Add root UI endpoint
@app.get("/", response_class=HTMLResponse)
async def root_ui():
"""Serve a simple HTML UI"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Multi-Endpoint Agent</title>
<style>
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
background-color: #f5f5f5;
}
h1 {
color: #333;
text-align: center;
}
.endpoints {
background: white;
padding: 20px;
border-radius: 8px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.endpoint {
margin: 10px 0;
padding: 10px;
background: #f9f9f9;
border-radius: 4px;
}
.endpoint code {
background: #e0e0e0;
padding: 2px 6px;
border-radius: 3px;
}
.hello-world {
text-align: center;
font-size: 24px;
color: #4CAF50;
margin: 30px 0;
}
</style>
</head>
<body>
<h1>Multi-Endpoint Agent</h1>
<div class="hello-world">Hello World!</div>
<div class="endpoints">
<h2>Available Endpoints:</h2>
<div class="endpoint">
<strong>Voice AI (SWML):</strong> <code>POST /swml</code>
<br>SignalWire voice AI endpoint
</div>
<div class="endpoint">
<strong>SWAIG Webhooks:</strong> <code>POST /swml/swaig</code>
<br>Function callbacks for voice AI
</div>
<div class="endpoint">
<strong>Web UI:</strong> <code>GET /</code>
<br>This page you're viewing now
</div>
<div class="endpoint">
<strong>API:</strong> <code>GET /api</code>
<br>JSON API endpoint
</div>
<div class="endpoint">
<strong>Static Files:</strong> <code>GET /static/file.txt</code>
<br>Serves static text file
</div>
<div class="endpoint">
<strong>Health Check:</strong> <code>GET /health</code>
<br>Service health status
</div>
</div>
</body>
</html>
"""
return HTMLResponse(content=html_content)
# Add API endpoint
@app.get("/api")
async def api_endpoint():
"""Return JSON response"""
return JSONResponse(content={
"status": "ok",
"message": "Hello from API endpoint",
"timestamp": datetime.now().isoformat(),
"agent": self.get_name(),
"endpoints": {
"swml": "/swml",
"swaig": "/swml/swaig",
"ui": "/",
"api": "/api",
"static": "/static/file.txt"
}
})
# Add static file serving
@app.get("/static/{file_path:path}")
async def serve_static_file(file_path: str):
"""Serve static files"""
file_full_path = self.static_dir / file_path
# Security: ensure path doesn't escape static directory
try:
file_full_path = file_full_path.resolve()
if not str(file_full_path).startswith(str(self.static_dir.resolve())):
return PlainTextResponse("Access denied", status_code=403)
except Exception:
return PlainTextResponse("Invalid path", status_code=400)
# Check if file exists
if not file_full_path.exists() or not file_full_path.is_file():
return PlainTextResponse("File not found", status_code=404)
# Read and return file content
try:
content = file_full_path.read_text()
return PlainTextResponse(content)
except Exception as e:
return PlainTextResponse(f"Error reading file: {e}", status_code=500)
# Create router for SWML endpoints
router = self.as_router()
# Mount the SWML router at /swml
# This gives us /swml and /swml/swaig endpoints
app.include_router(router, prefix=self.route)
self._app = app
return self._app
def serve(self, host=None, port=None):
"""Override serve to use our custom app"""
import uvicorn
host = host or self.host or "0.0.0.0"
port = port or self.port or 8080
# Get our custom app with all endpoints
app = self.get_app()
# Get auth credentials
username, password, _ = self.get_basic_auth_credentials(include_source=True)
print(f"\nMulti-Endpoint Agent starting...")
print(f"Server: http://{host}:{port}")
print(f"Basic Auth: {username}:{password}")
print("\nEndpoints:")
print(f" Web UI: http://{host}:{port}/")
print(f" API: http://{host}:{port}/api")
print(f" Static: http://{host}:{port}/static/file.txt")
print(f" SWML: http://{host}:{port}/swml")
print(f" SWAIG: http://{host}:{port}/swml/swaig")
print(f" Health: http://{host}:{port}/health")
print("\nPress Ctrl+C to stop\n")
try:
uvicorn.run(app, host=host, port=port)
except KeyboardInterrupt:
print("\nStopping agent...")
if __name__ == "__main__":
agent = MultiEndpointAgent()
agent.serve()