How to Build a QwenPaw Agent Workspace with Custom Skills, Model Providers, Console Access, and Streaming API Testing

import os
import sys
import json
import time
import uuid
import shlex
import sign
import shutil
import socket
import secrets and techniques
import pathlib
import subprocess
from datetime import datetime
RESET_QWENPAW = False
PORT = int(os.environ.get("QWENPAW_COLAB_PORT", "8088"))
ROOT = pathlib.Path("/content material/qwenpaw_colab")
WORKING_DIR = ROOT / "working"
SECRET_DIR = ROOT / "secrets and techniques"
LOG_DIR = ROOT / "logs"
WORKSPACE_DIR = WORKING_DIR / "workspaces" / "default"
PID_FILE = ROOT / "qwenpaw_app.pid"
APP_LOG = LOG_DIR / "qwenpaw_app.log"
if RESET_QWENPAW and ROOT.exists():
   shutil.rmtree(ROOT)
for p in [ROOT, WORKING_DIR, SECRET_DIR, LOG_DIR, WORKSPACE_DIR]:
   p.mkdir(dad and mom=True, exist_ok=True)
os.environ["QWENPAW_WORKING_DIR"] = str(WORKING_DIR)
os.environ["QWENPAW_SECRET_DIR"] = str(SECRET_DIR)
os.environ["QWENPAW_AUTH_ENABLED"] = "true"
os.environ["QWENPAW_AUTH_USERNAME"] = os.environ.get("QWENPAW_AUTH_USERNAME", "admin")
os.environ["QWENPAW_LOG_LEVEL"] = os.environ.get("QWENPAW_LOG_LEVEL", "information")
os.environ["QWENPAW_SKILL_SCAN_MODE"] = os.environ.get("QWENPAW_SKILL_SCAN_MODE", "warn")
os.environ["QWENPAW_TOOL_GUARD_ENABLED"] = os.environ.get("QWENPAW_TOOL_GUARD_ENABLED", "true")
password_file = SECRET_DIR / ".colab_ui_password"
if not password_file.exists():
   password_file.write_text("qpw-" + secrets and techniques.token_urlsafe(18), encoding="utf-8")
os.environ["QWENPAW_AUTH_PASSWORD"] = password_file.read_text(encoding="utf-8").strip()
def run(cmd, verify=False, env=None, cwd=None, stream=False):
   if isinstance(cmd, str):
       display_cmd = cmd
       shell = True
   else:
       display_cmd = " ".be part of(shlex.quote(str(x)) for x in cmd)
       shell = False
   print(f"n$ {display_cmd}")
   if stream:
       proc = subprocess.Popen(cmd, shell=shell, env=env, cwd=cwd, textual content=True)
       rc = proc.wait()
       if verify and rc != 0:
           elevate RuntimeError(f"Command failed with exit code {rc}: {display_cmd}")
       return rc, ""
   out = subprocess.run(
       cmd,
       shell=shell,
       env=env,
       cwd=cwd,
       textual content=True,
       stdout=subprocess.PIPE,
       stderr=subprocess.STDOUT,
   )
   print(out.stdout[-4000:])
   if verify and out.returncode != 0:
       elevate RuntimeError(f"Command failed with exit code {out.returncode}: {display_cmd}")
   return out.returncode, out.stdout
def port_open(host="127.0.0.1", port=8088, timeout=0.5):
   attempt:
       with socket.create_connection((host, port), timeout=timeout):
           return True
   besides OSError:
       return False
def wait_for_port(port, seconds=90):
   begin = time.time()
   whereas time.time() - begin < seconds:
       if port_open("127.0.0.1", port):
           return True
       time.sleep(1)
   return False
def stop_previous_app():
   if PID_FILE.exists():
       attempt:
           pid = int(PID_FILE.read_text().strip())
           os.kill(pid, sign.SIGTERM)
           time.sleep(2)
           attempt:
               os.kill(pid, 0)
               os.kill(pid, sign.SIGKILL)
           besides OSError:
               cross
       besides Exception:
           cross
       PID_FILE.unlink(missing_ok=True)
def qwenpaw_cmd(*args):
   exe = shutil.which("qwenpaw")
   if exe:
       return [exe, *args]
   return [sys.executable, "-m", "qwenpaw", *args]
def colab_secret_or_env(identify):
   worth = os.environ.get(identify, "")
   attempt:
       from google.colab import userdata
       secret_value = userdata.get(identify)
       if secret_value:
           worth = secret_value
   besides Exception:
       cross
   return worth or ""
print("Python:", sys.model)
assert sys.version_info >= (3, 10), "QwenPaw wants Python 3.10+."
pip_spec = os.environ.get("QWENPAW_PIP_SPEC", "qwenpaw")
run([sys.executable, "-m", "pip", "install", "-q", "-U", "pip", "setuptools", "wheel"], verify=False)
run([sys.executable, "-m", "pip", "install", "-q", "-U", pip_spec, "requests"], verify=True)
attempt:
   import requests
besides Exception:
   run([sys.executable, "-m", "pip", "install", "-q", "-U", "requests"], verify=True)
   import requests

if not (WORKING_DIR / "config.json").exists():
   run(qwenpaw_cmd("init", "--defaults"), verify=False)
else:
   print("QwenPaw working listing already initialized:", WORKING_DIR)
provider_candidates = [
   {
       "env": "OPENAI_API_KEY",
       "provider_id": "openai",
       "name": "OpenAI",
       "base_url": "https://api.openai.com/v1",
       "model": os.environ.get("QWENPAW_MODEL", "gpt-4o-mini"),
       "chat_model": "OpenAIChatModel",
       "prefix": "sk-",
   },
   {
       "env": "OPENROUTER_API_KEY",
       "provider_id": "openrouter",
       "name": "OpenRouter",
       "base_url": "https://openrouter.ai/api/v1",
       "model": os.environ.get("QWENPAW_MODEL", "openai/gpt-4o-mini"),
       "chat_model": "OpenAIChatModel",
       "prefix": "sk-or-",
   },
   {
       "env": "DASHSCOPE_API_KEY",
       "provider_id": "dashscope",
       "name": "DashScope",
       "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
       "model": os.environ.get("QWENPAW_MODEL", "qwen-plus"),
       "chat_model": "OpenAIChatModel",
       "prefix": "sk-",
   },
   {
       "env": "DEEPSEEK_API_KEY",
       "provider_id": "deepseek",
       "name": "DeepSeek",
       "base_url": "https://api.deepseek.com",
       "model": os.environ.get("QWENPAW_MODEL", "deepseek-chat"),
       "chat_model": "OpenAIChatModel",
       "prefix": "sk-",
   },
   {
       "env": "GEMINI_API_KEY",
       "provider_id": "gemini",
       "name": "Google Gemini",
       "base_url": "https://generativelanguage.googleapis.com",
       "model": os.environ.get("QWENPAW_MODEL", "gemini-2.5-flash"),
       "chat_model": "GeminiChatModel",
       "prefix": "",
   },
   {
       "env": "GOOGLE_API_KEY",
       "provider_id": "gemini",
       "name": "Google Gemini",
       "base_url": "https://generativelanguage.googleapis.com",
       "model": os.environ.get("QWENPAW_MODEL", "gemini-2.5-flash"),
       "chat_model": "GeminiChatModel",
       "prefix": "",
   },
]
chosen = None
for candidate in provider_candidates:
   api_key = colab_secret_or_env(candidate["env"])
   if api_key:
       chosen = {**candidate, "api_key": api_key}
       break
def read_json(path, default):
   attempt:
       if path.exists():
           return json.hundreds(path.read_text(encoding="utf-8"))
   besides Exception:
       cross
   return default
def write_json(path, information):
   path.guardian.mkdir(dad and mom=True, exist_ok=True)
   path.write_text(json.dumps(information, indent=2, ensure_ascii=False), encoding="utf-8")
config_path = WORKING_DIR / "config.json"
config = read_json(config_path, {})
config.setdefault("brokers", {})
config["agents"].setdefault("active_agent", "default")
config["agents"].setdefault("agent_order", ["default"])
config["agents"].setdefault("profiles", {})
config["agents"]["profiles"].setdefault("default", {})
config["agents"]["profiles"]["default"].replace(
   {
       "id": "default",
       "identify": "Colab Research Assistant",
       "description": "A QwenPaw agent configured for Google Colab tutorials, native information, customized abilities, and API testing.",
       "workspace_dir": str(WORKSPACE_DIR),
       "enabled": True,
   }
)
config["last_api"] = {"host": "127.0.0.1", "port": PORT}
config["show_tool_details"] = True
config["user_timezone"] = "Asia/Kolkata"
write_json(config_path, config)

agent_dir = WORKING_DIR / "brokers" / "default"
agent_dir.mkdir(dad and mom=True, exist_ok=True)
agent_path = agent_dir / "agent.json"
agent = read_json(agent_path, {})
agent.replace(
   {
       "id": "default",
       "identify": "Colab Research Assistant",
       "description": "Advanced QwenPaw tutorial agent for Colab: file-aware, skill-aware, API-testable, and guarded.",
       "language": "en",
       "workspace_dir": str(WORKSPACE_DIR),
       "enabled": True,
       "channels": {
           "console": {
               "enabled": True
           }
       },
       "operating": {
           "max_iters": 30,
           "llm_retry_enabled": True,
           "stream_output": True
       },
       "safety": {
           "tool_guard": True,
           "file_guard": True,
           "skill_scanner": True,
           "skill_scan_mode": "warn"
       },
       "tool_filter": {
           "enabled": False,
           "permit": [],
           "deny": []
       },
       "reminiscence": {
           "enabled": True
       }
   }
)
if chosen:
   provider_dir = SECRET_DIR / "suppliers" / "builtin"
   provider_dir.mkdir(dad and mom=True, exist_ok=True)
   provider_payload = {
       "id": chosen["provider_id"],
       "identify": chosen["name"],
       "base_url": chosen["base_url"],
       "api_key": chosen["api_key"],
       "chat_model": chosen["chat_model"],
       "fashions": [],
       "extra_models": [
           {
               "id": selected["model"],
               "identify": chosen["model"],
               "supports_image": None,
               "supports_video": None,
               "supports_multimodal": None,
               "is_free": False,
               "max_tokens": int(os.environ.get("QWENPAW_MAX_TOKENS", "2048")),
               "max_input_length": int(os.environ.get("QWENPAW_MAX_INPUT_LENGTH", "131072")),
               "generate_kwargs": {
                   "temperature": float(os.environ.get("QWENPAW_TEMPERATURE", "0.2")),
                   "max_tokens": int(os.environ.get("QWENPAW_MAX_TOKENS", "2048")),
               },
           }
       ],
       "api_key_prefix": chosen["prefix"],
       "is_local": False,
       "freeze_url": True,
       "require_api_key": True,
       "is_custom": False,
       "support_model_discovery": False,
       "support_connection_check": False,
       "generate_kwargs": {
           "temperature": float(os.environ.get("QWENPAW_TEMPERATURE", "0.2")),
           "max_tokens": int(os.environ.get("QWENPAW_MAX_TOKENS", "2048")),
       },
       "custom_headers": {},
       "auth_mode": "api_key",
       "meta": {},
   }
   write_json(provider_dir / f"{chosen['provider_id']}.json", provider_payload)
   write_json(
       SECRET_DIR / "suppliers" / "active_model.json",
       {"provider_id": chosen["provider_id"], "mannequin": chosen["model"]},
   )
   agent["active_model"] = {"provider_id": chosen["provider_id"], "mannequin": chosen["model"]}
   print(f"Configured mannequin supplier: {chosen['name']} / {chosen['model']}")
else:
   print(
       "No mannequin key discovered. The internet app will nonetheless launch, however chat requires a configured mannequin.n"
       "Add one Colab secret or surroundings variable reminiscent of OPENAI_API_KEY, OPENROUTER_API_KEY, "
       "DASHSCOPE_API_KEY, DEEPSEEK_API_KEY, GEMINI_API_KEY, or GOOGLE_API_KEY, then rerun."
   )
write_json(agent_path, agent)

skill_dir = WORKSPACE_DIR / "abilities" / "research_brief"
skill_dir.mkdir(dad and mom=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
   """---
identify: research_brief
description: Create rigorous analysis briefs from consumer questions, native notes, uploaded information, and out there instruments.
---
# Research Brief Skill
Use this ability when the consumer asks for analysis, product evaluation, market mapping, technical due diligence, paper evaluation, repo evaluation, or a choice memo.
## Procedure
1. Restate the consumer's goal in a single sentence.
2. Identify a very powerful entities, assumptions, and constraints.
3. Search out there native workspace information first.
4. Use instruments solely when they're related and allowed.
5. Separate verified details from inference.
6. Produce a compact transient with:
  - reply
  - proof
  - dangers or caveats
  - beneficial subsequent step
## Output Style
Prefer clear sections, quick paragraphs, and express uncertainty.
Do not invent citations, file contents, instructions, or outcomes.
""",
   encoding="utf-8",
)
demo_dir = WORKSPACE_DIR / "demo_knowledge"
demo_dir.mkdir(dad and mom=True, exist_ok=True)
(demo_dir / "qwenpaw_colab_notes.md").write_text(
   f"""# QwenPaw Colab Demo Notes
Created: {datetime.now().isoformat(timespec="seconds")}
This workspace is ready by a Google Colab tutorial.
The tutorial demonstrates:
- QwenPaw set up and initialization
- supplier auto-configuration from Colab secrets and techniques or surroundings variables
- authenticated Console launch
- customized workspace ability creation
- native workspace data information
- streaming REST API calls
- optionally available public tunnel publicity
Recommended first immediate within the Console:
"Read my workspace notes and clarify what this QwenPaw Colab setup can do. Then use the research_brief ability model to suggest three superior experiments."
""",
   encoding="utf-8",
)
(WORKSPACE_DIR / "README_COLAB_TUTORIAL.md").write_text(
   """# QwenPaw Advanced Colab Workspace
This workspace is deliberately small however structured like a actual assistant workspace.
Suggested experiments:
1. Ask QwenPaw to examine the demo_knowledge folder.
2. Ask it to use the research_brief ability model.
3. Use the REST API shopper on this pocket book for automated checks.
4. Add extra SKILL.md folders underneath workspace/abilities.
5. Add extra notes, CSVs, markdown information, or job briefs underneath workspace folders.
""",
   encoding="utf-8",
)
print("nWorkspace ready:")
print("Working dir:", WORKING_DIR)
print("Secret dir :", SECRET_DIR)
print("Workspace  :", WORKSPACE_DIR)
print("Skill file :", skill_dir / "SKILL.md")
run(qwenpaw_cmd("daemon", "model"), verify=False)
run(qwenpaw_cmd("fashions", "checklist"), verify=False)
run(qwenpaw_cmd("abilities", "checklist", "--agent-id", "default"), verify=False)

stop_previous_app()
APP_LOG.guardian.mkdir(dad and mom=True, exist_ok=True)
log_fh = APP_LOG.open("w", encoding="utf-8")
app_proc = subprocess.Popen(
   qwenpaw_cmd("app", "--host", "0.0.0.0", "--port", str(PORT), "--log-level", os.environ["QWENPAW_LOG_LEVEL"]),
   stdout=log_fh,
   stderr=subprocess.STDOUT,
   env=os.environ.copy(),
)
PID_FILE.write_text(str(app_proc.pid), encoding="utf-8")
if not wait_for_port(PORT, seconds=120):
   print("nQwenPaw didn't open the port. Last log strains:")
   attempt:
       print(APP_LOG.read_text(encoding="utf-8")[-6000:])
   besides Exception as e:
       print("Could not learn log:", e)
   elevate RuntimeError("QwenPaw app failed to begin.")
print(f"nQwenPaw app is operating on http://127.0.0.1:{PORT}")
print("Username:", os.environ["QWENPAW_AUTH_USERNAME"])
print("Password:", os.environ["QWENPAW_AUTH_PASSWORD"])
print("App log:", APP_LOG)
attempt:
   from google.colab import output
   proxy_url = output.eval_js(f"google.colab.kernel.proxyPort({PORT})")
   print("nColab proxied Console URL:")
   print(proxy_url)
   attempt:
       output.serve_kernel_port_as_window(PORT)
   besides Exception:
       cross
besides Exception as e:
   print("nNot operating inside Google Colab proxy surroundings:", e)
def start_cloudflared_tunnel(port):
   system_bin = pathlib.Path("/usr/native/bin/cloudflared")
   local_bin = ROOT / "cloudflared"
   cloudflared = system_bin if system_bin.exists() else local_bin
   if not cloudflared.exists():
       url = "https://github.com/cloudflare/cloudflared/releases/newest/obtain/cloudflared-linux-amd64"
       goal = str(system_bin)
       rc, _ = run(f"wget -q {shlex.quote(url)} -O {shlex.quote(goal)} && chmod +x {shlex.quote(goal)}", verify=False)
       if rc != 0 or not system_bin.exists():
           goal = str(local_bin)
           rc, _ = run(f"wget -q {shlex.quote(url)} -O {shlex.quote(goal)} && chmod +x {shlex.quote(goal)}", verify=False)
       cloudflared = pathlib.Path(goal)
   if not cloudflared.exists():
       print("cloudflared tunnel unavailable. Use the Colab proxy URL above.")
       return None, None
   tunnel_log = LOG_DIR / "cloudflared.log"
   fh = tunnel_log.open("w", encoding="utf-8")
   proc = subprocess.Popen(
       [str(cloudflared), "tunnel", "--url", f"http://127.0.0.1:{port}", "--no-autoupdate"],
       stdout=fh,
       stderr=subprocess.STDOUT,
       textual content=True,
   )
   public_url = None
   begin = time.time()
   whereas time.time() - begin < 45:
       time.sleep(1)
       attempt:
           textual content = tunnel_log.read_text(encoding="utf-8", errors="ignore")
       besides Exception:
           textual content = ""
       for token in textual content.substitute("|", " ").break up():
           if token.startswith("https://") and "trycloudflare.com" in token:
               public_url = token.strip()
               break
       if public_url:
           break
   if public_url:
       print("nTemporary public tunnel URL:")
       print(public_url)
       print("Use the identical username/password printed above.")
   else:
       print("nCloudflare tunnel began however no URL was detected but.")
       print("Tunnel log:", tunnel_log)
   return proc, public_url
ENABLE_CLOUDFLARE_TUNNEL = os.environ.get("ENABLE_QWENPAW_TUNNEL", "1") == "1"
cloudflared_proc, public_url = (None, None)
if ENABLE_CLOUDFLARE_TUNNEL:
   cloudflared_proc, public_url = start_cloudflared_tunnel(PORT)

def qwenpaw_chat(message, session_id=None, user_id="colab-user", agent_id="default", timeout=180):
   session_id = session_id or f"colab-{uuid.uuid4().hex[:10]}"
   url = f"http://127.0.0.1:{PORT}/api/console/chat"
   headers = {
       "Content-Type": "software/json",
       "X-Agent-Id": agent_id,
   }
   payload = {
       "message": message,
       "session_id": session_id,
       "user_id": user_id,
   }
   print("nAPI request:")
   print(json.dumps({**payload, "message": message[:300]}, indent=2))
   with requests.put up(url, headers=headers, json=payload, stream=True, timeout=timeout) as response:
       print("HTTP standing:", response.status_code)
       if response.status_code >= 400:
           print(response.textual content[:4000])
           response.raise_for_status()
       last_text = ""
       final_text = ""
       raw_events_seen = 0
       for uncooked in response.iter_lines(decode_unicode=True):
           if not uncooked:
               proceed
           if uncooked.startswith("information:"):
               raw_events_seen += 1
               information = uncooked[len("data:"):].strip()
               if information == "[DONE]":
                   break
               attempt:
                   occasion = json.hundreds(information)
               besides Exception:
                   proceed
               candidate_texts = []
               def stroll(x):
                   if isinstance(x, dict):
                       for key, worth in x.gadgets():
                           if key in {"textual content", "content material", "message", "delta"} and isinstance(worth, str):
                               candidate_texts.append(worth)
                           else:
                               stroll(worth)
                   elif isinstance(x, checklist):
                       for merchandise in x:
                           stroll(merchandise)
               stroll(occasion)
               if candidate_texts:
                   textual content = candidate_texts[-1]
                   if textual content and len(textual content) >= len(final_text):
                       final_text = textual content
                       if textual content.startswith(last_text):
                           print(textual content[len(last_text):], finish="", flush=True)
                       else:
                           print("n" + textual content, finish="", flush=True)
                       last_text = textual content
       print("nnStreaming occasions seen:", raw_events_seen)
       return {"session_id": session_id, "textual content": final_text}
if chosen:
   attempt:
       consequence = qwenpaw_chat(
           "Read the native workspace notes if out there. Then clarify this Colab QwenPaw setup in 5 bullets and counsel two superior experiments.",
           session_id="qwenpaw-colab-demo",
       )
       print("nFinal session_id:", consequence["session_id"])
   besides Exception as e:
       print("nAPI demo failed. The Console should still work; examine the app log under.")
       print("Error:", repr(e))
       attempt:
           print(APP_LOG.read_text(encoding="utf-8")[-8000:])
       besides Exception:
           cross
else:
   print(
       "nSkipping API chat demo as a result of no mannequin supplier key was discovered.n"
       "Open the Console URL above, or add a Colab secret reminiscent of OPENAI_API_KEY / OPENROUTER_API_KEY / DASHSCOPE_API_KEY / DEEPSEEK_API_KEY / GEMINI_API_KEY and rerun."
   )
print("nSummary")
print("Console username:", os.environ["QWENPAW_AUTH_USERNAME"])
print("Console password:", os.environ["QWENPAW_AUTH_PASSWORD"])
print("Local URL:", f"http://127.0.0.1:{PORT}")
if public_url:
   print("Public tunnel:", public_url)
print("Workspace:", WORKSPACE_DIR)
print("Logs:", APP_LOG)
print("nTo cease the server manually, run:")
print(f"import os, sign; os.kill({app_proc.pid}, sign.SIGTERM)")