สร้าง AI Agent วิวัฒนาการตัวเองด้วย OpenSpace: คู่มือพัฒนาทักษะผ่าน MCP และ Lineage ต้นทุนต่ำ

import os, sys, subprocess, sqlite3, json, textwrap, shutil, time, pathlib
ANTHROPIC_API_KEY = ""
OPENAI_API_KEY = ""
OPENSPACE_MODEL = "anthropic/claude-sonnet-4-5"
OPENSPACE_CLOUD_KEY = ""
assert sys.version_info >= (3, 12), (
f"OpenSpace needs Python 3.12+, Colab has {sys.version}. "
"Runtime > Change runtime type, or use a fallback py312 venv."
)
print("✅ Python:", sys.version.split()[0])
REPO_DIR = "/content/OpenSpace"
if not os.path.exists(REPO_DIR):
subprocess.run(
["git", "clone", "--filter=blob:none", "--sparse",
"https://github.com/HKUDS/OpenSpace.git", REPO_DIR],
check=True,
)
subprocess.run(
["git", "sparse-checkout", "set", "--no-cone", "/*", "!/assets/"],
cwd=REPO_DIR, check=True,
)
print("✅ Cloned to", REPO_DIR)
print("Top-level:", sorted(os.listdir(REPO_DIR)))
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-e", REPO_DIR], check=True)
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "nest_asyncio"], check=True)
for cli in ["openspace-mcp", "openspace-dashboard"]:
path = shutil.which(cli)
print(f"{'✅' if path else '⚠️ '} {cli}: {path}")
subprocess.run(["openspace-mcp", "--help"], check=False)
WORKSPACE = "/content/openspace_workspace"
SKILLS_DIR = "/content/my_agent_skills"
os.makedirs(WORKSPACE, exist_ok=True)
os.makedirs(SKILLS_DIR, exist_ok=True)
env_lines = [
f"OPENSPACE_MODEL={OPENSPACE_MODEL}",
f"OPENSPACE_WORKSPACE={WORKSPACE}",
f"OPENSPACE_HOST_SKILL_DIRS={SKILLS_DIR}",
]
if ANTHROPIC_API_KEY: env_lines.append(f"ANTHROPIC_API_KEY={ANTHROPIC_API_KEY}")
if OPENAI_API_KEY: env_lines.append(f"OPENAI_API_KEY={OPENAI_API_KEY}")
if OPENSPACE_CLOUD_KEY: env_lines.append(f"OPENSPACE_API_KEY={OPENSPACE_CLOUD_KEY}")
env_path = os.path.join(REPO_DIR, "openspace", ".env")
pathlib.Path(env_path).write_text("\n".join(env_lines) + "\n")
pathlib.Path("/content/.env").write_text("\n".join(env_lines) + "\n")
for line in env_lines:
k, _, v = line.partition("=")
os.environ[k] = v
print("✅ .env written:\n" + "\n".join(l.split("=")[0] + "=***" if "KEY" in l else l for l in env_lines))
HAS_LLM_KEY = bool(ANTHROPIC_API_KEY or OPENAI_API_KEY)
if not HAS_LLM_KEY:
print("⚠️ No LLM key set — Steps 4/6 (live execution) will be skipped.")

import asyncio, nest_asyncio
nest_asyncio.apply()
async def run_task(query: str):
from openspace import OpenSpace
async with OpenSpace() as cs:
result = await cs.execute(query)
print("── RESPONSE " + "─" * 50)
print(result["response"][:3000])
for skill in result.get("evolved_skills", []):
print(f" 🧬 Evolved: {skill['name']} (origin={skill['origin']})")
return result
if HAS_LLM_KEY:
result_1 = asyncio.run(run_task(
"Write a Python function that parses a CSV of employee hours and "
"computes weekly payroll with overtime (1.5x beyond 40h). Test it "
"on a small synthetic example and show the output."
))
else:
print("⏭️ Skipped live task (no key).")
def dump_db(db_path, max_rows=8):
if not os.path.exists(db_path):
print("No DB at", db_path); return
con = sqlite3.connect(db_path)
cur = con.cursor()
tables = [r[0] for r in cur.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
print(f"📀 {db_path}\n tables: {tables}")
for t in tables:
try:
cols = [c[1] for c in cur.execute(f"PRAGMA table_info({t})").fetchall()]
rows = cur.execute(f"SELECT * FROM {t} LIMIT {max_rows}").fetchall()
print(f"\n▶ {t} ({len(rows)} shown) cols={cols[:8]}{'…' if len(cols)>8 else ''}")
for r in rows:
print(" ", str(r)[:160])
except Exception as e:
print(f" (skip {t}: {e})")
con.close()
runtime_db = os.path.join(WORKSPACE, ".openspace", "openspace.db")
alt_db = os.path.join(REPO_DIR, ".openspace", "openspace.db")
dump_db(runtime_db if os.path.exists(runtime_db) else alt_db)
try:
from openspace.skill_engine.registry import SkillRegistry
from openspace.skill_engine import types as sk_types
print("\n✅ skill_engine importable:",
[n for n in dir(sk_types) if n[0].isupper()][:6])
except Exception as e:
print("ℹ️ registry import note:", e)

if HAS_LLM_KEY:
result_2 = asyncio.run(run_task(
"Extend the payroll logic: add a second CSV of tax withholding rates "
"per employee and produce net pay. Reuse any prior payroll skill."
))
dump_db(runtime_db if os.path.exists(runtime_db) else alt_db, max_rows=12)
else:
print("⏭️ Skipped warm-rerun demo (no key).")
custom = pathlib.Path(SKILLS_DIR) / "colab-csv-report"
custom.mkdir(parents=True, exist_ok=True)
(custom / "SKILL.md").write_text(textwrap.dedent("""\
---
name: colab-csv-report
description: Turn any CSV into a short markdown report with summary stats,
null counts, dtypes, and 3 key observations. Use pandas; never plot.
---
# colab-csv-report
1. Load the CSV with pandas (`on_bad_lines="skip"` fallback).
2. Emit: shape, dtypes table, describe(), null counts.
3. Write 3 bullet observations in plain markdown.
4. If parsing fails, retry with `sep=None, engine="python"`.
"""))
print("✅ Custom skill written:", custom / "SKILL.md")
for host_skill in ["delegate-task", "skill-discovery"]:
src = os.path.join(REPO_DIR, "openspace", "host_skills", host_skill)
dst = os.path.join(SKILLS_DIR, host_skill)
if os.path.isdir(src) and not os.path.isdir(dst):
shutil.copytree(src, dst)
print("✅ Host skills installed into agent dir:", sorted(os.listdir(SKILLS_DIR)))
if HAS_LLM_KEY:
df_csv = "/content/demo.csv"
pathlib.Path(df_csv).write_text(
"name,dept,hours,rate\nAda,Eng,45,90\nGrace,Eng,38,95\nAlan,Math,50,80\n")
asyncio.run(run_task(
f"Using the colab-csv-report skill, produce a markdown report for {df_csv}"))

mcp_proc = subprocess.Popen(
["openspace-mcp", "--transport", "streamable-http",
"--host", "127.0.0.1", "--port", "8081"],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
env={**os.environ},
)
time.sleep(8)
try:
import urllib.request
req = urllib.request.Request("http://127.0.0.1:8081/mcp", method="GET")
try:
urllib.request.urlopen(req, timeout=5)
print("✅ MCP streamable-HTTP endpoint is up at http://127.0.0.1:8081/mcp")
except urllib.error.HTTPError as e:
print(f"✅ MCP server alive (HTTP {e.code} on bare GET, expected for MCP)")
except Exception as e:
print("⚠️ MCP probe failed:", e)
print(json.dumps({
"mcpServers": {
"openspace": {
"command": "openspace-mcp",
"toolTimeout": 600,
"env": {
"OPENSPACE_HOST_SKILL_DIRS": SKILLS_DIR,
"OPENSPACE_WORKSPACE": WORKSPACE,
"OPENSPACE_API_KEY": "sk-xxx (optional, for cloud)",
},
}
}
}, indent=2))
mcp_proc.terminate()

if OPENSPACE_CLOUD_KEY:
subprocess.run(["openspace-upload-skill", str(custom)], check=False)
print("✅ Cloud CLI demo executed (upload).")
else:
print("ℹ️ Cloud skipped — set OPENSPACE_CLOUD_KEY to enable "
"openspace-upload-skill / openspace-download-skill.")
showcase_db = os.path.join(REPO_DIR, "showcase", ".openspace", "openspace.db")
dump_db(showcase_db, max_rows=10)
if os.path.exists(showcase_db):
con = sqlite3.connect(showcase_db)
cur = con.cursor()
for t in [r[0] for r in cur.execute(
"SELECT name FROM sqlite_master WHERE type='table'")]:
cols = [c[1].lower() for c in cur.execute(f"PRAGMA table_info({t})")]
if "origin" in cols:
print(f"\n📊 Evolution-mode breakdown in table '{t}':")
for origin, n in cur.execute(
f"SELECT origin, COUNT(*) FROM {t} GROUP BY origin ORDER BY 2 DESC"):
print(f" {origin:>10}: {n}")
con.close()
print("""
══════════════════════════════════════════════════════════════════
🎓 บทช่วยสอนเสร็จสมบูรณ์ — สิ่งที่คุณมีในตอนนี้:
• ติดตั้งและตั้งค่า OpenSpace ใน Colab เรียบร้อยแล้ว
• การรันงานจริงผ่าน Python API (หากมีการตั้งค่า Key)
• ทักษะใน SKILL.md ที่กำหนดเองซึ่ง Agent จะค้นพบโดยอัตโนมัติ
• ทักษะฝั่งโฮสต์ (delegate-task, skill-discovery) ที่จัดเตรียมไว้สำหรับ
Agent ที่รองรับ SKILL.md (Claude Code / Codex / OpenClaw / nanobot)
• เซิร์ฟเวอร์ MCP ที่เปิดใช้งานผ่าน Streamable HTTP
• การตรวจสอบพงศาวลี (Lineage) ทั้งหมดของฐานข้อมูลวิวัฒนาการที่มีมากกว่า 60 ทักษ
ขั้นตอนต่อไป: รันงานที่เกี่ยวข้องเพิ่มเติมและคอยดูยอดการใช้ Token ที่ลดลงเนื่องจากทักษะทำการ 
FIX / DERIVE / CAPTURE ตัวเอง Dashboard (ต้องการ Node ≥ 20):
openspace-dashboard --port 7788 + cd frontend && npm i && npm run dev
══════════════════════════════════════════════════════════════════
""")