Building a Prompt Registry: Centralized Prompt Storage and Retrieval for Teams

The Problem with Scattered Prompts

As AI adoption grows within an organization, prompts proliferate. The support team has prompts in a Notion doc. The engineering team has them in Python files. The product team has variations in a spreadsheet. Nobody knows which version is running in production, and duplicated effort is rampant.

A prompt registry solves this by providing a single source of truth — a centralized service where prompts are stored, versioned, tagged, and retrieved through a consistent API.

Data Model Design

The registry needs to track prompts, their versions, and metadata that enables discovery.

flowchart TD
    SPEC(["Task spec"])
    SYSTEM["System prompt<br/>role plus rules"]
    SHOTS["Few shot examples<br/>3 to 5"]
    VARS["Variable injection<br/>Jinja or f-string"]
    COT["Chain of thought<br/>or scratchpad"]
    CONSTR["Output constraint<br/>JSON schema"]
    LLM["LLM call"]
    EVAL["Offline eval<br/>LLM as judge plus regex"]
    GATE{"Score over<br/>threshold?"}
    COMMIT(["Promote to prod<br/>version pinned"])
    REVISE(["Revise prompt"])
    SPEC --> SYSTEM --> SHOTS --> VARS --> COT --> CONSTR --> LLM --> EVAL --> GATE
    GATE -->|Yes| COMMIT
    GATE -->|No| REVISE --> SYSTEM
    style LLM fill:#4f46e5,stroke:#4338ca,color:#fff
    style EVAL fill:#f59e0b,stroke:#d97706,color:#1f2937
    style COMMIT fill:#059669,stroke:#047857,color:#fff

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

class PromptStatus(str, Enum):
    DRAFT = "draft"
    REVIEW = "review"
    APPROVED = "approved"
    DEPRECATED = "deprecated"

@dataclass
class PromptVersion:
    version: int
    content: str
    author: str
    created_at: datetime
    change_description: str
    status: PromptStatus = PromptStatus.DRAFT
    metrics: dict = field(default_factory=dict)

@dataclass
class PromptEntry:
    id: str
    name: str
    description: str
    tags: list[str]
    team: str
    created_at: datetime
    updated_at: datetime
    versions: list[PromptVersion] = field(default_factory=list)
    active_version: int = 1

    @property
    def current(self) -> PromptVersion:
        for v in self.versions:
            if v.version == self.active_version:
                return v
        raise ValueError("No active version found")

Each prompt entry holds multiple versions. The active_version field points to whichever version is currently in use, allowing you to publish a new version without immediately activating it.

Registry Implementation

Build the core registry with storage, retrieval, and search capabilities.

import hashlib
import json
from pathlib import Path
from datetime import datetime, timezone

class PromptRegistry:
    """Centralized prompt storage and retrieval service."""

    def __init__(self, storage_path: str = "registry_data"):
        self.storage = Path(storage_path)
        self.storage.mkdir(exist_ok=True)
        self._index: dict[str, PromptEntry] = {}
        self._load_index()

    def _load_index(self):
        index_file = self.storage / "index.json"
        if index_file.exists():
            data = json.loads(index_file.read_text())
            for entry_data in data:
                entry = self._deserialize_entry(entry_data)
                self._index[entry.id] = entry

    def register(
        self, name: str, content: str, author: str,
        description: str = "", tags: list[str] = None,
        team: str = "default"
    ) -> PromptEntry:
        """Register a new prompt in the registry."""
        prompt_id = hashlib.sha256(
            f"{team}/{name}".encode()
        ).hexdigest()[:12]
        now = datetime.now(timezone.utc)
        version = PromptVersion(
            version=1, content=content, author=author,
            created_at=now, change_description="Initial version",
        )
        entry = PromptEntry(
            id=prompt_id, name=name, description=description,
            tags=tags or [], team=team,
            created_at=now, updated_at=now,
            versions=[version], active_version=1,
        )
        self._index[prompt_id] = entry
        self._persist()
        return entry

    def add_version(
        self, prompt_id: str, content: str, author: str,
        change_description: str, activate: bool = False
    ) -> PromptVersion:
        """Add a new version to an existing prompt."""
        entry = self._index[prompt_id]
        new_version_num = max(
            v.version for v in entry.versions
        ) + 1
        version = PromptVersion(
            version=new_version_num, content=content,
            author=author, created_at=datetime.now(timezone.utc),
            change_description=change_description,
        )
        entry.versions.append(version)
        if activate:
            entry.active_version = new_version_num
        entry.updated_at = datetime.now(timezone.utc)
        self._persist()
        return version

    def get(self, prompt_id: str, version: int = None) -> str:
        """Retrieve prompt content by ID and optional version."""
        entry = self._index[prompt_id]
        if version is None:
            return entry.current.content
        for v in entry.versions:
            if v.version == version:
                return v.content
        raise ValueError(f"Version {version} not found")

    def search(
        self, query: str = "", tags: list[str] = None,
        team: str = None
    ) -> list[PromptEntry]:
        """Search prompts by text query, tags, or team."""
        results = list(self._index.values())
        if query:
            query_lower = query.lower()
            results = [
                e for e in results
                if query_lower in e.name.lower()
                or query_lower in e.description.lower()
            ]
        if tags:
            tag_set = set(tags)
            results = [
                e for e in results
                if tag_set.intersection(set(e.tags))
            ]
        if team:
            results = [
                e for e in results if e.team == team
            ]
        return results

    def _persist(self):
        index_file = self.storage / "index.json"
        data = [
            self._serialize_entry(e)
            for e in self._index.values()
        ]
        index_file.write_text(json.dumps(data, default=str))

API Layer

Expose the registry through a FastAPI service that teams consume programmatically.

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI(title="Prompt Registry API")
registry = PromptRegistry()

class RegisterRequest(BaseModel):
    name: str
    content: str
    author: str
    description: str = ""
    tags: list[str] = []
    team: str = "default"

@app.post("/prompts")
def register_prompt(req: RegisterRequest):
    entry = registry.register(
        name=req.name, content=req.content,
        author=req.author, description=req.description,
        tags=req.tags, team=req.team,
    )
    return {"id": entry.id, "name": entry.name, "version": 1}

@app.get("/prompts/{prompt_id}")
def get_prompt(prompt_id: str, version: int = None):
    try:
        content = registry.get(prompt_id, version)
        return {"content": content}
    except KeyError:
        raise HTTPException(404, "Prompt not found")

@app.get("/prompts")
def search_prompts(
    q: str = "", tag: list[str] = None, team: str = None
):
    results = registry.search(query=q, tags=tag, team=team)
    return [
        {"id": r.id, "name": r.name, "tags": r.tags,
         "team": r.team, "active_version": r.active_version}
        for r in results
    ]

Access Control

Not every team should edit every prompt. Add role-based permissions.

class AccessControl:
    """Role-based access control for prompt registry."""

    ROLES = {
        "viewer": {"read", "search"},
        "editor": {"read", "search", "create", "update"},
        "admin": {"read", "search", "create", "update",
                  "delete", "activate"},
    }

    def __init__(self):
        self._grants: dict[str, dict[str, str]] = {}

    def grant(self, user: str, team: str, role: str):
        self._grants.setdefault(user, {})[team] = role

    def check(self, user: str, team: str, action: str) -> bool:
        role = self._grants.get(user, {}).get(team, "viewer")
        return action in self.ROLES.get(role, set())

FAQ

How does a prompt registry differ from just using a config service?

A config service stores key-value pairs. A prompt registry adds prompt-specific features: multi-version tracking, approval workflows, usage analytics, and search by tags or descriptions. These features are critical when managing hundreds of prompts across teams.

Should I use a database or file storage for the registry?

For small teams (under 50 prompts), file-based storage backed by Git works well. For larger organizations, use PostgreSQL for the metadata and index, with prompt content stored as text columns. This gives you fast search, transactional updates, and easy backups.

How do I migrate existing prompts into the registry?

Write a one-time migration script that scans your codebase for inline prompts (search for common patterns like system_prompt = or messages = [{"role": "system"). Extract each into the registry with metadata about where it was found, then replace the inline strings with registry client calls.

---

#PromptRegistry #APIDesign #PromptManagement #TeamCollaboration #AIInfrastructure #AgenticAI #LearnAI #AIEngineering