[Feature Request]: Dynamic Server Catalog via Rule, Regexp, Tags - or Embedding / LLM-Based Selection

[crivetimihai]

🧭 Epic

Title: Dynamic Virtual Server API (/dynamic)
Goal: Allow users to define and retrieve virtual MCP servers dynamically, using rules and optionally LLMs, instead of fixed entity IDs.
Why now: MCP Gateway supports static servers composed of explicitly-associated tools, resources, and prompts. But real-world use cases often demand flexible, dynamic groupings—based on tags, names, metadata, or higher-order logic (Embedding Search, Graph, or LLM). A new /dynamic API unlocks rule-driven and AI-driven server construction.

---

🧭 Type of Feature

• Enhancement to existing functionality
• New feature or capability
• Security Related (requires review)

---

🙋‍♂️ User Story 1

As a: gateway operator
I want: to define a virtual server that includes tools tagged "analytics"
So that: new analytics tools are automatically included

✅ Acceptance Criteria

Scenario: Rule matches tools by metadata
  Given a dynamic rule: type = "tool", filter = "$.tags[?(@ == 'analytics')]"
  When a tool is registered with tag = "analytics"
  Then it appears under GET /dynamic/<server_id>/tools

---

🙋‍♂️ User Story 2

As a: admin
I want: to group all prompts with name starting legal_ and not marked deprecated
So that: my dynamic server reflects compliance-related prompts only

✅ Acceptance Criteria

Scenario: Prompt filter with regex and status
  Given a rule: type = "prompt", filter = "name =~ '^legal_' && !deprecated"
  When matching prompts exist
  Then they are included in GET /dynamic/<server_id>/prompts

---

🙋‍♂️ User Story 3

As a: power user
I want: to preview which tools a dynamic server would expose
So that: I can validate rule correctness before activation

✅ Acceptance Criteria

Scenario: Preview mode
  Given I POST to /dynamic/preview with a list of rules
  Then I get a simulated catalog response with matching tools/prompts/resources

---

🙋‍♂️ User Story 4

As a: LLM builder
I want: to defer rule resolution to an LLM function
So that: server content can reflect dynamic business logic

✅ Acceptance Criteria

Scenario: LLM-powered filter
  Given a dynamic rule: type = "tool", mode = "llm", prompt = "Select tools related to finance or risk"
  When the LLM returns a set of names
  Then those tools appear in the dynamic server's catalog

---

🙋‍♂️ User Story 5

As a: API consumer
I want: to fetch /dynamic/<id>/tools or /resources or /prompts
So that: I can treat a dynamic server like any other catalog surface

---

🙋‍♂️ User Story 6: Taggable tools (Dependency – Required for Dynamic Servers)

As a: gateway admin
I want: to assign and query tags on tools (e.g. ["finance", "internal"])
So that: dynamic server rules can filter based on tag values

✅ Acceptance Criteria

Scenario: Tagging a tool
  Given I PATCH /tools/{id} with {"tags": ["finance", "internal"]}
  Then the tool includes a `tags` field
  And GET /tools supports ?tag=... filter

Scenario: Filtering in dynamic server rule
  Given a rule: type = "tool", filter = "$.tags[?(@ == 'finance')]"
  Then only tools with tag "finance" are matched

---

📐 Design Sketch

📁 New API group: /dynamic

Endpoint	Description
GET /dynamic	List defined dynamic servers
POST /dynamic	Create new dynamic server
GET /dynamic/{id}	Inspect rule definitions
POST /dynamic/{id}/preview	Dry-run the rule logic
GET /dynamic/{id}/tools	Live-evaluated tool catalog
GET /dynamic/{id}/prompts	Live-evaluated prompt catalog
GET /dynamic/{id}/resources	Live-evaluated resource catalog

🧩 Rule Definition Schema

{
  "rules": [
    {
      "type": "tool" | "resource" | "prompt",
      "mode": "rule" | "llm",
      "filter": "jsonpath or jq or python-expression",     // if mode = "rule"
      "llm_prompt": "string",                               // if mode = "llm"
      "llm_provider": "openai" | "anthropic" | "local",     // optional
      "include_federated": true                             // optional
    }
  ],
  "name": "RiskOps Dynamic Server",
  "description": "Everything related to legal, risk, compliance"
}

🔄 Evaluation Modes

• mode: rule: uses JMESPath/JSONPath or simplified JQ expressions
• mode: llm: invokes system-wide LLM to resolve selection logic
• include_federated: allows federated tools/prompts/resources to be matched

📦 Runtime

• Catalogs are evaluated at request time (lazy), or pre-cached with TTL if configured
• LLM results can be cached per llm_prompt to avoid repeated calls

---

🔄 Alternatives Considered

• Requiring manual updates to static server.associated_* lists → burdensome and error-prone
• Storing dynamic results as static data → risks stale catalogs
• External logic layer → less discoverable and harder to reason about

---

📓 Additional Context

• LLM integration should reuse existing ClientSession auth and timeouts
• Dynamic server rules should be stored in the DB (new dynamic_servers table)
• Dry-run support (preview) is essential to safe usage
• Must sandbox any filter expressions (e.g. via jq or jsonpath_ng)

---

🧭 Tasks

Area	Task	Notes
Schema	[ ] Add DynamicRule Pydantic model	Support rule and LLM modes
	[ ] Add DynamicServerCreate, DynamicServerRead, etc.
API	[ ] Add CRUD endpoints under /dynamic	Use FastAPI
	[ ] Add preview and catalog expansion endpoints
Logic	[ ] Implement rule evaluator using jsonpath_ng, fallback jq	Configurable
	[ ] Add LLM integration via tool call or streaming
Security	[ ] Validate expression parsing, reject unsafe evals
UI	[ ] Extend Admin UI to support rule-based dynamic servers	Preview + builder interface
Tests	[ ] Unit: rule evaluation, LLM mock evaluation
	[ ] Integration: dynamic server behaves like static catalog
Docs	[ ] Add dynamic-servers.md with examples
Metrics	[ ] Emit Prometheus events for evals, LLM hits

---

🔗 MCP Standards Check

• Does not affect wire-level MCP spec
• No backward-incompatibility for existing APIs
• Optional—defaults to static-only mode if unused

[gabe-l-hart]

Nice, I really like this direction! Depending on the implementation, this could also be extensible to additional cataloging mechanisms like some kind of tree-based clustering data structure. One question: would we ever want the virtual server to be generated on-the-fly or would we always imagine this as a separate atomic unit that gets created once, then used many times?

[crivetimihai]

I think we should support both modes:

• On-the-fly - like a k8s Service with selector: app=frontend or a Prometheus query up{job=~"api-.*"}; the virtual server re-evaluates every request. Rules can be simple regex/JSONPath filters or LLM prompts ("return tools related to fraud detection") that run each time. Need to find a tiny model that works for this. Could also be vector embedding or cosine similarity based.

• Materialized snapshot - like a fixed Endpoints IP list or a Grafana panel with hard-coded metrics; evaluate once, cache the result, and version it. Here the rule output-whether produced by regex, JSONPath, or a one-off LLM run-is stored as an immutable set until manually refreshed.

[gabe-l-hart]

oooh, I like that a lot! Basically a selector pattern that can be ephemeral or concrete

[manavgup]

Nice! I love this enhancement.

This would be a great mechanism to dynamically compose MCP servers based on metadata and AI logic, allowing dynamic grouping and management of tools, prompts, and resources.

1. Are you thinking to have support for scheduling dynamic server refresh intervals or event-driven updates beyond TTL caching?
2. What about support for multiple LLM providers with fallback strategies (if one of them fails / is not available / no credit)?
3. RBAC (ie, who can create, modify, or view dynamic servers and their rules)?

[crivetimihai]

1. Dynamic refresh sounds like a good idea. We already implement a time based HEALTHCHECK probe to check if the MCP servers are still working / tools get refreshed, so i think this can work the same way.
2. This also makes sense, when implementing the models, we can also add a retry mechanism (exponential backoff, etc). Considering leveraging litellm libraries for this, but to be discussed.
3. RBAC - yes, I believe this will need to align with per-user / team / global catalogs.

We are also considering adding models such as llama-guard, Granite Guardian, etc. to help with security.

[manavgup]

ok here is my suggestion for how we can implement this:

Phase 1: Database Schema Extensions

1.1 New Database Models

Add to mcpgateway/db.py:

class DynamicServer(Base):
    """ORM model for Dynamic Virtual Servers"""
    __tablename__ = "dynamic_servers"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    description: Mapped[Optional[str]]
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
    is_active: Mapped[bool] = mapped_column(default=True)
    
    # Relationship to rules
    rules: Mapped[List["DynamicRule"]] = relationship("DynamicRule", back_populates="dynamic_server", cascade="all, delete-orphan")

class DynamicRule(Base):
    """ORM model for Dynamic Server Rules"""
    __tablename__ = "dynamic_rules"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    dynamic_server_id: Mapped[int] = mapped_column(ForeignKey("dynamic_servers.id"))
    type: Mapped[str]  # "tool", "resource", "prompt"
    mode: Mapped[str]  # "rule", "llm"
    filter_expression: Mapped[Optional[str]]  # JSONPath/JQ expression for rule mode
    llm_prompt: Mapped[Optional[str]]  # LLM prompt for llm mode
    llm_provider: Mapped[Optional[str]]  # "openai", "anthropic", "local"
    include_federated: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
    
    # Relationship back to dynamic server
    dynamic_server: Mapped["DynamicServer"] = relationship("DynamicServer", back_populates="rules")

1.2 Add Tags Support to Existing Models

Extend Tool, Resource, Prompt models in mcpgateway/db.py:

# Add to Tool class:
tags: Mapped[Optional[List[str]]] = mapped_column(JSON, default_factory=list)

# Add to Resource class:
tags: Mapped[Optional[List[str]]] = mapped_column(JSON, default_factory=list)

# Add to Prompt class:
tags: Mapped[Optional[List[str]]] = mapped_column(JSON, default_factory=list)

Phase 2: Pydantic Schemas

2.1 Dynamic Server Schemas

Add to mcpgateway/schemas.py:

class DynamicRuleCreate(BaseModelWithConfig):
    """Schema for creating dynamic server rules"""
    type: Literal["tool", "resource", "prompt"]
    mode: Literal["rule", "llm"]
    filter_expression: Optional[str] = None  # for rule mode
    llm_prompt: Optional[str] = None  # for llm mode
    llm_provider: Optional[Literal["openai", "anthropic", "local"]] = None
    include_federated: bool = False

class DynamicRuleRead(BaseModelWithConfig):
    """Schema for reading dynamic server rules"""
    id: int
    type: str
    mode: str
    filter_expression: Optional[str]
    llm_prompt: Optional[str]
    llm_provider: Optional[str]
    include_federated: bool
    created_at: datetime

class DynamicServerCreate(BaseModelWithConfig):
    """Schema for creating dynamic servers"""
    name: str
    description: Optional[str] = None
    rules: List[DynamicRuleCreate]

class DynamicServerRead(BaseModelWithConfig):
    """Schema for reading dynamic servers"""
    id: int
    name: str
    description: Optional[str]
    created_at: datetime
    updated_at: datetime
    is_active: bool
    rules: List[DynamicRuleRead]

class DynamicServerUpdate(BaseModelWithConfig):
    """Schema for updating dynamic servers"""
    name: Optional[str] = None
    description: Optional[str] = None
    rules: Optional[List[DynamicRuleCreate]] = None

class DynamicPreviewRequest(BaseModelWithConfig):
    """Schema for previewing dynamic server content"""
    rules: List[DynamicRuleCreate]

2.2 Update Existing Schemas for Tags

Update ToolCreate, ToolUpdate, ToolRead schemas:

# Add to existing schemas
tags: Optional[List[str]] = Field(default_factory=list, description="Tags for categorization")

# Similar updates for ResourceCreate/Update/Read and PromptCreate/Update/Read

Phase 3: Rule Evaluation Engine

3.1 Rule Evaluator Service

Create mcpgateway/services/rule_evaluator.py:

# Create mcpgateway/services/rule_evaluator.py

class RuleEvaluator:
    """Service for evaluating dynamic server rules"""
    
    async def evaluate_rule(self, db: Session, rule: DynamicRule) -> List[int]:
        """Evaluate a single rule and return matching entity IDs"""
        
    async def evaluate_jsonpath_rule(self, db: Session, rule: DynamicRule) -> List[int]:
        """Evaluate JSONPath/JQ expression rules"""
        
    async def evaluate_llm_rule(self, db: Session, rule: DynamicRule) -> List[int]:
        """Evaluate LLM-based rules"""
        
    async def get_entities_by_type(self, db: Session, entity_type: str, include_federated: bool = False) -> List[Dict]:
        """Get all entities of a specific type for rule evaluation"""

3.2 LLM Integration Service

Create mcpgateway/services/llm_service.py:

class LLMService:
    """Service for LLM-based rule evaluation"""
    
    async def evaluate_selection(self, prompt: str, entities: List[Dict], provider: str = "openai") -> List[str]:
        """Use LLM to select entities based on prompt"""

Phase 4: Dynamic Server Service

4.1 Core Service Implementation

Create mcpgateway/services/dynamic_server_service.py:

class DynamicServerService:
    """Service for managing dynamic virtual servers"""
    
    async def create_dynamic_server(self, db: Session, server: DynamicServerCreate) -> DynamicServerRead:
        """Create a new dynamic server with rules"""
        
    async def get_dynamic_server(self, db: Session, server_id: int) -> DynamicServerRead:
        """Get dynamic server by ID"""
        
    async def list_dynamic_servers(self, db: Session) -> List[DynamicServerRead]:
        """List all dynamic servers"""
        
    async def update_dynamic_server(self, db: Session, server_id: int, update: DynamicServerUpdate) -> DynamicServerRead:
        """Update dynamic server"""
        
    async def delete_dynamic_server(self, db: Session, server_id: int) -> None:
        """Delete dynamic server"""
        
    async def preview_dynamic_server(self, db: Session, preview: DynamicPreviewRequest) -> Dict[str, List]:
        """Preview what a dynamic server would contain"""
        
    async def get_dynamic_tools(self, db: Session, server_id: int) -> List[ToolRead]:
        """Get tools for a dynamic server"""
        
    async def get_dynamic_resources(self, db: Session, server_id: int) -> List[ResourceRead]:
        """Get resources for a dynamic server"""
        
    async def get_dynamic_prompts(self, db: Session, server_id: int) -> List[PromptRead]:
        """Get prompts for a dynamic server"""

Phase 5: API Endpoints

5.1 Dynamic Server Router

Add to mcpgateway/main.py:

# Add to mcpgateway/main.py

dynamic_router = APIRouter(prefix="/dynamic", tags=["Dynamic Servers"])

@dynamic_router.get("", response_model=List[DynamicServerRead])
async def list_dynamic_servers(db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """List all dynamic servers"""

@dynamic_router.post("", response_model=DynamicServerRead, status_code=201)
async def create_dynamic_server(server: DynamicServerCreate, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Create a new dynamic server"""

@dynamic_router.get("/{server_id}", response_model=DynamicServerRead)
async def get_dynamic_server(server_id: int, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Get dynamic server by ID"""

@dynamic_router.put("/{server_id}", response_model=DynamicServerRead)
async def update_dynamic_server(server_id: int, update: DynamicServerUpdate, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Update dynamic server"""

@dynamic_router.delete("/{server_id}")
async def delete_dynamic_server(server_id: int, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Delete dynamic server"""

@dynamic_router.post("/preview")
async def preview_dynamic_server(preview: DynamicPreviewRequest, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Preview dynamic server content"""

@dynamic_router.get("/{server_id}/tools", response_model=List[ToolRead])
async def get_dynamic_server_tools(server_id: int, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Get tools for dynamic server"""

@dynamic_router.get("/{server_id}/resources", response_model=List[ResourceRead])
async def get_dynamic_server_resources(server_id: int, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Get resources for dynamic server"""

@dynamic_router.get("/{server_id}/prompts", response_model=List[PromptRead])
async def get_dynamic_server_prompts(server_id: int, db: Session = Depends(get_db), user: str = Depends(require_auth)):
    """Get prompts for dynamic server"""

5.2 Tags Support for Existing Endpoints

# Update existing tool/resource/prompt endpoints to support tag filtering
@tool_router.get("", response_model=List[ToolRead])
async def list_tools(
    tag: Optional[str] = None,  # Add tag filter
    include_inactive: bool = False,
    db: Session = Depends(get_db),
    user: str = Depends(require_auth)
):
    """List tools with optional tag filtering"""

Phase 6: Security and Validation

6.1 Expression Sandboxing

# Create mcpgateway/utils/expression_validator.py

class ExpressionValidator:
    """Validates and sandboxes filter expressions"""
    
    @staticmethod
    def validate_jsonpath(expression: str) -> bool:
        """Validate JSONPath expression safety"""
        
    @staticmethod
    def validate_jq(expression: str) -> bool:
        """Validate JQ expression safety"""
        
    @staticmethod
    def sanitize_expression(expression: str) -> str:
        """Sanitize expression to prevent code injection"""

6.2 LLM Integration Security

# Add rate limiting and input validation for LLM calls
# Implement caching for LLM results to avoid repeated calls
# Add timeout handling for LLM requests

Phase 7: Admin UI Extensions
7.1 Dynamic Server Management UI
Add dynamic server creation/editing forms
Rule builder interface with visual rule construction
Preview functionality to test rules before saving
Tag management interface for existing entities
7.2 Enhanced Entity Management
Add tag fields to tool/resource/prompt forms
Tag-based filtering in list views
Bulk tag operations
Phase 8: Testing Strategy

8.1 Unit Tests

# tests/unit/services/test_dynamic_server_service.py
# tests/unit/services/test_rule_evaluator.py
# tests/unit/utils/test_expression_validator.py

8.2 Integration Tests

# tests/integration/test_dynamic_server_api.py
# tests/integration/test_rule_evaluation.py
# tests/integration/test_llm_integration.py

Phase 9: Documentation
Phase 10: Metrics and Monitoring

[manavgup]

@crivetimihai - let me know what you think. This will be a couple of weeks of effort.