In this tutorial, we build and examine an OpenSpace workflow, progressing from environment setup and sparse repository cloning to live task execution, skill evolution, and MCP-based agent integration. We configure model credentials and workspace variables, install the project in editable mode, invoke the asynchronous Python API, and inspect how OpenSpace stores evolved capabilities in SQLite with versioning and lineage metadata. We also create a custom SKILL.md, connect host-agent skills, test warm-task reuse, launch the streamable HTTP MCP server, and analyze the showcase evolution database to understand how FIX, DERIVED, and CAPTURED skills support lower-cost, reusable agent behavior.
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.")
We verify the Python runtime, define the required API credentials, and configure the OpenSpace model and optional cloud access settings. We clone the repository with sparse checkout, install the package in editable mode, and confirm that the OpenSpace command-line tools are available. We then create the workspace and skill directories, write the environment configuration files, export the required variables, and detect whether live LLM execution is enabled.
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)
We initialize asynchronous execution in Google Colab and define a reusable function to submit tasks via the OpenSpace Python API. We run an initial payroll-generation task and inspect any skills that OpenSpace evolves during post-execution analysis. We also examine the SQLite database structure, display stored records, and verify that the skill registry and type definitions are accessible programmatically.
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,ratenAda,Eng,45,90nGrace,Eng,38,95nAlan,Math,50,80n")
asyncio.run(run_task(
f"Using the colab-csv-report skill, produce a markdown report for {df_csv}"))
We submit a related payroll task to observe how OpenSpace reuses or derives capabilities from previously generated skills. We create a custom SKILL.md that instructs the agent to analyze CSV files and produce structured Markdown reports. We then install the OpenSpace host skills, generate a demonstration dataset, and execute the custom capability through the same evolving agent workflow.
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()
We start the OpenSpace MCP server using the streamable HTTP transport and bind it to a local Colab endpoint. We probe the endpoint to confirm that the server process is running, even when a basic HTTP request returns an MCP-specific response status. We also generate an example MCP host configuration that external agents can use to access the OpenSpace workspace and skill directories.
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("""
══════════════════════════════════════════════════════════════════
🎓 TUTORIAL COMPLETE — what you now have:
• OpenSpace installed + configured in Colab
• Live task execution via the Python API (if key set)
• A custom SKILL.md your agent auto-discovers
• Host skills (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+ skill evolution DB
Next: run more related tasks and watch token usage drop as skills
FIX / DERIVE / CAPTURE themselves. Dashboard (needs Node ≥ 20):
openspace-dashboard --port 7788 + cd frontend && npm i && npm run dev
══════════════════════════════════════════════════════════════════
""")
We conditionally upload the custom skill to the OpenSpace cloud community when a valid cloud API key is available. We inspect the repository’s showcase SQLite database to study the stored skills, metadata, quality information, and complete evolution lineage. We finally aggregate skills by origin type and summarize the Colab environment, custom capability, MCP integration, and self-evolving workflow that we establish throughout the tutorial.
In conclusion, we established a practical OpenSpace environment that demonstrates how agent capabilities are executed, persisted, reused, and progressively improved. We worked directly with the Python API, local skill directories, SQLite-backed registries, MCP transports, and optional cloud skill-sharing commands, giving us visibility into both the user-facing workflow and the underlying skill-engine architecture. We also verified how related tasks can reuse previously evolved knowledge, how custom skills become discoverable at runtime, and how lineage records expose the development history of each capability, leaving us with a strong foundation for building self-evolving, cost-efficient agent systems in Colab.
Check out the Full Code here. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
Sana Hassan, a consulting intern at Marktechpost and dual-degree student at IIT Madras, is passionate about applying technology and AI to address real-world challenges. With a keen interest in solving practical problems, he brings a fresh perspective to the intersection of AI and real-life solutions.
