Skills Catalog, Tool Bundles & Power Features

from pathlib import Path
import sys

ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from IPython.display import Markdown, display
from shipit_agent import Agent, DeepAgent, DEFAULT_SKILLS_PATH, FileSkillRegistry
from shipit_agent.builtins import get_builtin_tool_map
from shipit_agent.stores import InMemoryMemoryStore
from shipit_agent.skills.tool_bundles import (
    SKILL_TOOL_BUNDLES,
    validate_tool_bundles,
)

registry = FileSkillRegistry(DEFAULT_SKILLS_PATH)
print(f"Catalog path: {DEFAULT_SKILLS_PATH}")
print(f"Total skills: {len(registry.list())}")

all_skills = registry.list()

print(f"{'ID':35} {'Category':20} Name")
print("-" * 85)
for skill in all_skills:
    print(
        f"{skill.id:35} {(skill.category or 'uncategorized'):20} {skill.display_name or skill.name}"
    )

for query in ["database", "web scrape", "security", "marketing", "devops"]:
    matches = registry.search(query)
    print(f"\n'{query}' → {len(matches)} match(es)")
    for skill in matches[:3]:
        print(f"  - {skill.id}: {skill.description[:80]}")

print(f"Total skills with tool bundles: {len(SKILL_TOOL_BUNDLES)}\n")

for skill_id in [
    "web-scraper-pro",
    "code-workflow-assistant",
    "full-stack-developer",
    "security-engineer",
]:
    tools = SKILL_TOOL_BUNDLES.get(skill_id, [])
    print(f"{skill_id}:")
    print(f"  Tools ({len(tools)}): {', '.join(tools)}\n")

class StubLLM:
    def complete(self, **kwargs):
        pass


builtin_names = set(get_builtin_tool_map(llm=StubLLM()).keys())
print(f"Built-in tools: {len(builtin_names)}")

errors = validate_tool_bundles(builtin_names)
print(f"Validation: {'All valid' if not errors else errors}")

from examples.run_multi_tool_agent import build_llm_from_env

llm = build_llm_from_env("bedrock")
memory = InMemoryMemoryStore()

agent = Agent.with_builtins(
    llm=llm,
    name="skills-memory-agent",
    project_root="/tmp",
    skills=["database-architect", "code-workflow-assistant"],
    auto_use_skills=True,
    skill_match_limit=2,
    memory_store=memory,
    max_iterations=12,
)

print("Skills:", [s.id for s in agent.skills])
print(f"max_iterations: {agent.max_iterations}")
print(f"memory: {type(memory).__name__}")

result = agent.run(
    """
    We are adding tenant-based billing alerts to an existing FastAPI service.

    Project context:
    - Backend: FastAPI + PostgreSQL 15
    - Current endpoint: GET /api/billing/alerts
    - Current performance: 4.8s p95
    - Schema areas: alerts, tenants, invoices, users
    - Constraint: no breaking API changes
    - Goal: reduce latency below 500ms

    Please:
    1. Summarize the likely problem areas
    2. Propose a concrete implementation plan
    3. Suggest database/index/query improvements
    4. Call out testing and rollout risks
    """
)

print(f"Used skills: {result.metadata['used_skills']}")
print(f"Skill tools: {result.metadata['used_skill_tools']}\n")
display(Markdown(result.output))

stream_agent = Agent.with_builtins(
    llm=llm,
    project_root="/tmp",
    skills=["devops-automation"],
    auto_use_skills=False,
    memory_store=InMemoryMemoryStore(),
    max_iterations=10,
)

print("Streaming events:\n")
final_output = ""
for event in stream_agent.stream(
    """
    Use the write_file tool to create these files in the project root:
    1. Dockerfile — multi-stage Python build, non-root user, uvicorn CMD
    2. docker-compose.yml — app + PostgreSQL + Redis services

    You MUST use the write_file tool to create each file. Do NOT just show the code.
    After creating the files, use glob_files to verify they exist.
    """
):
    if event.type == "step_started":
        iteration = event.payload.get("iteration", "?")
        print(f"  [step {iteration}] LLM thinking...")
    elif event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "tool_completed":
        output_preview = event.payload.get("output", "")[:80]
        print(f"  [done] {event.message} → {output_preview}...")
    elif event.type == "tool_failed":
        print(f"  [FAIL] {event.message}")
    elif event.type == "run_completed":
        final_output = event.payload.get("output", "")
        print("  [finished]")

print()
display(Markdown(final_output))

deep_stream = DeepAgent.with_builtins(
    llm=llm,
    project_root="/tmp/deep-stream",
    skills=["full-stack-developer"],
    auto_use_skills=False,
    memory_store=InMemoryMemoryStore(),
    verify=True,
    max_iterations=10,
)

print("DeepAgent streaming with verify=True:\n")
final_output = ""
for event in deep_stream.stream(
    """
    Use the write_file tool to create these files:
    1. cli.py — a Python CLI tool that:
    - Accepts a directory path as argument
    - Counts files by extension
    - Prints a summary table
    - Handles errors gracefully
    2. README.md — explains usage and examples.

    You MUST use the write_file tool to create each file on disk. Do NOT just show code.
    """
):
    if event.type == "step_started":
        print(f"  [step {event.payload.get('iteration', '?')}]")
    elif event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "tool_completed":
        print(f"  [done] {event.message}")
    elif event.type == "run_completed":
        final_output = event.payload.get("output", "")
        verification = event.payload.get("verification", None)
        print("  [finished]")
        if verification:
            print(f"  [verify] {verification}")

print()
display(Markdown(final_output))

chat = agent.chat_session(session_id="billing-debug-session")

# Turn 1: Provide context
r1 = chat.send(
    """
    I'm working on the billing-api service. The GET /api/billing/alerts endpoint
    joins alerts, tenants, invoices, and users tables. It's currently at 4.8s p95.
    PostgreSQL 15, FastAPI backend. Help me debug and fix this.
    """
)
print("=== Turn 1 ===")
display(Markdown(r1.output))

# Turn 2: Follow-up — agent remembers the context.
r2 = chat.send(
    "Show me the specific CREATE INDEX statements for the join path you identified."
)
print("=== Turn 2 (no re-asking) ===")
display(Markdown(r2.output))

# Turn 3: Another follow-up.
r3 = chat.send("How do I deploy these index changes safely with zero downtime?")
print("=== Turn 3 (full context retained) ===")
display(Markdown(r3.output))

stream_chat_agent = Agent.with_builtins(
    llm=llm,
    project_root="/tmp",
    skills=["full-stack-developer"],
    auto_use_skills=False,
    memory_store=InMemoryMemoryStore(),
    max_iterations=10,
)

chat = stream_chat_agent.chat_session(session_id="stream-chat-demo")

# Turn 1: Stream the response
print("=== Turn 1 (streaming) ===")
final_output = ""
for event in chat.stream(
    "Use the write_file tool to create a FastAPI project with a User model at /tmp/stream-chat-project. Write app/main.py and app/models.py. You MUST use write_file to create the actual files."
):
    if event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "tool_completed":
        print(f"  [done] {event.message}")
    elif event.type == "run_completed":
        final_output = event.payload.get("output", "")

display(
    Markdown(final_output[:500] + "..." if len(final_output) > 500 else final_output)
)

# Turn 2: Stream follow-up — agent remembers the project from turn 1.
print("=== Turn 2 (streaming follow-up) ===")
final_output = ""
for event in chat.stream(
    "Now add a Task model with CRUD endpoints. Tasks belong to users."
):
    if event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "tool_completed":
        print(f"  [done] {event.message}")
    elif event.type == "run_completed":
        final_output = event.payload.get("output", "")

display(
    Markdown(final_output[:500] + "..." if len(final_output) > 500 else final_output)
)

builder = Agent.with_builtins(
    llm=llm,
    project_root="/tmp/fullstack-demo",
    skills=["full-stack-developer"],
    auto_use_skills=False,
    memory_store=InMemoryMemoryStore(),
    max_iterations=15,
)

result = builder.run(
    """
    Create a FastAPI REST API project at /tmp/fullstack-demo with:

    1. Project structure:
       - app/main.py (FastAPI app with CORS)
       - app/models.py (SQLAlchemy models: User, Task)
       - app/routes/tasks.py (CRUD endpoints)
       - app/database.py (SQLite connection)
       - requirements.txt
       - README.md

    2. Task model: id, title, description, status, created_at, user_id
    3. Endpoints: GET/POST /tasks, GET/PUT/DELETE /tasks/{id}
    4. Proper error handling and status codes

    IMPORTANT: You MUST use the write_file tool to create each file on disk.
    Do NOT just show the code — actually write the files.
    After creating all files, use glob_files to list what was created.
    """
)

print(f"Skills: {result.metadata['used_skills']}")
print(f"Tools injected: {result.metadata['used_skill_tools']}\n")
display(Markdown(result.output))

scraper = Agent.with_builtins(
    llm=llm,
    project_root="/tmp/scrape-output",
    skills=["web-scraper-pro"],
    auto_use_skills=False,
    memory_store=InMemoryMemoryStore(),
    max_iterations=12,
)

result = scraper.run(
    """
    Research the top 5 Python web frameworks in 2024.
    For each framework, find: GitHub stars, latest version, key differentiator.

    Save results to /tmp/scrape-output/frameworks.json as structured JSON,
    then create a markdown summary in /tmp/scrape-output/summary.md.
    """
)

print(f"Skills: {result.metadata['used_skills']}")
print(f"Tools injected: {result.metadata['used_skill_tools']}\n")
display(Markdown(result.output))

deep_chat = DeepAgent.with_builtins(
    llm=llm,
    project_root="/tmp/deep-chat-demo",
    skills=["full-stack-developer", "database-architect"],
    auto_use_skills=True,
    memory_store=InMemoryMemoryStore(),
    max_iterations=10,
)

chat = deep_chat.chat(session_id="deep-iterative-build")

# Turn 1: Scaffold (streaming)
print("=== Turn 1: Scaffold ===")
t1_output = ""
for event in chat.stream(
    "Use the write_file tool to create a FastAPI app with a User model and auth endpoints at /tmp/deep-chat-demo. Write app/main.py, app/models.py, app/auth.py. You MUST use write_file to create the actual files on disk."
):
    if event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "run_completed":
        t1_output = event.payload.get("output", "")

display(Markdown(t1_output[:400] + "..."))

# Turn 2: Add features — agent remembers the project structure.
print("=== Turn 2: Add Task model ===")
t2_output = ""
for event in chat.stream(
    "Now add a Task model with CRUD endpoints. Tasks belong to users. Add indexes."
):
    if event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "run_completed":
        t2_output = event.payload.get("output", "")

display(Markdown(t2_output[:400] + "..."))

# Turn 3: Deploy config — agent still has full context.
print("=== Turn 3: Deploy config ===")
t3_output = ""
for event in chat.stream(
    "Create a Dockerfile and docker-compose.yml for this project."
):
    if event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "run_completed":
        t3_output = event.payload.get("output", "")

display(Markdown(t3_output[:400] + "..."))

added = agent.add_skill("security-engineer")
print(f"Added: {added.id}")
print("All skills:", [s.id for s in agent.skills])

print("\nSearch:")
for skill in agent.search_skills("devops")[:3]:
    print(f"  - {skill.id}")

all_skills = registry.list()
missing = [s.id for s in all_skills if s.id not in SKILL_TOOL_BUNDLES]
print(f"Skills with bundles: {len(all_skills) - len(missing)}/{len(all_skills)}")
print("Missing:" if missing else "All skills have tool bundles.", missing or "")