Context7 Integration Skill

Expert integration of Context7 for document ingestion, semantic search, and role-scoped context retrieval in ERP applications.

Quick Reference

Project Structure

backend/
├── app/
│   ├── services/
│   │   └── context7_client.py    # Core Context7 client
│   ├── api/
│   │   └── knowledge/
│   │       └── routes.py         # Knowledge API endpoints
│   └── schemas/
│       └── knowledge.py          # Pydantic schemas
frontend/
├── hooks/
│   └── useContext7Search.ts     # Search hook
└── components/
    └── knowledge/
        └── ContextSearch.tsx    # Search component
docs/
├── policies/                     # Source documents
├── faq/                          # FAQ documents
└── procedures/                   # Procedure documents

Context7 Client

Core Client Class

# backend/app/services/context7_client.py
import os
from typing import Optional
from pydantic import BaseModel
from enum import Enum
from datetime import datetime


class DocumentType(str, Enum):
    MARKDOWN = "markdown"
    PDF = "pdf"
    HTML = "html"
    TEXT = "text"


class DocumentMetadata(BaseModel):
    """Metadata for context documents."""
    title: str
    description: Optional[str] = None
    # Role-based access
    allowed_roles: list[str] = []  # Empty = all roles
    # Organization scope
    school_id: Optional[str] = None
    # Content categorization
    module: str  # e.g., "fees", "attendance", "policies"
    category: Optional[str] = None  # e.g., "faq", "procedure", "policy"
    # Language
    language: str = "en"
    # Versioning
    version: str = "1.0"
    effective_date: Optional[datetime] = None
    expiry_date: Optional[datetime] = None
    # Source tracking
    source_file: Optional[str] = None
    source_url: Optional[str] = None


class ContextDocument(BaseModel):
    """Document in Context7."""
    id: str
    content: str
    metadata: DocumentMetadata
    chunk_id: Optional[str] = None
    similarity_score: Optional[float] = None


class Context7Client:
    """Client for Context7 knowledge store."""

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        default_namespace: str = "default",
    ):
        self.api_key = api_key or os.getenv("CONTEXT7_API_KEY")
        self.base_url = base_url or os.getenv("CONTEXT7_API_URL", "https://api.context7.com")
        self.default_namespace = default_namespace
        self._session = None

    def _get_session(self):
        """Get or create requests session."""
        if not self._session:
            import requests
            self._session = requests.Session()
            if self.api_key:
                self._session.headers.update({"Authorization": f"Bearer {self.api_key}"})
        return self._session

    def _request(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> dict:
        """Make API request to Context7."""
        session = self._get_session()
        response = session.request(
            method,
            f"{self.base_url}{endpoint}",
            **kwargs
        )
        response.raise_for_status()
        return response.json()

    # === INGESTION ===

    def ingest_document(
        self,
        content: str,
        metadata: DocumentMetadata,
        namespace: Optional[str] = None,
        document_id: Optional[str] = None,
    ) -> dict:
        """
        Ingest a single document into Context7.

        Args:
            content: Document content (markdown, HTML, or text)
            metadata: Document metadata with tags and access control
            namespace: Optional namespace (defaults to default_namespace)
            document_id: Optional document ID for idempotent updates

        Returns:
            Ingestion result with document ID
        """
        payload = {
            "content": content,
            "metadata": metadata.model_dump(),
            "document_id": document_id,
            "namespace": namespace or self.default_namespace,
        }

        return self._request("POST", "/v1/documents", json=payload)

    def ingest_batch(
        self,
        documents: list[tuple[str, DocumentMetadata]],
        namespace: Optional[str] = None,
        batch_size: int = 10,
    ) -> dict:
        """
        Ingest multiple documents in batches.

        Args:
            documents: List of (content, metadata) tuples
            namespace: Optional namespace
            batch_size: Number of documents per batch

        Returns:
            Batch ingestion result with success/failure counts
        """
        results = {"successful": 0, "failed": 0, "documents": []}
        namespace = namespace or self.default_namespace

        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            batch_payload = [
                {
                    "content": content,
                    "metadata": metadata.model_dump(),
                    "namespace": namespace,
                }
                for content, metadata in batch
            ]

            try:
                response = self._request(
                    "POST",
                    "/v1/documents/batch",
                    json={"documents": batch_payload}
                )
                results["successful"] += len(batch)
                results["documents"].extend(response.get("documents", []))
            except Exception as e:
                results["failed"] += len(batch)
                # Log failed batch for retry

        return results

    def ingest_from_file(
        self,
        file_path: str,
        metadata: DocumentMetadata,
        namespace: Optional[str] = None,
    ) -> dict:
        """
        Ingest a document from a file.

        Args:
            file_path: Path to file (markdown, PDF, or HTML)
            metadata: Document metadata
            namespace: Optional namespace

        Returns:
            Ingestion result
        """
        # Determine document type from extension
        ext = os.path.splitext(file_path)[1].lower()

        if ext == ".md":
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
            doc_type = DocumentType.MARKDOWN
        elif ext == ".pdf":
            content = self._extract_pdf_text(file_path)
            doc_type = DocumentType.PDF
        elif ext in [".html", ".htm"]:
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
            content = self._strip_html(content)
            doc_type = DocumentType.HTML
        else:
            # Default to text
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
            doc_type = DocumentType.TEXT

        # Update metadata with source file
        metadata.source_file = file_path

        return self.ingest_document(content, metadata, namespace)

    def _extract_pdf_text(self, file_path: str) -> str:
        """Extract text from PDF file."""
        try:
            import PyPDF2
            with open(file_path, "rb") as f:
                reader = PyPDF2.PdfReader(f)
                text = "\n".join(page.extract_text() for page in reader.pages)
            return text
        except ImportError:
            raise ImportError("PyPDF2 required for PDF ingestion: pip install PyPDF2")

    def _strip_html(self, html: str) -> str:
        """Strip HTML tags from content."""
        import re
        clean = re.compile("<.*?>")
        return re.sub(clean, "", html)

    # === RETRIEVAL ===

    def search(
        self,