skills_get_body

Instructions

Implementation Reference

• skill_mcp/tools/get_skill_body.py:28-106 (handler)

  Core handler function `get_skill_body` — retrieves full instructions for a skill by ID, with optional version pinning, inline version suffix parsing, caching via TTLCache, tier3_manifest attachment, deprecation notices, and error handling.

  def get_skill_body(skill_id: str, version: Optional[str] = None) -> str:
      """Retrieve the full body (instructions + system prompt addition) for a skill.
  
      Call this after skills_find_relevant once you have decided which skill to
      activate. The body contains step-by-step instructions and any system-prompt
      text the agent should prepend for this skill.
  
      The response also includes tier3_manifest — a lightweight index of available
      reference files, scripts, and assets. Check it and load what you need:
        - references: call skills_get_reference(skill_id, filename)
        - scripts: call skills_run_script(skill_id, filename, input_data)
        - assets: call skills_get_asset(skill_id, filename)
  
      Args:
          skill_id: The skill_id string returned by skills_find_relevant.
                    May include an inline version suffix: "stripe-integration@1.2"
          version:  Optional version string (e.g. "1.2"). If provided and found,
                    the pinned version body is returned. If not found, the latest
                    version is returned with a version_note field.
  
      Returns:
          JSON string matching the SkillBody schema plus tier3_manifest field,
          or an error object if the skill_id is not found.
      """
      # Parse inline version suffix: "stripe-integration@1.2"
      if "@" in skill_id and version is None:
          skill_id, version = skill_id.rsplit("@", 1)
  
      cache_key = f"body|{skill_id}|{version or 'latest'}"
      cached = _body_cache.get(cache_key)
      if cached is not None:
          return cached
  
      version_note: Optional[str] = None
      actual_version: Optional[str] = None
  
      if version:
          body = qdrant_manager.get_body_versioned(skill_id, version)
          if body is None:
              # Requested version not found — fall back to latest
              body = qdrant_manager.get_body(skill_id)
              if body is not None:
                  version_note = (
                      f"Requested version {version!r} not found in registry; "
                      f"returning latest version."
                  )
          else:
              actual_version = version
      else:
          body = qdrant_manager.get_body(skill_id)
  
      if body is None:
          return json.dumps(
              {"error": f"skill_id '{skill_id}' not found in body collection"}
          )
  
      body_dict = body.model_dump()
      body_dict["tier3_manifest"] = qdrant_manager.get_tier3_manifest(skill_id)
  
      if version_note:
          body_dict["version_note"] = version_note
      if actual_version:
          body_dict["pinned_version"] = actual_version
  
      # Attach deprecation notice if the skill is marked deprecated in frontmatter.
      try:
          fm_payload = qdrant_manager.get_frontmatter_payload(skill_id)
          if fm_payload and fm_payload.get("deprecated"):
              replaced_by = fm_payload.get("replaced_by", "")
              msg = "This skill is deprecated."
              if replaced_by:
                  msg += f" Use '{replaced_by}' instead."
              body_dict["deprecation_notice"] = msg
      except Exception:
          pass
  
      result = json.dumps(body_dict, indent=2)
      _body_cache.set(cache_key, result)
      return result

• skill_mcp/server.py:123-143 (registration)

  MCP tool registration for 'skills_get_body' using @mcp.tool decorator. The decorator registers the tool name and description, and the function `_skills_get_body` delegates to `get_skill_body`.

  @mcp.tool(
      name="skills_get_body",
      description=(
          "STEP 2 — Load full skill instructions. Call after skills_find_relevant "
          "once you have identified the best-matching skill_id.\n\n"
          "Returns three fields:\n"
          "  • instructions         — expert step-by-step guidance; read and follow these\n"
          "  • system_prompt_addition — optional context to add to your persona (may be empty)\n"
          "  • tier3_manifest       — lists available references, scripts, and assets by filename\n\n"
          "After loading: apply the instructions. "
          "If tier3_manifest lists files that the instructions explicitly reference, "
          "fetch them with skills_get_reference, skills_run_script, or skills_get_asset. "
          "Most tasks are fully served by the instructions alone — do not load Tier 3 speculatively.\n\n"
          "Version pinning: pass version='1.2' to pin to a specific skill version, or use the "
          "inline form skill_id='stripe-integration@1.2'. If the requested version is not found, "
          "the latest version is returned with a version_note explaining the fallback. "
          "Deprecated skills include a deprecation_notice field naming the replacement."
      ),
  )
  def _skills_get_body(skill_id: str, version: Optional[str] = None) -> str:
      return get_skill_body(skill_id=skill_id, version=version)

• skill_mcp/models/skill.py:35-41 (schema)

  SkillBody Pydantic model — defines the schema for the skill body returned by skills_get_body (skill_id, instructions, system_prompt_addition, example_usage).

  class SkillBody(BaseModel):
      """Full instructions — fetched on demand by skill_id, no vector needed."""
  
      skill_id: str
      instructions: str
      system_prompt_addition: str = ""
      example_usage: str = ""

• src/worker.py:479-546 (handler)

  Production Cloudflare Worker handler `_skills_get_body` — async version for the worker runtime, fetches from OpenSearch collections, attaches tier3_manifest by aggregating refs/scripts/assets, deprecation notices, and version notes.

  async def _skills_get_body(skill_id: str, version: str | None = None) -> str:
      QU, QK = _creds()
  
      # Parse inline version suffix: "stripe-integration@1.2"
      if "@" in skill_id and version is None:
          skill_id, version = skill_id.rsplit("@", 1)
  
      version_note: str | None = None
  
      if version:
          # Try pinned version point first
          ver_key = f"{skill_id}@{version}"
          payloads = await _scroll(QU, QK, C_BODY, _by_version_key(ver_key), 1)
          if not payloads:
              # Fall back to latest with a note
              payloads = await _scroll(QU, QK, C_BODY, _by_skill(skill_id), 1)
              if payloads:
                  version_note = (
                      f"Requested version {version!r} not found in registry; "
                      f"returning latest version."
                  )
      else:
          payloads = await _scroll(QU, QK, C_BODY, _by_skill(skill_id), 1)
  
      if not payloads:
          return json.dumps(
              {"error": f"skill_id '{skill_id}' not found in body collection"}
          )
      body = payloads[0]
  
      # Attach version_note if version was requested but not found
      if version_note:
          body = {**body, "version_note": version_note}
  
      # Attach deprecation notice from frontmatter if deprecated
      fm_payloads = await _scroll(QU, QK, C_FM, _by_skill(skill_id), 1)
      if fm_payloads and fm_payloads[0].get("deprecated"):
          replaced_by = fm_payloads[0].get("replaced_by", "")
          msg = "This skill is deprecated."
          if replaced_by:
              msg += f" Use '{replaced_by}' instead."
          body = {**body, "deprecation_notice": msg}
      refs = sorted(
          p.get("filename", "")
          for p in await _scroll(QU, QK, C_REFS, _by_skill(skill_id))
          if p.get("filename")
      )
      scripts = sorted(
          p.get("filename", "")
          for p in await _scroll(QU, QK, C_SCRIPTS, _by_skill(skill_id))
          if p.get("filename")
      )
      assets = sorted(
          p.get("filename", "")
          for p in await _scroll(QU, QK, C_ASSETS, _by_skill(skill_id))
          if p.get("filename")
      )
      return json.dumps(
          {
              **body,
              "tier3_manifest": {
                  "references": refs,
                  "scripts": scripts,
                  "assets": assets,
              },
          },
          indent=2,
      )

• src/worker.py:739-743 (registration)

  Worker dispatcher routing: when tool name is 'skills_get_body', calls the async `_skills_get_body` handler with skill_id and optional version.

  elif name == "skills_get_body":
      return await _skills_get_body(
          skill_id=str(args.get("skill_id", "")),
          version=args.get("version") or None,
      )