🏷️ Tags System¶

import httpx
from typing import List, Dict, Any, Optional

class MCPTagClient:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {token}"}

    def get_tags(
        self,
        entity_types: Optional[List[str]] = None,
        include_entities: bool = False
    ) -> List[Dict[str, Any]]:
        """Get all tags with optional filtering."""
        params = {}
        if entity_types:
            params["entity_types"] = ",".join(entity_types)
        if include_entities:
            params["include_entities"] = "true"

        response = httpx.get(
            f"{self.base_url}/tags",
            params=params,
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def get_entities_by_tag(
        self,
        tag: str,
        entity_types: Optional[List[str]] = None
    ) -> List[Dict[str, Any]]:
        """Get all entities with a specific tag."""
        params = {}
        if entity_types:
            params["entity_types"] = ",".join(entity_types)

        response = httpx.get(
            f"{self.base_url}/tags/{tag}/entities",
            params=params,
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def find_related_entities(
        self,
        entity_id: str,
        entity_type: str
    ) -> List[Dict[str, Any]]:
        """Find entities with similar tags."""
        # First, get the entity's tags
        entity = httpx.get(
            f"{self.base_url}/{entity_type}s/{entity_id}",
            headers=self.headers
        ).json()

        # Then find entities with those tags
        related = []
        for tag in entity.get("tags", []):
            entities = self.get_entities_by_tag(tag)
            related.extend([
                e for e in entities
                if e["id"] != entity_id
            ])

        # Deduplicate by ID
        seen = set()
        unique = []
        for e in related:
            if e["id"] not in seen:
                seen.add(e["id"])
                unique.append(e)

        return unique

# Usage
client = MCPTagClient(
    base_url="http://localhost:4444",
    token="your-jwt-token"
)

# Get all tags with statistics
tags = client.get_tags()

# Get all database tools
db_tools = client.get_entities_by_tag(
    "database",
    entity_types=["tools"]
)

# Find related entities
related = client.find_related_entities(
    entity_id="tool-123",
    entity_type="tool"
)

async def create_tag_based_server(tag: str, db: Session):
    """Create a virtual server with all tools having a specific tag."""

    # Get all tools with the tag
    tools = await tool_service.list_tools(db, tags=[tag])
    tool_ids = [tool.id for tool in tools]

    # Create virtual server
    server = await server_service.create_server(
        db,
        ServerCreate(
            name=f"{tag.title()} Tools Server",
            description=f"Virtual server for {tag} tools",
            tags=["virtual", tag],
            tool_ids=tool_ids
        )
    )

    return server

from typing import List, Set

class TagBasedAccessControl:
    """Control access to entities based on tags."""

    ROLE_PERMISSIONS = {
        "admin": {"*"},  # Access all tags
        "developer": {"dev", "test", "internal", "public"},
        "user": {"public"},
        "guest": {"public", "demo"}
    }

    @classmethod
    def filter_by_access(
        cls,
        entities: List[dict],
        user_role: str
    ) -> List[dict]:
        """Filter entities based on user's tag permissions."""
        allowed_tags = cls.ROLE_PERMISSIONS.get(user_role, {"public"})

        if "*" in allowed_tags:
            return entities

        filtered = []
        for entity in entities:
            entity_tags = set(entity.get("tags", []))
            if entity_tags & allowed_tags:  # Has at least one allowed tag
                filtered.append(entity)

        return filtered

async def discover_related(
    entity_type: str,
    entity_id: str,
    db: Session
) -> Dict[str, List]:
    """Find related entities based on shared tags."""

    # Get the source entity
    if entity_type == "tool":
        entity = await tool_service.get_tool(db, entity_id)
    elif entity_type == "resource":
        entity = await resource_service.get_resource(db, entity_id)
    # ... handle other types

    if not entity or not entity.tags:
        return {}

    # Find entities with overlapping tags
    related = {}
    for tag in entity.tags:
        entities = await tag_service.get_entities_by_tag(
            db,
            tag_name=tag
        )

        for e in entities:
            if e.id != entity_id:  # Exclude self
                if e.type not in related:
                    related[e.type] = []
                related[e.type].append(e)

    # Deduplicate and sort by relevance
    for entity_type in related:
        # Count tag overlaps
        overlap_counts = {}
        for e in related[entity_type]:
            overlap = len(set(e.tags) & set(entity.tags))
            overlap_counts[e.id] = overlap

        # Sort by overlap count
        related[entity_type] = sorted(
            related[entity_type],
            key=lambda e: overlap_counts[e.id],
            reverse=True
        )

    return related