import os
import re
import shutil
import subprocess
import tempfile
import gradio as gr
# Ensure workspace folder exists
WORKSPACE_DIR = os.path.abspath("./workspace")
os.makedirs(WORKSPACE_DIR, exist_ok=True)
# Helper to strip ANSI escape codes from terminal outputs
def strip_ansi(text):
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
return ansi_escape.sub('', text)
# Helper to get files list
def list_workspace_files():
files_list = []
for root, dirs, files in os.walk(WORKSPACE_DIR):
# Ignore hidden files/directories like .git or .qwen
if ".git" in root or ".qwen" in root:
continue
for file in files:
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, WORKSPACE_DIR)
files_list.append(rel_path)
return sorted(files_list)
# Load file content
def load_file_content(filepath):
if not filepath:
return "No file selected."
full_path = os.path.join(WORKSPACE_DIR, filepath)
if not os.path.exists(full_path) or os.path.isdir(full_path):
return f"File {filepath} not found or is a directory."
try:
with open(full_path, "r", encoding="utf-8", errors="ignore") as f:
return f.read()
except Exception as e:
return f"Error reading file: {str(e)}"
# Save file content
def save_file_content(filepath, content):
if not filepath:
return "No file selected.", list_workspace_files()
full_path = os.path.join(WORKSPACE_DIR, filepath)
try:
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "w", encoding="utf-8") as f:
f.write(content)
return f"Successfully saved {filepath}!", gr.update(choices=list_workspace_files(), value=filepath)
except Exception as e:
return f"Error saving file: {str(e)}", list_workspace_files()
# Initialize templates
def load_template(template_name):
# Clear existing workspace
for item in os.listdir(WORKSPACE_DIR):
item_path = os.path.join(WORKSPACE_DIR, item)
if item == ".qwen":
continue
if os.path.isdir(item_path):
shutil.rmtree(item_path)
else:
os.remove(item_path)
if template_name == "Python Math Library":
os.makedirs(os.path.join(WORKSPACE_DIR, "pymath"), exist_ok=True)
with open(os.path.join(WORKSPACE_DIR, "pymath", "math.py"), "w") as f:
f.write("def add(a, b):\n return a + b\n\ndef multiply(a, b):\n return a * b\n")
with open(os.path.join(WORKSPACE_DIR, "README.md"), "w") as f:
f.write("# Python Math Library\n\nThis is a simple math library. Ask Qwen-Code to add features or tests!\n")
elif template_name == "NodeJS Web App":
with open(os.path.join(WORKSPACE_DIR, "package.json"), "w") as f:
f.write('{\n "name": "simple-web-app",\n "version": "1.0.0",\n "main": "index.js",\n "dependencies": {}\n}\n')
with open(os.path.join(WORKSPACE_DIR, "index.js"), "w") as f:
f.write("console.log('Hello from NodeJS web app!');\n")
elif template_name == "HTML Landing Page":
with open(os.path.join(WORKSPACE_DIR, "index.html"), "w") as f:
f.write("\n\n
\n My Landing Page\n\n\n Welcome to my Qwen-Code Generated Page!
\n\n\n")
return f"Initialized workspace with '{template_name}' template.", gr.update(choices=list_workspace_files())
# Zip workspace for download
def download_workspace():
temp_dir = tempfile.gettempdir()
zip_path = os.path.join(temp_dir, "workspace_archive")
shutil.make_archive(zip_path, 'zip', WORKSPACE_DIR)
return zip_path + ".zip"
# Execute Qwen-Code agent
def run_agent(provider, api_key, model_name, custom_model, prompt, history):
if not api_key:
yield history, "⚠️ Please enter an API key under API settings first!", gr.update()
return
selected_model = custom_model if model_name == "custom" else model_name
if not selected_model:
yield history, "⚠️ Please select or specify a model first!", gr.update()
return
# Prepare command and env
cmd = ["qwen", "-p", prompt, "--yolo"]
env = os.environ.copy()
env["QWEN_SANDBOX"] = "false" # Run inside the container's environment natively
# Configure env variables based on provider
if provider == "DashScope (Qwen-compatible)":
env["DASHSCOPE_API_KEY"] = api_key
env["OPENAI_API_KEY"] = api_key
env["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"
env["OPENAI_MODEL"] = selected_model
elif provider == "Alibaba Cloud Coding Plan (Beijing)":
env["BAILIAN_CODING_PLAN_API_KEY"] = api_key
env["OPENAI_BASE_URL"] = "https://coding.dashscope.aliyuncs.com/v1"
env["OPENAI_MODEL"] = selected_model
elif provider == "Alibaba Cloud Coding Plan (International)":
env["BAILIAN_CODING_PLAN_API_KEY"] = api_key
env["OPENAI_BASE_URL"] = "https://coding-intl.dashscope.aliyuncs.com/v1"
env["OPENAI_MODEL"] = selected_model
elif provider == "OpenAI":
env["OPENAI_API_KEY"] = api_key
env["OPENAI_MODEL"] = selected_model
elif provider == "Anthropic Claude":
env["ANTHROPIC_API_KEY"] = api_key
env["ANTHROPIC_MODEL"] = selected_model
elif provider == "Google Gemini":
env["GEMINI_API_KEY"] = api_key
env["GEMINI_MODEL"] = selected_model
history.append((prompt, "... Starting Qwen Code Agent ..."))
yield history, "Initializing Qwen Code CLI process...", gr.update()
# Launch subprocess and read stdout in real-time
console_output = ""
try:
proc = subprocess.Popen(
cmd,
cwd=WORKSPACE_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1
)
for line in iter(proc.stdout.readline, ""):
cleaned_line = strip_ansi(line)
console_output += cleaned_line
# Update chat history and terminal output dynamically
history[-1] = (prompt, f"🤖 Agent is working...\n\n```\n{console_output[-4000:]}\n```")
yield history, console_output, gr.update()
proc.stdout.close()
return_code = proc.wait()
if return_code == 0:
status_msg = "✅ Qwen Code Agent finished successfully!"
else:
status_msg = f"❌ Qwen Code Agent exited with return code {return_code}."
history[-1] = (prompt, f"{status_msg}\n\n### Terminal Log Summary:\n```\n{console_output[-4000:]}\n```")
yield history, console_output, gr.update(choices=list_workspace_files())
except Exception as e:
error_msg = f"Failed to execute command: {str(e)}"
history[-1] = (prompt, f"❌ Error: {error_msg}")
yield history, error_msg, gr.update()
# UI Layout
with gr.Blocks(theme=gr.themes.Soft(primary_hue="purple", secondary_hue="blue")) as demo:
gr.HTML("""
🤖 Qwen-Code3 AI Coding Workspace
The ultimate terminal coding agent powered by Qwen. Connect your API key and let Qwen-Code code for you!
""")
with gr.Accordion("⚙️ API Configuration (Credentials are kept in-memory and never saved to disk)", open=True):
with gr.Row():
provider = gr.Dropdown(
label="API Provider",
choices=[
"DashScope (Qwen-compatible)",
"Alibaba Cloud Coding Plan (Beijing)",
"Alibaba Cloud Coding Plan (International)",
"OpenAI",
"Anthropic Claude",
"Google Gemini"
],
value="DashScope (Qwen-compatible)"
)
api_key = gr.Textbox(
label="API Key",
placeholder="Enter your sk-... or other provider credentials",
type="password"
)
model_name = gr.Dropdown(
label="Model",
choices=["qwen3-coder-plus", "qwen3.5-plus", "gpt-4o", "claude-3-5-sonnet-latest", "gemini-1.5-pro", "custom"],
value="qwen3-coder-plus"
)
custom_model = gr.Textbox(
label="Custom Model ID (if selected 'custom')",
placeholder="e.g. qwen-max",
visible=False
)
# Automatically show/hide custom model input
def update_model_visibility(model_choice):
return gr.update(visible=(model_choice == "custom"))
model_name.change(update_model_visibility, inputs=[model_name], outputs=[custom_model])
with gr.Tabs():
with gr.Tab("💬 Chat & Agent Agentic Terminal"):
with gr.Row():
with gr.Column(scale=3):
chatbot = gr.Chatbot(label="Agent Workspace Log", bubble_full_width=False, height=500)
prompt_input = gr.Textbox(
label="What should the Qwen Code Agent build or fix?",
placeholder="e.g. 'Build a python server in main.py that returns health status'",
lines=2
)
run_btn = gr.Button("🚀 Run Qwen Code Agent", variant="primary")
with gr.Column(scale=2):
console_log = gr.Textbox(
label="📟 Live Terminal Stdout (Scrolls with Agent progress)",
placeholder="Stdout will stream here...",
interactive=False,
lines=25,
max_lines=30
)
with gr.Tab("📂 Workspace File Explorer"):
with gr.Row():
with gr.Column(scale=1):
file_list = gr.Dropdown(
label="Workspace Files",
choices=list_workspace_files(),
interactive=True
)
refresh_files_btn = gr.Button("🔄 Refresh Files")
download_btn = gr.Button("📥 Download Workspace (.ZIP)", variant="secondary")
download_file_output = gr.File(label="Workspace Download Link")
with gr.Column(scale=3):
file_path_display = gr.Textbox(label="Editing File Path", interactive=False)
file_content_editor = gr.Code(label="File Content Editor / Viewer", language="python", lines=20)
save_file_btn = gr.Button("💾 Save Changes", variant="primary")
save_status = gr.Markdown()
# Explorer actions
def on_file_selected(filepath):
if not filepath:
return "", ""
return filepath, load_file_content(filepath)
file_list.change(on_file_selected, inputs=[file_list], outputs=[file_path_display, file_content_editor])
refresh_files_btn.click(
lambda: gr.update(choices=list_workspace_files()),
outputs=[file_list]
)
save_file_btn.click(
save_file_content,
inputs=[file_path_display, file_content_editor],
outputs=[save_status, file_list]
)
download_btn.click(
download_workspace,
outputs=[download_file_output]
)
with gr.Tab("📋 Workspace Templates"):
gr.Markdown("### Initialize your Workspace with a sample project template:")
with gr.Row():
py_template_btn = gr.Button("🐍 Python Math Library Template", variant="secondary")
node_template_btn = gr.Button("🟢 NodeJS Web App Template", variant="secondary")
html_template_btn = gr.Button("🎨 HTML Landing Page Template", variant="secondary")
template_status = gr.Markdown()
py_template_btn.click(
lambda: load_template("Python Math Library"),
outputs=[template_status, file_list]
)
node_template_btn.click(
lambda: load_template("NodeJS Web App"),
outputs=[template_status, file_list]
)
html_template_btn.click(
lambda: load_template("HTML Landing Page"),
outputs=[template_status, file_list]
)
# Connect Run Button
run_btn.click(
run_agent,
inputs=[provider, api_key, model_name, custom_model, prompt_input, chatbot],
outputs=[chatbot, console_log, file_list]
)
if __name__ == "__main__":
demo.queue()
demo.launch(server_name="0.0.0.0", server_port=7860)