Repository navigation
Expand file tree
/
Copy path_cli.py
More file actions
203 lines (156 loc) · 6.46 KB
/
Copy path_cli.py
File metadata and controls
203 lines (156 loc) · 6.46 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
# Copyright (c) Microsoft. All rights reserved.
"""Command line interface for Agent Framework DevUI."""
import argparse
import logging
import os
import sys
logger = logging.getLogger(__name__)
def setup_logging(level: str = "INFO") -> None:
"""Configure logging for the server."""
log_format = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
logging.basicConfig(level=getattr(logging, level.upper()), format=log_format, datefmt="%Y-%m-%d %H:%M:%S")
def create_cli_parser() -> argparse.ArgumentParser:
"""Create the command line argument parser."""
parser = argparse.ArgumentParser(
prog="devui",
description="Launch Agent Framework DevUI - Debug interface with OpenAI compatible API",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
devui # Scan current directory
devui ./agents # Scan specific directory
devui --port 8000 # Custom port
devui --headless # API only, no UI
devui --instrumentation # Enable OpenTelemetry instrumentation
""",
)
parser.add_argument(
"directory", nargs="?", default=".", help="Directory to scan for entities (default: current directory)"
)
parser.add_argument("--port", "-p", type=int, default=8080, help="Port to run server on (default: 8080)")
parser.add_argument("--host", default="127.0.0.1", help="Host to bind server to (default: 127.0.0.1)")
parser.add_argument("--no-open", action="store_true", help="Don't automatically open browser")
parser.add_argument("--headless", action="store_true", help="Run without UI (API only)")
parser.add_argument(
"--log-level",
choices=["DEBUG", "INFO", "WARNING", "ERROR"],
default="INFO",
help="Logging level (default: INFO)",
)
parser.add_argument("--reload", action="store_true", help="Enable auto-reload for development")
parser.add_argument("--instrumentation", action="store_true", help="Enable OpenTelemetry instrumentation")
parser.add_argument(
"--mode",
choices=["developer", "user"],
default=None,
help="Server mode - 'developer' (full access, verbose errors) or 'user' (restricted APIs, generic errors)",
)
# Add --dev/--no-dev as a convenient alternative to --mode
parser.add_argument(
"--dev",
dest="dev_mode",
action="store_true",
default=None,
help="Enable developer mode (shorthand for --mode developer)",
)
parser.add_argument(
"--no-dev",
dest="dev_mode",
action="store_false",
help="Disable developer mode (shorthand for --mode user)",
)
parser.add_argument(
"--no-auth",
action="store_true",
help=(
"Disable Bearer token authentication for loopback-only local development. Non-loopback hosts require auth."
),
)
parser.add_argument(
"--auth-token",
type=str,
help="Custom Bearer token. Required for non-loopback hosts when DEVUI_AUTH_TOKEN is not set.",
)
parser.add_argument("--version", action="version", version=f"Agent Framework DevUI {get_version()}")
return parser
def get_version() -> str:
"""Get the package version."""
try:
from . import __version__
return __version__
except ImportError:
return "unknown"
def validate_directory(directory: str) -> str:
"""Validate and normalize the entities directory."""
if not directory:
directory = "."
abs_dir = os.path.abspath(directory)
if not os.path.exists(abs_dir):
print(f"Error: Directory '{directory}' does not exist", file=sys.stderr) # ruff:ignore[print]
sys.exit(1)
if not os.path.isdir(abs_dir):
print(f"Error: '{directory}' is not a directory", file=sys.stderr) # ruff:ignore[print]
sys.exit(1)
return abs_dir
def print_startup_info(
entities_dir: str, host: str, port: int, ui_enabled: bool, reload: bool, auth_token: str | None = None
) -> None:
"""Print startup information."""
print("Agent Framework DevUI") # ruff:ignore[print]
print("=" * 50) # ruff:ignore[print]
print(f"Entities directory: {entities_dir}") # ruff:ignore[print]
print(f"Server URL: http://{host}:{port}") # ruff:ignore[print]
print(f"UI enabled: {'Yes' if ui_enabled else 'No'}") # ruff:ignore[print]
print(f"Auto-reload: {'Yes' if reload else 'No'}") # ruff:ignore[print]
# Display auth token if authentication is enabled
if auth_token:
print("Authentication: Enabled") # ruff:ignore[print]
print(f"Auth token: {auth_token}") # ruff:ignore[print]
print("💡 Use this token in Authorization: Bearer <token> header") # ruff:ignore[print]
print("=" * 50) # ruff:ignore[print]
print("Scanning for entities...") # ruff:ignore[print]
def main() -> None:
"""Main CLI entry point."""
parser = create_cli_parser()
args = parser.parse_args()
# Setup logging
setup_logging(args.log_level)
# Validate directory
entities_dir = validate_directory(args.directory)
# Extract parameters directly from args
ui_enabled = not args.headless
# Determine mode from --mode or --dev/--no-dev flags
if args.dev_mode is not None:
# --dev or --no-dev was specified
mode = "developer" if args.dev_mode else "user"
elif args.mode is not None:
# --mode was specified
mode = args.mode
else:
# Default to developer mode
mode = "developer"
# Print startup info (don't show token - serve() will handle it)
print_startup_info(entities_dir, args.host, args.port, ui_enabled, args.reload, None)
# Import and start server
try:
from . import serve
serve(
entities_dir=entities_dir,
port=args.port,
host=args.host,
auto_open=not args.no_open,
ui_enabled=ui_enabled,
instrumentation_enabled=args.instrumentation,
mode=mode,
auth_enabled=not args.no_auth,
auth_token=args.auth_token, # Pass through explicit token only
)
except KeyboardInterrupt:
print("\nShutting down Agent Framework DevUI...") # ruff:ignore[print]
sys.exit(0)
except Exception as e:
logger.exception("Failed to start server")
print(f"Error: {e}", file=sys.stderr) # ruff:ignore[print]
sys.exit(1)
if __name__ == "__main__":
main()