feat: implement skills marketplace API with CRUD operations and gateway integration

backend/app/api/skills_marketplace.py

@@ -0,0 +1,313 @@
+"""Skills marketplace API for catalog management and gateway install actions."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from urllib.parse import unquote, urlparse
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlmodel import col
+
+from app.api.deps import require_org_admin
+from app.core.time import utcnow
+from app.db.session import get_session
+from app.models.gateway_installed_skills import GatewayInstalledSkill
+from app.models.gateways import Gateway
+from app.models.marketplace_skills import MarketplaceSkill
+from app.schemas.common import OkResponse
+from app.schemas.skills_marketplace import (
+    MarketplaceSkillActionResponse,
+    MarketplaceSkillCardRead,
+    MarketplaceSkillCreate,
+    MarketplaceSkillRead,
+)
+from app.services.openclaw.gateway_dispatch import GatewayDispatchService
+from app.services.openclaw.gateway_resolver import gateway_client_config, require_gateway_workspace_root
+from app.services.openclaw.gateway_rpc import OpenClawGatewayError
+from app.services.openclaw.shared import GatewayAgentIdentity
+from app.services.organizations import OrganizationContext
+
+if TYPE_CHECKING:
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
+router = APIRouter(prefix="/skills", tags=["skills"])
+SESSION_DEP = Depends(get_session)
+ORG_ADMIN_DEP = Depends(require_org_admin)
+GATEWAY_ID_QUERY = Query(...)
+
+
+def _skills_install_dir(workspace_root: str) -> str:
+    normalized = workspace_root.rstrip("/\\")
+    if not normalized:
+        return "skills"
+    return f"{normalized}/skills"
+
+
+def _infer_skill_name(source_url: str) -> str:
+    parsed = urlparse(source_url)
+    path = parsed.path.rstrip("/")
+    candidate = path.rsplit("/", maxsplit=1)[-1] if path else parsed.netloc
+    candidate = unquote(candidate).removesuffix(".git").replace("-", " ").replace("_", " ")
+    if candidate.strip():
+        return candidate.strip()
+    return "Skill"
+
+
+def _install_instruction(*, skill: MarketplaceSkill, gateway: Gateway) -> str:
+    install_dir = _skills_install_dir(gateway.workspace_root)
+    return (
+        "MISSION CONTROL SKILL INSTALL REQUEST\n"
+        f"Skill name: {skill.name}\n"
+        f"Skill source URL: {skill.source_url}\n"
+        f"Install destination: {install_dir}\n\n"
+        "Actions:\n"
+        "1. Ensure the install destination exists.\n"
+        "2. Install or update the skill from the source URL into the destination.\n"
+        "3. Verify the skill is discoverable by the runtime.\n"
+        "4. Reply with success or failure details."
+    )
+
+
+def _uninstall_instruction(*, skill: MarketplaceSkill, gateway: Gateway) -> str:
+    install_dir = _skills_install_dir(gateway.workspace_root)
+    return (
+        "MISSION CONTROL SKILL UNINSTALL REQUEST\n"
+        f"Skill name: {skill.name}\n"
+        f"Skill source URL: {skill.source_url}\n"
+        f"Install destination: {install_dir}\n\n"
+        "Actions:\n"
+        "1. Remove the skill assets previously installed from this source URL.\n"
+        "2. Ensure the skill is no longer discoverable by the runtime.\n"
+        "3. Reply with success or failure details."
+    )
+
+
+def _as_card(
+    *,
+    skill: MarketplaceSkill,
+    installation: GatewayInstalledSkill | None,
+) -> MarketplaceSkillCardRead:
+    return MarketplaceSkillCardRead(
+        id=skill.id,
+        organization_id=skill.organization_id,
+        name=skill.name,
+        description=skill.description,
+        source_url=skill.source_url,
+        created_at=skill.created_at,
+        updated_at=skill.updated_at,
+        installed=installation is not None,
+        installed_at=installation.created_at if installation is not None else None,
+    )
+
+
+async def _require_gateway_for_org(
+    *,
+    gateway_id: UUID,
+    session: AsyncSession,
+    ctx: OrganizationContext,
+) -> Gateway:
+    gateway = await Gateway.objects.by_id(gateway_id).first(session)
+    if gateway is None or gateway.organization_id != ctx.organization.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Gateway not found",
+        )
+    return gateway
+
+
+async def _require_marketplace_skill_for_org(
+    *,
+    skill_id: UUID,
+    session: AsyncSession,
+    ctx: OrganizationContext,
+) -> MarketplaceSkill:
+    skill = await MarketplaceSkill.objects.by_id(skill_id).first(session)
+    if skill is None or skill.organization_id != ctx.organization.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Marketplace skill not found",
+        )
+    return skill
+
+
+async def _dispatch_gateway_instruction(
+    *,
+    session: AsyncSession,
+    gateway: Gateway,
+    message: str,
+) -> None:
+    dispatch = GatewayDispatchService(session)
+    config = gateway_client_config(gateway)
+    session_key = GatewayAgentIdentity.session_key(gateway)
+    await dispatch.send_agent_message(
+        session_key=session_key,
+        config=config,
+        agent_name="Gateway Agent",
+        message=message,
+        deliver=True,
+    )
+
+
+@router.get("/marketplace", response_model=list[MarketplaceSkillCardRead])
+async def list_marketplace_skills(
+    gateway_id: UUID = GATEWAY_ID_QUERY,
+    session: AsyncSession = SESSION_DEP,
+    ctx: OrganizationContext = ORG_ADMIN_DEP,
+) -> list[MarketplaceSkillCardRead]:
+    """List marketplace cards for an org and annotate install state for a gateway."""
+    gateway = await _require_gateway_for_org(gateway_id=gateway_id, session=session, ctx=ctx)
+    skills = (
+        await MarketplaceSkill.objects.filter_by(organization_id=ctx.organization.id)
+        .order_by(col(MarketplaceSkill.created_at).desc())
+        .all(session)
+    )
+    installations = await GatewayInstalledSkill.objects.filter_by(gateway_id=gateway.id).all(session)
+    installed_by_skill_id = {record.skill_id: record for record in installations}
+    return [
+        _as_card(skill=skill, installation=installed_by_skill_id.get(skill.id))
+        for skill in skills
+    ]
+
+
+@router.post("/marketplace", response_model=MarketplaceSkillRead)
+async def create_marketplace_skill(
+    payload: MarketplaceSkillCreate,
+    session: AsyncSession = SESSION_DEP,
+    ctx: OrganizationContext = ORG_ADMIN_DEP,
+) -> MarketplaceSkill:
+    """Register a skill source URL in the organization's marketplace catalog."""
+    source_url = str(payload.source_url).strip()
+    existing = await MarketplaceSkill.objects.filter_by(
+        organization_id=ctx.organization.id,
+        source_url=source_url,
+    ).first(session)
+    if existing is not None:
+        changed = False
+        if payload.name and existing.name != payload.name:
+            existing.name = payload.name
+            changed = True
+        if payload.description is not None and existing.description != payload.description:
+            existing.description = payload.description
+            changed = True
+        if changed:
+            existing.updated_at = utcnow()
+            session.add(existing)
+            await session.commit()
+            await session.refresh(existing)
+        return existing
+
+    skill = MarketplaceSkill(
+        organization_id=ctx.organization.id,
+        source_url=source_url,
+        name=payload.name or _infer_skill_name(source_url),
+        description=payload.description,
+    )
+    session.add(skill)
+    await session.commit()
+    await session.refresh(skill)
+    return skill
+
+
+@router.delete("/marketplace/{skill_id}", response_model=OkResponse)
+async def delete_marketplace_skill(
+    skill_id: UUID,
+    session: AsyncSession = SESSION_DEP,
+    ctx: OrganizationContext = ORG_ADMIN_DEP,
+) -> OkResponse:
+    """Delete a marketplace catalog entry and any install records that reference it."""
+    skill = await _require_marketplace_skill_for_org(skill_id=skill_id, session=session, ctx=ctx)
+    installations = await GatewayInstalledSkill.objects.filter_by(skill_id=skill.id).all(session)
+    for installation in installations:
+        await session.delete(installation)
+    await session.delete(skill)
+    await session.commit()
+    return OkResponse()
+
+
+@router.post(
+    "/marketplace/{skill_id}/install",
+    response_model=MarketplaceSkillActionResponse,
+)
+async def install_marketplace_skill(
+    skill_id: UUID,
+    gateway_id: UUID = GATEWAY_ID_QUERY,
+    session: AsyncSession = SESSION_DEP,
+    ctx: OrganizationContext = ORG_ADMIN_DEP,
+) -> MarketplaceSkillActionResponse:
+    """Install a marketplace skill by dispatching instructions to the gateway agent."""
+    gateway = await _require_gateway_for_org(gateway_id=gateway_id, session=session, ctx=ctx)
+    require_gateway_workspace_root(gateway)
+    skill = await _require_marketplace_skill_for_org(skill_id=skill_id, session=session, ctx=ctx)
+    try:
+        await _dispatch_gateway_instruction(
+            session=session,
+            gateway=gateway,
+            message=_install_instruction(skill=skill, gateway=gateway),
+        )
+    except OpenClawGatewayError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=str(exc),
+        ) from exc
+
+    installation = await GatewayInstalledSkill.objects.filter_by(
+        gateway_id=gateway.id,
+        skill_id=skill.id,
+    ).first(session)
+    if installation is None:
+        session.add(
+            GatewayInstalledSkill(
+                gateway_id=gateway.id,
+                skill_id=skill.id,
+            ),
+        )
+    else:
+        installation.updated_at = utcnow()
+        session.add(installation)
+    await session.commit()
+    return MarketplaceSkillActionResponse(
+        skill_id=skill.id,
+        gateway_id=gateway.id,
+        installed=True,
+    )
+
+
+@router.post(
+    "/marketplace/{skill_id}/uninstall",
+    response_model=MarketplaceSkillActionResponse,
+)
+async def uninstall_marketplace_skill(
+    skill_id: UUID,
+    gateway_id: UUID = GATEWAY_ID_QUERY,
+    session: AsyncSession = SESSION_DEP,
+    ctx: OrganizationContext = ORG_ADMIN_DEP,
+) -> MarketplaceSkillActionResponse:
+    """Uninstall a marketplace skill by dispatching instructions to the gateway agent."""
+    gateway = await _require_gateway_for_org(gateway_id=gateway_id, session=session, ctx=ctx)
+    require_gateway_workspace_root(gateway)
+    skill = await _require_marketplace_skill_for_org(skill_id=skill_id, session=session, ctx=ctx)
+    try:
+        await _dispatch_gateway_instruction(
+            session=session,
+            gateway=gateway,
+            message=_uninstall_instruction(skill=skill, gateway=gateway),
+        )
+    except OpenClawGatewayError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=str(exc),
+        ) from exc
+
+    installation = await GatewayInstalledSkill.objects.filter_by(
+        gateway_id=gateway.id,
+        skill_id=skill.id,
+    ).first(session)
+    if installation is not None:
+        await session.delete(installation)
+        await session.commit()
+    return MarketplaceSkillActionResponse(
+        skill_id=skill.id,
+        gateway_id=gateway.id,
+        installed=False,
+    )

backend/app/main.py

@@ -25,6 +25,7 @@ from app.api.gateways import router as gateways_router
 from app.api.metrics import router as metrics_router
 from app.api.organizations import router as organizations_router
 from app.api.souls_directory import router as souls_directory_router
+from app.api.skills_marketplace import router as skills_marketplace_router
 from app.api.tags import router as tags_router
 from app.api.task_custom_fields import router as task_custom_fields_router
 from app.api.tasks import router as tasks_router
@@ -138,6 +139,7 @@ api_v1.include_router(gateways_router)
 api_v1.include_router(metrics_router)
 api_v1.include_router(organizations_router)
 api_v1.include_router(souls_directory_router)
+api_v1.include_router(skills_marketplace_router)
 api_v1.include_router(board_groups_router)
 api_v1.include_router(board_group_memory_router)
 api_v1.include_router(boards_router)

backend/app/models/__init__.py

@@ -12,6 +12,8 @@ from app.models.board_webhook_payloads import BoardWebhookPayload
 from app.models.board_webhooks import BoardWebhook
 from app.models.boards import Board
 from app.models.gateways import Gateway
+from app.models.gateway_installed_skills import GatewayInstalledSkill
+from app.models.marketplace_skills import MarketplaceSkill
 from app.models.organization_board_access import OrganizationBoardAccess
 from app.models.organization_invite_board_access import OrganizationInviteBoardAccess
 from app.models.organization_invites import OrganizationInvite
@@ -42,6 +44,8 @@ __all__ = [
     "BoardGroup",
     "Board",
     "Gateway",
+    "GatewayInstalledSkill",
+    "MarketplaceSkill",
     "Organization",
     "BoardTaskCustomField",
     "TaskCustomFieldDefinition",

backend/app/models/gateway_installed_skills.py

@@ -0,0 +1,33 @@
+"""Gateway-to-skill installation state records."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from sqlalchemy import UniqueConstraint
+from sqlmodel import Field
+
+from app.core.time import utcnow
+from app.models.base import QueryModel
+
+RUNTIME_ANNOTATION_TYPES = (datetime,)
+
+
+class GatewayInstalledSkill(QueryModel, table=True):
+    """Marks that a marketplace skill is installed for a specific gateway."""
+
+    __tablename__ = "gateway_installed_skills"  # pyright: ignore[reportAssignmentType]
+    __table_args__ = (
+        UniqueConstraint(
+            "gateway_id",
+            "skill_id",
+            name="uq_gateway_installed_skills_gateway_id_skill_id",
+        ),
+    )
+
+    id: UUID = Field(default_factory=uuid4, primary_key=True)
+    gateway_id: UUID = Field(foreign_key="gateways.id", index=True)
+    skill_id: UUID = Field(foreign_key="marketplace_skills.id", index=True)
+    created_at: datetime = Field(default_factory=utcnow)
+    updated_at: datetime = Field(default_factory=utcnow)

backend/app/models/marketplace_skills.py

@@ -0,0 +1,35 @@
+"""Organization-scoped skill catalog entries for the skills marketplace."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from sqlalchemy import UniqueConstraint
+from sqlmodel import Field
+
+from app.core.time import utcnow
+from app.models.tenancy import TenantScoped
+
+RUNTIME_ANNOTATION_TYPES = (datetime,)
+
+
+class MarketplaceSkill(TenantScoped, table=True):
+    """A marketplace skill entry that can be installed onto one or more gateways."""
+
+    __tablename__ = "marketplace_skills"  # pyright: ignore[reportAssignmentType]
+    __table_args__ = (
+        UniqueConstraint(
+            "organization_id",
+            "source_url",
+            name="uq_marketplace_skills_org_source_url",
+        ),
+    )
+
+    id: UUID = Field(default_factory=uuid4, primary_key=True)
+    organization_id: UUID = Field(foreign_key="organizations.id", index=True)
+    name: str
+    description: str | None = Field(default=None)
+    source_url: str
+    created_at: datetime = Field(default_factory=utcnow)
+    updated_at: datetime = Field(default_factory=utcnow)

backend/app/schemas/__init__.py

@@ -33,6 +33,12 @@ from app.schemas.organizations import (
     OrganizationMemberUpdate,
     OrganizationRead,
 )
+from app.schemas.skills_marketplace import (
+    MarketplaceSkillActionResponse,
+    MarketplaceSkillCardRead,
+    MarketplaceSkillCreate,
+    MarketplaceSkillRead,
+)
 from app.schemas.souls_directory import (
     SoulsDirectoryMarkdownResponse,
     SoulsDirectorySearchResponse,
@@ -83,6 +89,10 @@ __all__ = [
     "SoulsDirectoryMarkdownResponse",
     "SoulsDirectorySearchResponse",
     "SoulsDirectorySoulRef",
+    "MarketplaceSkillActionResponse",
+    "MarketplaceSkillCardRead",
+    "MarketplaceSkillCreate",
+    "MarketplaceSkillRead",
     "TagCreate",
     "TagRead",
     "TagRef",

backend/app/schemas/skills_marketplace.py

@@ -0,0 +1,49 @@
+"""Schemas for skills marketplace listing and install/uninstall actions."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import AnyHttpUrl
+from sqlmodel import SQLModel
+
+from app.schemas.common import NonEmptyStr
+
+RUNTIME_ANNOTATION_TYPES = (datetime, UUID, NonEmptyStr)
+
+
+class MarketplaceSkillCreate(SQLModel):
+    """Payload used to register a skill URL in the organization marketplace."""
+
+    source_url: AnyHttpUrl
+    name: NonEmptyStr | None = None
+    description: str | None = None
+
+
+class MarketplaceSkillRead(SQLModel):
+    """Serialized marketplace skill catalog record."""
+
+    id: UUID
+    organization_id: UUID
+    name: str
+    description: str | None = None
+    source_url: str
+    created_at: datetime
+    updated_at: datetime
+
+
+class MarketplaceSkillCardRead(MarketplaceSkillRead):
+    """Marketplace card payload with gateway-specific install state."""
+
+    installed: bool
+    installed_at: datetime | None = None
+
+
+class MarketplaceSkillActionResponse(SQLModel):
+    """Install/uninstall action response payload."""
+
+    ok: bool = True
+    skill_id: UUID
+    gateway_id: UUID
+    installed: bool

backend/migrations/versions/c9d7e9b6a4f2_add_skills_marketplace_tables.py

@@ -0,0 +1,99 @@
+"""add skills marketplace tables
+
+Revision ID: c9d7e9b6a4f2
+Revises: b6f4c7d9e1a2
+Create Date: 2026-02-13 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+import sqlmodel
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c9d7e9b6a4f2"
+down_revision = "b6f4c7d9e1a2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "marketplace_skills",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("description", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("source_url", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["organization_id"],
+            ["organizations.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "organization_id",
+            "source_url",
+            name="uq_marketplace_skills_org_source_url",
+        ),
+    )
+    op.create_index(
+        op.f("ix_marketplace_skills_organization_id"),
+        "marketplace_skills",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_table(
+        "gateway_installed_skills",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("gateway_id", sa.Uuid(), nullable=False),
+        sa.Column("skill_id", sa.Uuid(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["gateway_id"],
+            ["gateways.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["skill_id"],
+            ["marketplace_skills.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "gateway_id",
+            "skill_id",
+            name="uq_gateway_installed_skills_gateway_id_skill_id",
+        ),
+    )
+    op.create_index(
+        op.f("ix_gateway_installed_skills_gateway_id"),
+        "gateway_installed_skills",
+        ["gateway_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_gateway_installed_skills_skill_id"),
+        "gateway_installed_skills",
+        ["skill_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f("ix_gateway_installed_skills_skill_id"),
+        table_name="gateway_installed_skills",
+    )
+    op.drop_index(
+        op.f("ix_gateway_installed_skills_gateway_id"),
+        table_name="gateway_installed_skills",
+    )
+    op.drop_table("gateway_installed_skills")
+    op.drop_index(
+        op.f("ix_marketplace_skills_organization_id"),
+        table_name="marketplace_skills",
+    )
+    op.drop_table("marketplace_skills")

backend/tests/test_skills_marketplace_api.py

@@ -0,0 +1,221 @@
+# ruff: noqa: INP001
+"""Integration tests for skills marketplace APIs."""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from fastapi import APIRouter, FastAPI
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+from sqlmodel import SQLModel, col, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from app.api.deps import require_org_admin
+from app.api.skills_marketplace import router as skills_marketplace_router
+from app.db.session import get_session
+from app.models.gateway_installed_skills import GatewayInstalledSkill
+from app.models.gateways import Gateway
+from app.models.marketplace_skills import MarketplaceSkill
+from app.models.organization_members import OrganizationMember
+from app.models.organizations import Organization
+from app.services.organizations import OrganizationContext
+
+
+async def _make_engine() -> AsyncEngine:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.connect() as conn, conn.begin():
+        await conn.run_sync(SQLModel.metadata.create_all)
+    return engine
+
+
+def _build_test_app(
+    session_maker: async_sessionmaker[AsyncSession],
+    *,
+    organization: Organization,
+) -> FastAPI:
+    app = FastAPI()
+    api_v1 = APIRouter(prefix="/api/v1")
+    api_v1.include_router(skills_marketplace_router)
+    app.include_router(api_v1)
+
+    async def _override_get_session() -> AsyncSession:
+        async with session_maker() as session:
+            yield session
+
+    async def _override_require_org_admin() -> OrganizationContext:
+        return OrganizationContext(
+            organization=organization,
+            member=OrganizationMember(
+                organization_id=organization.id,
+                user_id=uuid4(),
+                role="owner",
+                all_boards_read=True,
+                all_boards_write=True,
+            ),
+        )
+
+    app.dependency_overrides[get_session] = _override_get_session
+    app.dependency_overrides[require_org_admin] = _override_require_org_admin
+    return app
+
+
+async def _seed_base(
+    session: AsyncSession,
+) -> tuple[Organization, Gateway]:
+    organization = Organization(id=uuid4(), name="Org One")
+    gateway = Gateway(
+        id=uuid4(),
+        organization_id=organization.id,
+        name="Gateway One",
+        url="https://gateway.example.local",
+        workspace_root="/workspace/openclaw",
+    )
+    session.add(organization)
+    session.add(gateway)
+    await session.commit()
+    return organization, gateway
+
+
+@pytest.mark.asyncio
+async def test_install_skill_dispatches_instruction_and_persists_installation(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = await _make_engine()
+    session_maker = async_sessionmaker(
+        engine,
+        class_=AsyncSession,
+        expire_on_commit=False,
+    )
+    try:
+        async with session_maker() as session:
+            organization, gateway = await _seed_base(session)
+            skill = MarketplaceSkill(
+                organization_id=organization.id,
+                name="Deploy Helper",
+                source_url="https://example.com/skills/deploy-helper.git",
+                description="Handles deploy workflow checks.",
+            )
+            session.add(skill)
+            await session.commit()
+            await session.refresh(skill)
+
+        app = _build_test_app(session_maker, organization=organization)
+        sent_messages: list[dict[str, str | bool]] = []
+
+        async def _fake_send_agent_message(
+            _self: object,
+            *,
+            session_key: str,
+            config: object,
+            agent_name: str,
+            message: str,
+            deliver: bool = False,
+        ) -> None:
+            del config
+            sent_messages.append(
+                {
+                    "session_key": session_key,
+                    "agent_name": agent_name,
+                    "message": message,
+                    "deliver": deliver,
+                },
+            )
+
+        monkeypatch.setattr(
+            "app.api.skills_marketplace.GatewayDispatchService.send_agent_message",
+            _fake_send_agent_message,
+        )
+
+        async with AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://testserver",
+        ) as client:
+            response = await client.post(
+                f"/api/v1/skills/marketplace/{skill.id}/install",
+                params={"gateway_id": str(gateway.id)},
+            )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["installed"] is True
+        assert body["gateway_id"] == str(gateway.id)
+        assert len(sent_messages) == 1
+        assert sent_messages[0]["agent_name"] == "Gateway Agent"
+        assert sent_messages[0]["deliver"] is True
+        assert sent_messages[0]["session_key"] == f"agent:mc-gateway-{gateway.id}:main"
+        message = str(sent_messages[0]["message"])
+        assert "SKILL INSTALL REQUEST" in message
+        assert str(skill.source_url) in message
+        assert "/workspace/openclaw/skills" in message
+
+        async with session_maker() as session:
+            installed_rows = (
+                await session.exec(
+                    select(GatewayInstalledSkill).where(
+                        col(GatewayInstalledSkill.gateway_id) == gateway.id,
+                        col(GatewayInstalledSkill.skill_id) == skill.id,
+                    ),
+                )
+            ).all()
+            assert len(installed_rows) == 1
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_list_marketplace_skills_marks_installed_cards() -> None:
+    engine = await _make_engine()
+    session_maker = async_sessionmaker(
+        engine,
+        class_=AsyncSession,
+        expire_on_commit=False,
+    )
+    try:
+        async with session_maker() as session:
+            organization, gateway = await _seed_base(session)
+            first = MarketplaceSkill(
+                organization_id=organization.id,
+                name="First Skill",
+                source_url="https://example.com/skills/first",
+            )
+            second = MarketplaceSkill(
+                organization_id=organization.id,
+                name="Second Skill",
+                source_url="https://example.com/skills/second",
+            )
+            session.add(first)
+            session.add(second)
+            await session.commit()
+            await session.refresh(first)
+            await session.refresh(second)
+
+            session.add(
+                GatewayInstalledSkill(
+                    gateway_id=gateway.id,
+                    skill_id=first.id,
+                ),
+            )
+            await session.commit()
+
+        app = _build_test_app(session_maker, organization=organization)
+        async with AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://testserver",
+        ) as client:
+            response = await client.get(
+                "/api/v1/skills/marketplace",
+                params={"gateway_id": str(gateway.id)},
+            )
+
+        assert response.status_code == 200
+        cards = response.json()
+        assert len(cards) == 2
+        cards_by_id = {item["id"]: item for item in cards}
+        assert cards_by_id[str(first.id)]["installed"] is True
+        assert cards_by_id[str(first.id)]["installed_at"] is not None
+        assert cards_by_id[str(second.id)]["installed"] is False
+        assert cards_by_id[str(second.id)]["installed_at"] is None
+    finally:
+        await engine.dispose()