Write a custom detector

Quayside ships a few detectors out of the box (NullDetector, RegexDetector, PresidioDetector). Anything beyond those is a custom adapter you write.

The full pattern is also documented inline at services/api/src/quayside_api/detectors/README.md.

When to write one

• You need to call your own service (an internal LLM-judge, your homegrown PII scanner, a banned-words list backed by Redis)
• You need to inspect content with logic that doesn’t fit RegexDetector’s flat-pattern shape
• You need to plug a vendor that Quayside doesn’t ship a stub for

The Protocol

from quayside_api.common.envelope import Direction, Effect, RequestContext, Verdict
from quayside_api.detectors.base import WindowedStreamingDetector


class AcmeBrandGuard(WindowedStreamingDetector):
    """Flag mentions of our competitor's product."""

    name = "acme_brand_guard"

    def __init__(self, blocked_terms: list[str]) -> None:
        self._terms = [t.lower() for t in blocked_terms]

    async def inspect(
        self,
        content: str,
        *,
        direction: Direction,
        context: RequestContext,
    ) -> Verdict:
        text = content.lower()
        hits = [t for t in self._terms if t in text]
        if not hits:
            return Verdict(detector=self.name, effect=Effect.ALLOW)
        return Verdict(
            detector=self.name,
            effect=Effect.FLAG,
            reason=f"mentioned: {', '.join(hits)}",
            matched=hits,
        )

Three rules:

1. Subclass WindowedStreamingDetector to get a free inspect_stream() that accumulates incoming chunks and re-inspects every N characters.
2. Don’t translate failures yourself. If your detector errors or times out, just raise — the proxy applies the policy’s fail_mode / on_failure rules.
3. Pick a unique name. It must match what your policies reference in stages[].detectors[] and detectors[].

Calling an external service

For a detector that hits an HTTP service, use httpx:

import httpx

class MyServiceDetector(WindowedStreamingDetector):
    name = "my_service"

    def __init__(self, endpoint: str, timeout_s: float = 5.0) -> None:
        self._endpoint = endpoint.rstrip("/")
        self._timeout_s = timeout_s

    async def inspect(self, content, *, direction, context):
        async with httpx.AsyncClient(timeout=self._timeout_s) as client:
            r = await client.post(
                f"{self._endpoint}/score",
                json={"text": content},
            )
            r.raise_for_status()
            score = float(r.json()["score"])
        return Verdict(
            detector=self.name,
            effect=Effect.BLOCK if score >= 0.85 else Effect.ALLOW,
            score=score,
        )

PresidioDetector in the source is a complete worked example.

Calling an LLM judge

For something subjective (“is this off-topic?”), use a cheap model as a judge:

class LLMJudgeDetector(WindowedStreamingDetector):
    name = "topic_judge"

    def __init__(self, anthropic_api_key: str, prompt: str) -> None:
        import anthropic
        self._client = anthropic.AsyncAnthropic(api_key=anthropic_api_key)
        self._prompt = prompt

    async def inspect(self, content, *, direction, context):
        resp = await self._client.messages.create(
            model="claude-3-5-haiku-latest",
            max_tokens=10,
            messages=[{
                "role": "user",
                "content": f"{self._prompt}\n\nText: {content}\n\nAnswer ON or OFF only.",
            }],
        )
        is_off = "OFF" in resp.content[0].text
        return Verdict(
            detector=self.name,
            effect=Effect.FLAG if is_off else Effect.ALLOW,
            reason=resp.content[0].text.strip(),
        )

This sits inside your quayside service. The judge call is part of the cascade latency budget — keep timeouts realistic.

Registering it

Quayside’s composition root is services/api/src/quayside_api/app.py. The function _wire_services builds the LocalProxyService with a list of detector instances:

proxy = LocalProxyService(
    registry=registry,
    policy=policy,
    audit=audit,
    tokens=tokens,
    upstream=upstream,
    detectors=[
        NullDetector(),
        RegexDetector([...]),
        PresidioDetector(endpoint=settings.presidio_url, ...),
        AcmeBrandGuard(blocked_terms=settings.acme_brand_terms),  # ← your adapter
    ],
    shadow=shadow,
)

For v0.1, custom detectors live in your fork or your deployment overlay. A factory-registry pattern that lets customers plug adapters from a config file is on the roadmap.

Referencing it from a policy

Once registered in the composition root with name = "acme_brand_guard":

stages:
  - name: brand-check
    direction: request
    detectors: [acme_brand_guard]

detectors:
  acme_brand_guard:
    enabled: true
    weight: 1.0
    thresholds: { flag: 0.5, block: 0.85 }

Effect semantics — what to emit

Want to…	Emit
Let it through silently	Effect.ALLOW
Let it through, flag for review	Effect.FLAG (carries a reason)
Refuse the request	Effect.BLOCK (carries a reason)
Modify the content (v2)	Effect.MODIFY — currently logged but not applied
Pause for human approval (v2)	Effect.APPROVE — currently not wired

Block halts the cascade and refuses the request. Flag is non-decisive — the cascade continues.

Testing

Mirror the bundled detectors’ tests:

• Unit-test the adapter directly with a constructed RequestContext (no DB required)
• For HTTP-calling adapters, use respx to mock the upstream

See services/api/tests/unit/test_detectors_regex.py and services/api/tests/integration/test_detectors_presidio.py for the patterns.