Building Self-Evolving AI Agents with OpenSpace Using Skills, MCP, Lineage, and Low-Cost Reuse

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 wants Python 3.12+, Colab has {sys.model}. "
   "Runtime > Change runtime sort, or use a fallback py312 venv."
)
print(" Python:", sys.model.break up()[0])
REPO_DIR = "/content material/OpenSpace"
if not os.path.exists(REPO_DIR):
   subprocess.run(
       ["git", "clone", "--filter=blob:none", "--sparse",
        "https://github.com/HKUDS/OpenSpace.git", REPO_DIR],
       verify=True,
   )
   subprocess.run(
       ["git", "sparse-checkout", "set", "--no-cone", "/*", "!/assets/"],
       cwd=REPO_DIR, verify=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], verify=True)
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "nest_asyncio"], verify=True)
for cli in ["openspace-mcp", "openspace-dashboard"]:
   path = shutil.which(cli)
   print(f"{'' if path else ' '} {cli}: {path}")
subprocess.run(["openspace-mcp", "--help"], verify=False)
WORKSPACE = "/content material/openspace_workspace"
SKILLS_DIR = "/content material/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.be a part of(REPO_DIR, "openspace", ".env")
pathlib.Path(env_path).write_text("n".be a part of(env_lines) + "n")
pathlib.Path("/content material/.env").write_text("n".be a part of(env_lines) + "n")
for line in env_lines:
   okay, _, v = line.partition("=")
   os.environ[k] = v
print(" .env written:n" + "n".be a part of(l.break up("=")[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 (reside execution) might be skipped.")

import asyncio, nest_asyncio
nest_asyncio.apply()
async def run_task(question: str):
   from openspace import OpenSpace
   async with OpenSpace() as cs:
       consequence = await cs.execute(question)
       print("── RESPONSE " + "─" * 50)
       print(consequence["response"][:3000])
       for talent in consequence.get("evolved_skills", []):
           print(f"   Evolved: {talent['name']} (origin={talent['origin']})")
       return consequence
if HAS_LLM_KEY:
   result_1 = asyncio.run(run_task(
       "Write a Python operate that parses a CSV of worker hours and "
       "computes weekly payroll with extra time (1.5x past 40h). Test it "
       "on a small artificial instance and present the output."
   ))
else:
   print("  Skipped reside activity (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.join(db_path)
   cur = con.cursor()
   tables = [r[0] for r in cur.execute(
       "SELECT title FROM sqlite_master WHERE sort='desk'").fetchall()]
   print(f" {db_path}n   tables: {tables}")
   for t in tables:
       attempt:
           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)} proven) cols={cols[:8]}{'…' if len(cols)>8 else ''}")
           for r in rows:
               print("   ", str(r)[:160])
       besides Exception as e:
           print(f"   (skip {t}: {e})")
   con.shut()
runtime_db = os.path.be a part of(WORKSPACE, ".openspace", "openspace.db")
alt_db     = os.path.be a part of(REPO_DIR, ".openspace", "openspace.db")
dump_db(runtime_db if os.path.exists(runtime_db) else alt_db)
attempt:
   from openspace.skill_engine.registry import SkillRegistry
   from openspace.skill_engine import varieties as sk_types
   print("n skill_engine importable:",
         [n for n in dir(sk_types) if n[0].isupper()][:6])
besides Exception as e:
   print(" registry import be aware:", e)

if HAS_LLM_KEY:
   result_2 = asyncio.run(run_task(
       "Extend the payroll logic: add a second CSV of tax withholding charges "
       "per worker and produce internet pay. Reuse any prior payroll talent."
   ))
   dump_db(runtime_db if os.path.exists(runtime_db) else alt_db, max_rows=12)
else:
   print("  Skipped warm-rerun demo (no key).")
customized = pathlib.Path(SKILLS_DIR) / "colab-csv-report"
customized.mkdir(dad and mom=True, exist_ok=True)
(customized / "SKILL.md").write_text(textwrap.dedent("""
   ---
   title: colab-csv-report
   description: Turn any CSV into a brief markdown report with abstract stats,
     null counts, dtypes, and 3 key observations. Use pandas; by no means plot.
   ---
   # colab-csv-report
   1. Load the CSV with pandas (`on_bad_lines="skip"` fallback).
   2. Emit: form, dtypes desk, describe(), null counts.
   3. Write 3 bullet observations in plain markdown.
   4. If parsing fails, retry with `sep=None, engine="python"`.
   """))
print(" Custom talent written:", customized / "SKILL.md")
for host_skill in ["delegate-task", "skill-discovery"]:
   src = os.path.be a part of(REPO_DIR, "openspace", "host_skills", host_skill)
   dst = os.path.be a part of(SKILLS_DIR, host_skill)
   if os.path.isdir(src) and not os.path.isdir(dst):
       shutil.copytree(src, dst)
print(" Host expertise put in into agent dir:", sorted(os.listdir(SKILLS_DIR)))
if HAS_LLM_KEY:
   df_csv = "/content material/demo.csv"
   pathlib.Path(df_csv).write_text(
       "title,dept,hours,ratenAda,Eng,45,90nGrace,Eng,38,95nAlan,Math,50,80n")
   asyncio.run(run_task(
       f"Using the colab-csv-report talent, 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, textual content=True,
   env={**os.environ},
)
time.sleep(8)
attempt:
   import urllib.request
   req = urllib.request.Request("http://127.0.0.1:8081/mcp", methodology="GET")
   attempt:
       urllib.request.urlopen(req, timeout=5)
       print(" MCP streamable-HTTP endpoint is up at http://127.0.0.1:8081/mcp")
   besides urllib.error.HTTPError as e:
       print(f" MCP server alive (HTTP {e.code} on naked GET, anticipated for MCP)")
besides 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 (optionally available, for cloud)",
           },
       }
   }
}, indent=2))
mcp_proc.terminate()

if OPENSPACE_CLOUD_KEY:
   subprocess.run(["openspace-upload-skill", str(custom)], verify=False)
   print(" Cloud CLI demo executed (add).")
else:
   print(" Cloud skipped — set OPENSPACE_CLOUD_KEY to allow "
         "openspace-upload-skill / openspace-download-skill.")
showcase_db = os.path.be a part of(REPO_DIR, "showcase", ".openspace", "openspace.db")
dump_db(showcase_db, max_rows=10)
if os.path.exists(showcase_db):
   con = sqlite3.join(showcase_db)
   cur = con.cursor()
   for t in [r[0] for r in cur.execute(
           "SELECT title FROM sqlite_master WHERE sort='desk'")]:
       cols = [c[1].decrease() for c in cur.execute(f"PRAGMA table_info({t})")]
       if "origin" in cols:
           print(f"n Evolution-mode breakdown in desk '{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.shut()
print("""
══════════════════════════════════════════════════════════════════
 TUTORIAL COMPLETE — what you now have:
 • OpenSpace put in + configured in Colab
 • Live activity execution by way of the Python API (if key set)
 • A customized SKILL.md your agent auto-discovers
 • Host expertise (delegate-task, skill-discovery) staged for any
   SKILL.md-capable agent (Claude Code / Codex / OpenClaw / nanobot)
 • An MCP server you booted over streamable HTTP
 • Full lineage inspection of a 60+ talent evolution DB
Next: run extra associated duties and watch token utilization drop as expertise
FIX / DERIVE / CAPTURE themselves. Dashboard (wants Node ≥ 20):
 openspace-dashboard --port 7788   +   cd frontend && npm i && npm run dev
══════════════════════════════════════════════════════════════════
""")