Agents

from rhesis.sdk.agents import BaseAgent, get_rhesis_tools

agent = BaseAgent(
    model="vertex_ai/gemini-2.0-flash",
    tools=get_rhesis_tools(),
    max_iterations=8,
)

result = agent.run("List my endpoints and summarize what each one tests.")
print(result.final_answer)

from rhesis.sdk.agents import BaseAgent, BaseTool
from rhesis.sdk.agents.schemas import ToolResult

class EchoTool(BaseTool):
    @property
    def name(self) -> str:
        return "echo"

    @property
    def description(self) -> str:
        return "Echoes a short message back to the caller."

    @property
    def parameters_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "message": {
                    "type": "string",
                    "description": "Message to echo back.",
                },
            },
            "required": ["message"],
        }

    async def execute(self, message: str, **kwargs) -> ToolResult:
        return ToolResult(tool_name=self.name, success=True, content=message)

agent = BaseAgent(tools=[EchoTool()])
result = agent.run("Call echo with the message hello.")
print(result.final_answer)

from rhesis.sdk.agents import AgentEventHandler, BaseAgent, get_rhesis_tools

class ConsoleEvents(AgentEventHandler):
    async def on_tool_start(self, *, tool_name, arguments, reasoning=None, **kwargs):
        print(f"Starting {tool_name}: {arguments}")

    async def on_tool_end(self, *, tool_name, result, **kwargs):
        print(f"Finished {tool_name}: success={result.success}")

agent = BaseAgent(
    tools=get_rhesis_tools(),
    event_handlers=[ConsoleEvents()],
)

agent.run("Show the first five test sets.")

import asyncio
from rhesis.sdk.agents import ArchitectAgent, get_rhesis_tools

agent = ArchitectAgent(tools=get_rhesis_tools())

attachments = {
    "mentions": [{"type": "endpoint", "id": "endpoint-uuid", "display": "Support Bot"}],
    "files": [{"filename": "requirements.md", "content": "# Test requirements..."}],
}

async def run():
    text = await agent.chat_async(
        "Use this endpoint and file to propose a test plan.",
        attachments=attachments,
    )
    print(text)

asyncio.run(run())

import asyncio
from rhesis.sdk.agents import ExploreEndpointTool
from rhesis.sdk.entities import Endpoint

endpoint = Endpoint(id="00000000-0000-0000-0000-000000000000")
endpoint.pull()

tool = ExploreEndpointTool(endpoint=endpoint, max_turns=5)

async def main():
    result = await tool.execute(strategy="domain_probing")
    print(result.content)

asyncio.run(main())