Building automated AI agents with OpenSpace using Skills, MCP, Lineage, and low-cost reuse
In this tutorial, we build and test a OpenSpace workflow, progressing from environment setup and sparse repository cloning to live workflow, skill evolution, and MCP-based agent integration. We configure model credentials and workspace variables, put the project in editable mode, invoke an asynchronous Python API, and explore how OpenSpace maintains dynamic SQLite versioning and row metadata. We also built a custom SKILL.md, connected the host’s agent skills, tested warm reuse, introduced a distributed HTTP MCP server, and analyzed the demo database to understand how FIX, DERIVED, and CAPTURED skills support low-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",
" 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 details, and configure the OpenSpace model and cloud access settings. We compile the repository with minimal output, install the package in editable mode, and ensure that the OpenSpace command line tools are available. We then create the workspace and capabilities directories, write the environment configuration files, issue the necessary exceptions, and determine if LLM live 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’re starting to sync to Google Colab and define a reusable function to run tasks through the OpenSpace Python API. We conduct initial monetization work and evaluate any capabilities from OpenSpace during post-execution analysis. We also test the SQLite database structure, display the database records, and ensure that the registration of skills and type definitions are programmatically accessible.
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 post a related payload to look at how OpenSpace reuses or derives capabilities from previously generated capabilities. We created a custom SKILL.md that instructs the agent to analyze CSV files and generate structured Markdown reports. We then install OpenSpace hosting capabilities, generate demonstration datasets, and deploy custom capabilities with the same dynamic 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(" method="GET")
try:
urllib.request.urlopen(req, timeout=5)
print("✅ MCP streamable-HTTP endpoint is up at
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 a streamed HTTP transport and bind it to the Colab repository. We probe the endpoint to verify that the server process is running, and that the underlying HTTP request returns an MCP-specific response status. We also generate an example configuration for MCP hosts that external agents can use to access the OpenSpace workspace and capabilities 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 a custom capability to the OpenSpace cloud community if a valid cloud API key is available. We examine the SQLite database to learn the stored capabilities, metadata, quality information, and the complete list of evolution. Finally we group the skills by source type and summarize the Colab environment, custom capabilities, MCP integration, and automated workflows that we develop throughout the course.
In conclusion, we have created a working OpenSpace environment that demonstrates how agent capabilities are implemented, persisted, reused, and incrementally improved. We worked directly with the Python API, local skill indexes, SQLite-based registrations, MCP migration, and optional cloud sharing commands, giving us visibility into both the user-facing workflow and the skill engine structure. We also verified how related functions can reuse previously modified information, how custom capabilities are available at runtime, and how inventory records reveal the development history of each capability, leaving us with a solid foundation for building automated, cost-effective Collab agent systems.
Check it out Full Code here. Also, feel free to follow us Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to Our newspaper. Wait! are you on telegram? now you can join us on telegram too.
Need to work with us on developing your GitHub Repo OR Hug Face Page OR Product Release OR Webinar etc.? Connect with us
Sana Hassan, a consulting intern at Marktechpost and a dual graduate student at IIT Madras, is passionate about using technology and AI to address real-world challenges. With a deep interest in solving real-world problems, he brings a fresh perspective to the intersection of AI and real-life solutions.
