Building an Internal AI Agent Marketplace: Self-Service Agent Discovery and Deployment

Why Enterprises Need an Agent Marketplace

As organizations adopt AI agents, a common pattern emerges. The data team builds a SQL agent. The support team builds a ticket triage agent. The legal team contracts a vendor for contract review. Within a year, there are fifteen agents across the company, and no one knows what exists, who owns each agent, or how much they cost.

An internal agent marketplace solves this discovery problem. It is a catalog where teams publish agents they have built, and other teams can browse, evaluate, and provision those agents for their own use. Think of it as an internal app store, but for AI agents.

Agent Catalog Schema

The catalog stores metadata about each agent: what it does, who owns it, what data it accesses, and what approvals are required before a new team can use it.

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from uuid import uuid4


class AgentStatus(str, Enum):
    DRAFT = "draft"
    PUBLISHED = "published"
    DEPRECATED = "deprecated"
    RETIRED = "retired"


class ApprovalRequirement(str, Enum):
    NONE = "none"
    MANAGER = "manager"
    DATA_STEWARD = "data_steward"
    SECURITY_REVIEW = "security_review"


@dataclass
class AgentCatalogEntry:
    catalog_id: str = field(default_factory=lambda: str(uuid4()))
    agent_id: str = ""
    name: str = ""
    short_description: str = ""
    long_description: str = ""
    category: str = ""
    owner_team: str = ""
    owner_email: str = ""
    status: AgentStatus = AgentStatus.DRAFT
    data_classification: str = "internal"
    approval_required: ApprovalRequirement = ApprovalRequirement.NONE
    supported_input_types: list[str] = field(default_factory=list)
    example_prompts: list[str] = field(default_factory=list)
    documentation_url: str = ""
    cost_per_request_usd: float = 0.0
    avg_latency_ms: int = 0
    monthly_active_users: int = 0
    satisfaction_score: float = 0.0
    tags: list[str] = field(default_factory=list)
    published_at: str | None = None
    created_at: str = field(
        default_factory=lambda: datetime.utcnow().isoformat()
    )


class AgentCatalog:
    def __init__(self, db_pool):
        self.db = db_pool

    async def search(
        self,
        query: str = "",
        category: str = "",
        status: AgentStatus = AgentStatus.PUBLISHED,
        page: int = 1,
        per_page: int = 20,
    ) -> dict:
        conditions = ["status = $1"]
        params: list = [status.value]
        idx = 2

        if query:
            conditions.append(
                f"(name ILIKE ${idx} OR short_description ILIKE ${idx} "
                f"OR tags::text ILIKE ${idx})"
            )
            params.append(f"%{query}%")
            idx += 1

        if category:
            conditions.append(f"category = ${idx}")
            params.append(category)
            idx += 1

        where = " AND ".join(conditions)
        offset = (page - 1) * per_page

        rows = await self.db.fetch(
            f"SELECT * FROM agent_catalog WHERE {where} "
            f"ORDER BY monthly_active_users DESC "
            f"LIMIT ${idx} OFFSET ${idx + 1}",
            *params, per_page, offset,
        )
        total = await self.db.fetchval(
            f"SELECT COUNT(*) FROM agent_catalog WHERE {where}",
            *params,
        )
        return {
            "agents": [dict(r) for r in rows],
            "total": total,
            "page": page,
        }

Self-Service Provisioning Workflow

When a team wants to use a published agent, they submit a provisioning request through the marketplace. The request flows through the approval chain defined in the catalog entry. Once approved, the platform automatically configures access: creates the team's role mapping, sets up usage quotas, and notifies the agent owner.

@dataclass
class ProvisioningRequest:
    request_id: str = field(default_factory=lambda: str(uuid4()))
    catalog_id: str = ""
    requesting_team: str = ""
    requesting_user: str = ""
    business_justification: str = ""
    estimated_monthly_usage: int = 0
    cost_center: str = ""
    status: str = "pending"
    approvals: list[dict] = field(default_factory=list)
    created_at: str = field(
        default_factory=lambda: datetime.utcnow().isoformat()
    )


class ProvisioningService:
    def __init__(self, catalog: AgentCatalog, db_pool, notifier):
        self.catalog = catalog
        self.db = db_pool
        self.notifier = notifier

    async def submit_request(
        self, request: ProvisioningRequest
    ) -> ProvisioningRequest:
        entry = await self.db.fetchrow(
            "SELECT * FROM agent_catalog WHERE catalog_id = $1",
            request.catalog_id,
        )
        if not entry:
            raise ValueError("Agent not found in catalog")

        approval_req = entry["approval_required"]

        if approval_req == "none":
            request.status = "approved"
            await self.provision_access(request, entry)
        else:
            request.status = "pending_approval"
            await self.notifier.request_approval(
                approver_type=approval_req,
                request=request,
                agent_name=entry["name"],
            )

        await self.save_request(request)
        return request

    async def provision_access(self, request, catalog_entry) -> None:
        await self.db.execute(
            """
            INSERT INTO agent_access (
                team, agent_id, cost_center,
                monthly_quota, granted_at
            ) VALUES ($1, $2, $3, $4, NOW())
            """,
            request.requesting_team, catalog_entry["agent_id"],
            request.cost_center, request.estimated_monthly_usage,
        )
        await self.notifier.send(
            to=request.requesting_user,
            subject=f"Access granted: {catalog_entry['name']}",
            body=f"Your team now has access to {catalog_entry['name']}.",
        )

Usage Tracking and Lifecycle Management

Every provisioned agent tracks usage per team. When usage drops to zero for 90 days, the system flags the provisioning for review. Agent owners can deprecate agents, which triggers a migration notification to all active teams. Retired agents are removed from the catalog but remain accessible for 30 days to allow migrations.

Publishing Guidelines and Quality Gates

Before an agent can be published to the marketplace, it must pass quality gates: documentation completeness, evaluation score above a threshold, security review for data access patterns, and cost estimate accuracy. This prevents the catalog from filling with low-quality or abandoned agents.

FAQ

How do you prevent the marketplace from becoming a graveyard of abandoned agents?

Implement health scores based on owner responsiveness, update frequency, user satisfaction, and incident history. Agents below a health threshold get a "needs attention" badge. If the owner does not respond within 30 days, the agent moves to "deprecated" status automatically. Quarterly reviews with agent owners keep the catalog clean and current.

Should internal agents be free or have chargebacks?

Chargebacks create accountability. When agents are free, teams provision everything and use nothing, cluttering the catalog with phantom adoption. Even a lightweight showback model — showing teams their usage costs without actually charging — changes behavior. Teams start questioning whether they need five agents or two.

How do you handle agents that access different data for different teams?

Use parameterized agent configurations. The base agent logic is shared, but each team's provisioning includes data access scopes. The support team's instance connects to the support ticket database, while the sales team's instance connects to the CRM. The marketplace handles this through provisioning templates that configure data sources per team.

---

#EnterpriseAI #Marketplace #SelfService #AgentCatalog #Provisioning #LifecycleManagement #AgenticAI #LearnAI #AIEngineering