Skip to content

symfonic.evals.host_target

host_target

A started AgentHost as an in-process evaluation target.

HostTarget 

HostTarget(host_factory: Callable[[], Any], scope: Any | Mapping[str, Any] | Callable[[str], Any], *, evidence_factory: EvidenceFactory | None = None, resume: HostResumeSeam | None = None)

Start a public AgentHost and evaluate named tenant scopes.

Source code in src/symfonic/evals/host_target.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    host_factory: Callable[[], Any],
    scope: Any | Mapping[str, Any] | Callable[[str], Any],
    *,
    evidence_factory: EvidenceFactory | None = None,
    resume: HostResumeSeam | None = None,
) -> None:
    if not callable(host_factory):
        raise TypeError("HostTarget requires a callable host factory")
    self._host_factory = host_factory
    self._evidence_factory = evidence_factory
    self._evidence: TargetEvidenceAdapter | None = None
    self._scopes = scope
    self._host: Any = None
    self._resources: Any = None
    self._resume = resume
    self._agents: dict[str, AgentTarget] = {}
    self._sessions: dict[str, dict[str, str]] = {}
    self._started = False

capability_evidence async 

capability_evidence() -> CapabilityEvidence

Compile one scoped agent and report what the host actually serves.

Source code in src/symfonic/evals/host_target.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def capability_evidence(self) -> CapabilityEvidence:
    """Compile one scoped agent and report what the host actually serves."""
    await self._start()
    alias = next(iter(self._scopes), None) if isinstance(self._scopes, Mapping) else "default"
    if not alias:
        raise ValueError("HostTarget requires at least one scope for capability evidence")
    scope = self._scope_for(alias)
    agent = await self._host.agent_for(scope)
    channels = () if self._evidence is None else self._evidence.channels
    traits: Any = ()
    if self._evidence is not None:
        attest = getattr(self._evidence, "traits", None)
        if attest is not None:
            traits = attest(agent, self._resources, scope)
            if inspect.isawaitable(traits):
                traits = await traits
    return CapabilityEvidence.from_agent(
        agent, traits=traits, operations=self.operations,
        evidence_channels=channels,
    )