> ## Documentation Index > Fetch the complete documentation index at: https://docs.agentguardian.io/llms.txt > Use this file to discover all available pages before exploring further. # Write a custom target adapter > Implement the TargetAdapter protocol to point AgentGuardian at any agent runtime — a Python callable, a hosted HTTP API, or a framework-native object. ## What this is A `TargetAdapter` is the seam between AgentGuardian and your agent. The swarm sends one user-turn via `await adapter.call(prompt, session=...)` and reads one assistant text reply back; everything else — orchestration, judging, scoring — runs on top. Four modes ship in `agent_guardian.adapters`: | Mode | Class | Use when | | ----------- | ----------------------------- | ------------------------------------------------------------------------------------------- | | `prompt` | `PromptAdapter` | You only have a system prompt, no agent yet (pre-deployment review). | | `code` | `CodeAdapter` | Your agent is an importable Python callable / class / dotted path. | | `http` | `HttpAdapter` | Your agent is a hosted HTTP/JSON endpoint (OpenAI-, Anthropic-, or generic-shape). | | `framework` | `FrameworkAdapter` subclasses | Your agent is a LangGraph / CrewAI / AutoGen / ADK / OpenAI Agents / Strands native object. | ## When to use this * The four built-in modes don't cover your transport (custom RPC, gRPC, websocket, in-process queue, on-device runtime). * You want stricter fingerprinting than `CodeAdapter`'s heuristics give you — e.g. you know the target has memory + PII and want to force tier T1 from the start. * You want to register a new HTTP provider shape without subclassing `HttpAdapter` (use `register_shape` instead — see [Custom HTTP shapes](#custom-http-shapes)). ## The protocol Every adapter inherits `TargetAdapter` (declared in [`src/agent_guardian/adapters/base.py`](https://github.com/glacien-technologies/agent-guardian/blob/main/src/agent_guardian/adapters/base.py)) and satisfies three contracts: 1. **Set `self._fingerprint` in `__init__`.** A `TargetFingerprint` declares the static attack surface — `mode`, `ref`, `has_tools`, `has_memory`, `touches_pii`, `is_multi_agent`, `declared_tools`, `declared_memory_keys`. The recon agent refines these signals on phase 1 of the swarm; the swarm tiering logic reads them via `TargetFingerprint.to_observed_surface()`. 2. **Implement `async def call(self, prompt, *, session=None) -> str`.** Send one user-turn, return one assistant text reply. `session` is an opaque string the swarm uses to thread parallel conversations — distinct ASI agents pass distinct session IDs so per-session histories never cross-contaminate. 3. **Optionally override `profile_evidence()` and `aclose()`.** White-box adapters return `ProfileEvidence(box="white", text=...)` so the profiler can read your target instead of interrogating it; otherwise the default black-box evidence forces a behavioural audit. ## Build a custom adapter The example below wraps a hypothetical gRPC chat service. It's a black-box adapter — the swarm can only call it, not read its source. ```python my_grpc_adapter.py expandable theme={null} from __future__ import annotations import grpc from agent_guardian.adapters.base import ( TargetAdapter, TargetFingerprint, ) # Generated by `python -m grpc_tools.protoc ...` from myproto import chat_pb2, chat_pb2_grpc class GrpcChatAdapter(TargetAdapter): """Wraps a gRPC-fronted chat agent as an AgentGuardian target.""" mode = "code" # one of: "prompt" | "code" | "http" | "framework" def __init__( self, endpoint: str, *, api_key: str, has_tools: bool = False, has_memory: bool = False, touches_pii: bool = False, ) -> None: super().__init__() self._channel = grpc.aio.secure_channel( endpoint, grpc.ssl_channel_credentials() ) self._stub = chat_pb2_grpc.ChatServiceStub(self._channel) self._api_key = api_key self._endpoint = endpoint # MUST be set before __init__ returns — TargetAdapter.fingerprint() # raises RuntimeError if it sees None. self._fingerprint = TargetFingerprint( mode="code", ref=endpoint, has_tools=has_tools, has_memory=has_memory, touches_pii=touches_pii, is_multi_agent=False, notes="Custom gRPC chat target (black-box).", ) async def call(self, prompt: str, *, session: str | None = None) -> str: request = chat_pb2.ChatRequest( prompt=prompt, session_id=session or "", api_key=self._api_key, ) response = await self._stub.Chat(request) return response.text # MUST return str async def aclose(self) -> None: await self._channel.close() ``` ### Use it from the CLI The CLI's `scan` subcommand accepts a dotted-path target — pass it a `module:attr` reference to a no-arg-constructible callable that returns your adapter: ```python my_target.py theme={null} from my_grpc_adapter import GrpcChatAdapter def target() -> GrpcChatAdapter: return GrpcChatAdapter( endpoint="chat.internal:8443", api_key="dev-token", has_tools=True, ) ``` ```bash theme={null} uv run agent-guardian scan my_target:target \ --model gemini:gemini-2.5-flash \ --mode fast \ --budget-usd 0.20 ``` `CodeAdapter` is the path the dotted-path resolver instantiates by default (see `cli.py::build_target_adapter`). If you need to bypass it entirely — e.g. inject your adapter into the swarm programmatically — drive the swarm via the `SwarmCommander` API instead of the CLI. ### Use it programmatically ```python run_scan.py expandable theme={null} import asyncio from agent_guardian.core.memory import SharedMemory from agent_guardian.swarm.commander import SwarmCommander, SwarmConfig from my_grpc_adapter import GrpcChatAdapter async def main() -> None: target = GrpcChatAdapter( endpoint="chat.internal:8443", api_key="dev-token", has_tools=True, ) memory = SharedMemory() commander = SwarmCommander.from_config( SwarmConfig(model="gemini:gemini-2.5-flash", mode="fast"), ) scan = await commander.run(target=target, memory=memory) print(f"AIVSS={scan.aivss} band={scan.band} findings={len(scan.findings)}") await target.aclose() asyncio.run(main()) ``` ## White-box adapters: expose source for profiling White-box adapters let the profiler read the target's implementation rather than interrogate it through prompts. Override `profile_evidence()`: ```python theme={null} from pathlib import Path from agent_guardian.adapters.base import ProfileEvidence def profile_evidence(self) -> ProfileEvidence: return ProfileEvidence( box="white", text=self._system_prompt, # whatever describes the agent source_root=Path(__file__).parent, # so future agentic passes can read deeper ) ``` The built-in `PromptAdapter` returns the system prompt verbatim; `CodeAdapter` calls `safe_source_and_root()` to extract the defining module's source (scoped to the entry-point's transitive closure if the module exceeds 80 000 chars). ## Custom HTTP shapes If your target is HTTP/JSON-shaped but doesn't match any of the six built-in shapes (`openai`, `anthropic`, `bedrock`, `vertex`, `agentcore`, `generic`), register a new shape rather than subclassing `HttpAdapter`. A shape is a pair of pure functions wrapped in an `HttpShape` dataclass. ```python register_my_shape.py expandable theme={null} from typing import Any from agent_guardian.adapters import HttpAdapter, HttpShape, register_shape def build_request( prompt: str, *, model: str | None, session: str | None, extra: dict[str, Any] | None = None, ) -> dict[str, Any]: return { "input": {"text": prompt}, "conversation_id": session or "", "model": model or "default", } def extract_response_text(response_json: dict[str, Any]) -> str: return response_json["output"]["message"] register_shape( HttpShape( name="my_company", build_request=build_request, extract_response_text=extract_response_text, auth_header_name="x-api-key", auth_header_format="{token}", ) ) # After registration, the adapter constructor accepts shape="my_company": adapter = HttpAdapter( "https://api.mycompany.com/v1/chat", shape="my_company", auth_headers={"x-api-key": "..."}, model="prod-2", ) ``` The shape registry lives in [`src/agent_guardian/adapters/http_shapes/base.py`](https://github.com/glacien-technologies/agent-guardian/blob/main/src/agent_guardian/adapters/http_shapes/base.py). `register_shape` raises `ValueError` on duplicate names so import-time registration is idempotent only across distinct module loads. ## Expected output A correctly wired custom adapter looks indistinguishable from a built-in one in the CLI banner — same tier auto-detection, same per-agent feed, same final summary line: ```text theme={null} AgentGuardian v1.1.0 · mode=fast · budget=$0.20 · seed=0 target : chat.internal:8443 tier : T1 (auto-detected — tools + memory + PII) swarm : 16 agents (10 ASI + 1 gap-fill + 5 OWASP-LLM) ✓ recon probes=fingerprint spend=$0.001 ✓ goal_hijack probes=9 findings=2 spend=$0.018 ✓ tool_abuse probes=8 findings=1 spend=$0.022 ... scan cli-7e2a9b6c1f08 done: AIVSS=64 band=WARNING tier=T1 findings=6 report=scan.json ``` If your fingerprint declared `has_tools=True` and recon didn't observe any tool calls during phase 1, the `tool-abuse-agent` short-circuits via `is_applicable(fingerprint)` and you'll see `probes=0 findings=0 not_tested=0` — that's the swarm correctly skipping a category that doesn't apply, not a bug. ## How to interpret a failed adapter wire-up | Symptom | Cause | Fix | | ------------------------------------------------------------------------ | ------------------------------------------------------------------ | --------------------------------------------------------------------------------------- | | `RuntimeError: ... did not set _fingerprint in __init__` | You forgot to assign `self._fingerprint = TargetFingerprint(...)`. | Set it before `__init__` returns. | | `TypeError: CodeAdapter cannot instantiate ... with no args` | Your dotted-path target needs constructor arguments. | Wrap it in a no-arg factory (`def target(): return MyAgent(api_key=...)`). | | `UserWarning: CodeAdapter target returned dict, coercing via str()` | Your `call()` returned a non-string. | Return `str` directly; `dict` / `BaseModel` coercions are noisy and inspection-hostile. | | `EgressRefused` log lines with `terminated_by=not_tested` on every agent | Your contract or RoE forbids the destinations the attacker probed. | Widen the contract `egress_allowlist` or accept that the category is out-of-scope. | ## Next step Author a new YAML probe — the seed text, ASI/MITRE/CSA mapping, and the triple-framework gate the loader enforces. Where adapters sit in the six-phase swarm pipeline. Pick `fast` / `smart` / `full` for your CI-gate vs. release-gate runs. Every flag declared in `cli.py` with its default and meaning.