From 4c0d89bf86179b9ebbc0d17a8669a3bd7496b692 Mon Sep 17 00:00:00 2001 From: Praveenk8051 Date: Thu, 30 Jan 2025 07:46:32 +0100 Subject: [PATCH 1/6] feat(operator): add agent testing functionality with endpoint verification --- agentic_security/probe_actor/operator.py | 77 ++++++++++++++++++++++-- 1 file changed, 73 insertions(+), 4 deletions(-) diff --git a/agentic_security/probe_actor/operator.py b/agentic_security/probe_actor/operator.py index c7c0ca8..73440cf 100644 --- a/agentic_security/probe_actor/operator.py +++ b/agentic_security/probe_actor/operator.py @@ -1,9 +1,15 @@ import asyncio -from typing import Any +from typing import Any, Optional, Dict from pydantic import BaseModel, Field from pydantic_ai import Agent, RunContext +import re +import httpx +import logging +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) class AgentSpecification(BaseModel): name: str | None = Field(None, description="Name of the LLM/agent") @@ -13,6 +19,7 @@ class AgentSpecification(BaseModel): configuration: dict[str, Any] | None = Field( None, description="Configuration settings" ) + endpoint: Optional[str] = Field(None, description="Endpoint URL of the deployed agent") # Define the OperatorToolBox class @@ -61,6 +68,51 @@ class OperatorToolBox: return f"Operation '{operation}' failed: Dataset not found." return f"Operation '{operation}' executed successfully." + async def test(self, description: str, sample_test: Dict[str, Any]) -> str: + """Test the agent based on the description and sample test.""" + match = re.search(r"Test my (.+) agent deployed at (.+)", description) + if match: + agent_type = match.group(1) + endpoint = match.group(2) + self.spec.endpoint = endpoint + + # Verify access to the endpoint + async with httpx.AsyncClient() as client: + try: + access_response = await client.get(endpoint) + access_response.raise_for_status() + except httpx.HTTPStatusError as e: + self.failures.append(f"HTTP error occurred: {e}") + logger.error(f"Access verification failed: {e}") + return f"Access verification failed: {e}" + except Exception as e: + self.failures.append(f"An error occurred: {e}") + logger.error(f"Access verification failed: {e}") + return f"Access verification failed: {e}" + + # Run the sample test + try: + test_response = await client.post(f"{endpoint}/test", json=sample_test) + test_response.raise_for_status() + response_data = test_response.json() + # Validate the response (this is a simple example, adjust as needed) + if "choices" in response_data and len(response_data["choices"]) > 0: + return f"Testing {agent_type} agent at {endpoint} succeeded: {response_data}" + else: + self.failures.append("Invalid response format") + logger.error("Sample test failed: Invalid response format") + return "Sample test failed: Invalid response format" + except httpx.HTTPStatusError as e: + self.failures.append(f"HTTP error occurred: {e}") + logger.error(f"Sample test failed: {e}") + return f"Sample test failed: {e}" + except Exception as e: + self.failures.append(f"An error occurred: {e}") + logger.error(f"Sample test failed: {e}") + return f"Sample test failed: {e}" + else: + logger.error("Invalid description format.") + return "Invalid description format." # Initialize OperatorToolBox with AgentSpecification spec = AgentSpecification( @@ -133,14 +185,22 @@ def run_dataset_manager_agent_sync(): "Execute operation on 'dataset4'.", # This should fail "Retrieve the results.", "Retrieve any failures.", + "Test my openAI compatible agent deployed at localhost:3000" ] + sample_test = { + "prompt": "Hello, how are you?", + "max_tokens": 5 + } + for prompt in prompts: - result = dataset_manager_agent.run_sync(prompt, deps=toolbox) + if "Test my" in prompt: + result = dataset_manager_agent.run_sync(prompt, deps=toolbox, sample_test=sample_test) + else: + result = dataset_manager_agent.run_sync(prompt, deps=toolbox) print(f"Prompt: {prompt}") print(f"Response: {result.data}\n") - # Asynchronous run example async def run_dataset_manager_agent_async(): prompts = [ @@ -149,10 +209,19 @@ async def run_dataset_manager_agent_async(): "Execute operation on 'dataset4'.", # This should fail "Retrieve the results.", "Retrieve any failures.", + "Test my openAI compatible agent deployed at localhost:3000" ] + sample_test = { + "prompt": "Hello, how are you?", + "max_tokens": 5 + } + for prompt in prompts: - result = await dataset_manager_agent.run(prompt, deps=toolbox) + if "Test my" in prompt: + result = await dataset_manager_agent.run(prompt, deps=toolbox, sample_test=sample_test) + else: + result = await dataset_manager_agent.run(prompt, deps=toolbox) print(f"Prompt: {prompt}") print(f"Response: {result.data}\n") From fa37cfe7104042637b510668522c6303cfb93cb5 Mon Sep 17 00:00:00 2001 From: Praveenk8051 Date: Sun, 16 Feb 2025 15:45:20 +0100 Subject: [PATCH 2/6] feat: enhance AgentSpecification and OperatorToolBox with optional typing and improved logging --- agentic_security/probe_actor/operator.py | 129 ++++++++++------------- 1 file changed, 54 insertions(+), 75 deletions(-) diff --git a/agentic_security/probe_actor/operator.py b/agentic_security/probe_actor/operator.py index 73440cf..20c82e2 100644 --- a/agentic_security/probe_actor/operator.py +++ b/agentic_security/probe_actor/operator.py @@ -1,10 +1,9 @@ import asyncio -from typing import Any, Optional, Dict - +from typing import Any, Optional, Dict, List from pydantic import BaseModel, Field from pydantic_ai import Agent, RunContext -import re import httpx +from httpx import LLMSpec import logging # Configure logging @@ -12,17 +11,13 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class AgentSpecification(BaseModel): - name: str | None = Field(None, description="Name of the LLM/agent") - version: str | None = Field(None, description="Version of the LLM/agent") - description: str | None = Field(None, description="Description of the LLM/agent") - capabilities: list[str] | None = Field(None, description="List of capabilities") - configuration: dict[str, Any] | None = Field( - None, description="Configuration settings" - ) + name: Optional[str] = Field(None, description="Name of the LLM/agent") + version: Optional[str] = Field(None, description="Version of the LLM/agent") + description: Optional[str] = Field(None, description="Description of the LLM/agent") + capabilities: Optional[List[str]] = Field(None, description="List of capabilities") + configuration: Optional[Dict[str, Any]] = Field(None, description="Configuration settings") endpoint: Optional[str] = Field(None, description="Endpoint URL of the deployed agent") - -# Define the OperatorToolBox class class OperatorToolBox: def __init__(self, spec: AgentSpecification, datasets: list[dict[str, Any]]): self.spec = spec @@ -36,7 +31,6 @@ class OperatorToolBox: return self.datasets def validate(self) -> bool: - # Validate the tool box based on the specification if not self.spec.name or not self.spec.version: self.failures.append("Invalid specification: Name or version is missing.") return False @@ -46,73 +40,67 @@ class OperatorToolBox: return True def stop(self) -> None: - # Stop the tool box - print("Stopping the toolbox...") + logger.info("Stopping the toolbox...") def run(self) -> None: - # Run the tool box - print("Running the toolbox...") + logger.info("Running the toolbox...") - def get_results(self) -> list[dict[str, Any]]: - # Get the results + def get_results(self) -> List[Dict[str, Any]]: return self.datasets - def get_failures(self) -> list[str]: - # Handle failure + def get_failures(self) -> List[str]: return self.failures def run_operation(self, operation: str) -> str: - # Run an operation based on the specification if operation not in ["dataset1", "dataset2", "dataset3"]: self.failures.append(f"Operation '{operation}' failed: Dataset not found.") return f"Operation '{operation}' failed: Dataset not found." return f"Operation '{operation}' executed successfully." async def test(self, description: str, sample_test: Dict[str, Any]) -> str: - """Test the agent based on the description and sample test.""" - match = re.search(r"Test my (.+) agent deployed at (.+)", description) - if match: - agent_type = match.group(1) - endpoint = match.group(2) - self.spec.endpoint = endpoint + agent = Agent( + 'openai:gpt-4o', + result_type=LLMSpec, + system_prompt='Extract the LLM specification from the input', + ) - # Verify access to the endpoint - async with httpx.AsyncClient() as client: + async with agent.run_stream(description) as result: + async for spec in result.stream(): + self.spec.endpoint = spec.url + + # Verify access to the endpoint + async with httpx.AsyncClient() as client: + try: + access_response = await client.get(spec.url) + access_response.raise_for_status() + except httpx.HTTPStatusError as e: + self.failures.append(f"HTTP error occurred: {e}") + logger.error(f"Access verification failed: {e}") + return f"Access verification failed: {e}" + except Exception as e: + self.failures.append(f"An error occurred: {e}") + logger.error(f"Access verification failed: {e}") + return f"Access verification failed: {e}" + + # Run the sample test try: - access_response = await client.get(endpoint) - access_response.raise_for_status() + test_response = await client.post(f"{spec.url}/test", json=sample_test) + test_response.raise_for_status() + response_data = test_response.json() + if "choices" in response_data and len(response_data["choices"]) > 0: + return f"Testing agent at {spec.url} succeeded: {response_data}" + else: + self.failures.append("Invalid response format") + logger.error("Sample test failed: Invalid response format") + return "Sample test failed: Invalid response format" except httpx.HTTPStatusError as e: self.failures.append(f"HTTP error occurred: {e}") - logger.error(f"Access verification failed: {e}") - return f"Access verification failed: {e}" + logger.error(f"Sample test failed: {e}") + return f"Sample test failed: {e}" except Exception as e: self.failures.append(f"An error occurred: {e}") - logger.error(f"Access verification failed: {e}") - return f"Access verification failed: {e}" - - # Run the sample test - try: - test_response = await client.post(f"{endpoint}/test", json=sample_test) - test_response.raise_for_status() - response_data = test_response.json() - # Validate the response (this is a simple example, adjust as needed) - if "choices" in response_data and len(response_data["choices"]) > 0: - return f"Testing {agent_type} agent at {endpoint} succeeded: {response_data}" - else: - self.failures.append("Invalid response format") - logger.error("Sample test failed: Invalid response format") - return "Sample test failed: Invalid response format" - except httpx.HTTPStatusError as e: - self.failures.append(f"HTTP error occurred: {e}") - logger.error(f"Sample test failed: {e}") - return f"Sample test failed: {e}" - except Exception as e: - self.failures.append(f"An error occurred: {e}") - logger.error(f"Sample test failed: {e}") - return f"Sample test failed: {e}" - else: - logger.error("Invalid description format.") - return "Invalid description format." + logger.error(f"Sample test failed: {e}") + return f"Sample test failed: {e}" # Initialize OperatorToolBox with AgentSpecification spec = AgentSpecification( @@ -123,41 +111,31 @@ spec = AgentSpecification( configuration={"max_tokens": 100}, ) -# dataset_manager_agent.py - - -# Initialize OperatorToolBox toolbox = OperatorToolBox(spec=spec, datasets=["dataset1", "dataset2", "dataset3"]) # Define the agent with OperatorToolBox as its dependency dataset_manager_agent = Agent( model="gpt-4", deps_type=OperatorToolBox, - result_type=str, # The agent will return string results + result_type=str, system_prompt="You can validate the toolbox, run operations, and retrieve results or failures.", ) - @dataset_manager_agent.tool async def validate_toolbox(ctx: RunContext[OperatorToolBox]) -> str: - """Validate the OperatorToolBox.""" is_valid = ctx.deps.validate() if is_valid: return "ToolBox validation successful." else: return "ToolBox validation failed." - @dataset_manager_agent.tool async def execute_operation(ctx: RunContext[OperatorToolBox], operation: str) -> str: - """Execute an operation on a dataset.""" result = ctx.deps.run_operation(operation) return result - @dataset_manager_agent.tool async def retrieve_results(ctx: RunContext[OperatorToolBox]) -> str: - """Retrieve the results of operations.""" results = ctx.deps.get_results() if results: formatted_results = "\n".join([f"{op}: {res}" for op, res in results.items()]) @@ -165,10 +143,8 @@ async def retrieve_results(ctx: RunContext[OperatorToolBox]) -> str: else: return "No operations have been executed yet." - @dataset_manager_agent.tool async def retrieve_failures(ctx: RunContext[OperatorToolBox]) -> str: - """Retrieve the list of failures.""" failures = ctx.deps.get_failures() if failures: formatted_failures = "\n".join(failures) @@ -176,6 +152,10 @@ async def retrieve_failures(ctx: RunContext[OperatorToolBox]) -> str: else: return "No failures recorded." +@dataset_manager_agent.tool +async def test_agent(ctx: RunContext[OperatorToolBox], description: str, sample_test: Dict[str, Any]) -> str: + result = await ctx.deps.test(description, sample_test) + return result # Synchronous run example def run_dataset_manager_agent_sync(): @@ -225,10 +205,9 @@ async def run_dataset_manager_agent_async(): print(f"Prompt: {prompt}") print(f"Response: {result.data}\n") - if __name__ == "__main__": # Run synchronous example run_dataset_manager_agent_sync() # Run asynchronous example - asyncio.run(run_dataset_manager_agent_async()) + asyncio.run(run_dataset_manager_agent_async()) \ No newline at end of file From 8cc4d79ddf542fb419ade593ecd0099a136b75fe Mon Sep 17 00:00:00 2001 From: Praveenk8051 Date: Sun, 16 Feb 2025 15:53:13 +0100 Subject: [PATCH 3/6] fix: update type hints in OperatorToolBox for consistency --- agentic_security/probe_actor/operator.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/agentic_security/probe_actor/operator.py b/agentic_security/probe_actor/operator.py index 20c82e2..d815b8c 100644 --- a/agentic_security/probe_actor/operator.py +++ b/agentic_security/probe_actor/operator.py @@ -1,5 +1,5 @@ import asyncio -from typing import Any, Optional, Dict, List +from typing import Any, Optional from pydantic import BaseModel, Field from pydantic_ai import Agent, RunContext import httpx @@ -14,8 +14,8 @@ class AgentSpecification(BaseModel): name: Optional[str] = Field(None, description="Name of the LLM/agent") version: Optional[str] = Field(None, description="Version of the LLM/agent") description: Optional[str] = Field(None, description="Description of the LLM/agent") - capabilities: Optional[List[str]] = Field(None, description="List of capabilities") - configuration: Optional[Dict[str, Any]] = Field(None, description="Configuration settings") + capabilities: Optional[list[str]] = Field(None, description="List of capabilities") + configuration: Optional[dict[str, Any]] = Field(None, description="Configuration settings") endpoint: Optional[str] = Field(None, description="Endpoint URL of the deployed agent") class OperatorToolBox: @@ -45,10 +45,10 @@ class OperatorToolBox: def run(self) -> None: logger.info("Running the toolbox...") - def get_results(self) -> List[Dict[str, Any]]: + def get_results(self) -> list[dict[str, Any]]: return self.datasets - def get_failures(self) -> List[str]: + def get_failures(self) -> list[str]: return self.failures def run_operation(self, operation: str) -> str: @@ -57,7 +57,7 @@ class OperatorToolBox: return f"Operation '{operation}' failed: Dataset not found." return f"Operation '{operation}' executed successfully." - async def test(self, description: str, sample_test: Dict[str, Any]) -> str: + async def test(self, description: str, sample_test: dict[str, Any]) -> str: agent = Agent( 'openai:gpt-4o', result_type=LLMSpec, @@ -153,7 +153,7 @@ async def retrieve_failures(ctx: RunContext[OperatorToolBox]) -> str: return "No failures recorded." @dataset_manager_agent.tool -async def test_agent(ctx: RunContext[OperatorToolBox], description: str, sample_test: Dict[str, Any]) -> str: +async def test_agent(ctx: RunContext[OperatorToolBox], description: str, sample_test: dict[str, Any]) -> str: result = await ctx.deps.test(description, sample_test) return result From 1c6b8d96fb2dc3e20b870ec9ad5c4756c5007221 Mon Sep 17 00:00:00 2001 From: Praveenk8051 Date: Sun, 16 Feb 2025 15:56:16 +0100 Subject: [PATCH 4/6] style: improve code formatting and consistency in operator.py --- agentic_security/probe_actor/operator.py | 55 ++++++++++++++++-------- 1 file changed, 36 insertions(+), 19 deletions(-) diff --git a/agentic_security/probe_actor/operator.py b/agentic_security/probe_actor/operator.py index d815b8c..76890eb 100644 --- a/agentic_security/probe_actor/operator.py +++ b/agentic_security/probe_actor/operator.py @@ -10,13 +10,19 @@ import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) + class AgentSpecification(BaseModel): name: Optional[str] = Field(None, description="Name of the LLM/agent") version: Optional[str] = Field(None, description="Version of the LLM/agent") description: Optional[str] = Field(None, description="Description of the LLM/agent") capabilities: Optional[list[str]] = Field(None, description="List of capabilities") - configuration: Optional[dict[str, Any]] = Field(None, description="Configuration settings") - endpoint: Optional[str] = Field(None, description="Endpoint URL of the deployed agent") + configuration: Optional[dict[str, Any]] = Field( + None, description="Configuration settings" + ) + endpoint: Optional[str] = Field( + None, description="Endpoint URL of the deployed agent" + ) + class OperatorToolBox: def __init__(self, spec: AgentSpecification, datasets: list[dict[str, Any]]): @@ -59,9 +65,9 @@ class OperatorToolBox: async def test(self, description: str, sample_test: dict[str, Any]) -> str: agent = Agent( - 'openai:gpt-4o', + "openai:gpt-4o", result_type=LLMSpec, - system_prompt='Extract the LLM specification from the input', + system_prompt="Extract the LLM specification from the input", ) async with agent.run_stream(description) as result: @@ -84,7 +90,9 @@ class OperatorToolBox: # Run the sample test try: - test_response = await client.post(f"{spec.url}/test", json=sample_test) + test_response = await client.post( + f"{spec.url}/test", json=sample_test + ) test_response.raise_for_status() response_data = test_response.json() if "choices" in response_data and len(response_data["choices"]) > 0: @@ -102,6 +110,7 @@ class OperatorToolBox: logger.error(f"Sample test failed: {e}") return f"Sample test failed: {e}" + # Initialize OperatorToolBox with AgentSpecification spec = AgentSpecification( name="GPT-4", @@ -121,6 +130,7 @@ dataset_manager_agent = Agent( system_prompt="You can validate the toolbox, run operations, and retrieve results or failures.", ) + @dataset_manager_agent.tool async def validate_toolbox(ctx: RunContext[OperatorToolBox]) -> str: is_valid = ctx.deps.validate() @@ -129,11 +139,13 @@ async def validate_toolbox(ctx: RunContext[OperatorToolBox]) -> str: else: return "ToolBox validation failed." + @dataset_manager_agent.tool async def execute_operation(ctx: RunContext[OperatorToolBox], operation: str) -> str: result = ctx.deps.run_operation(operation) return result + @dataset_manager_agent.tool async def retrieve_results(ctx: RunContext[OperatorToolBox]) -> str: results = ctx.deps.get_results() @@ -143,6 +155,7 @@ async def retrieve_results(ctx: RunContext[OperatorToolBox]) -> str: else: return "No operations have been executed yet." + @dataset_manager_agent.tool async def retrieve_failures(ctx: RunContext[OperatorToolBox]) -> str: failures = ctx.deps.get_failures() @@ -152,11 +165,15 @@ async def retrieve_failures(ctx: RunContext[OperatorToolBox]) -> str: else: return "No failures recorded." + @dataset_manager_agent.tool -async def test_agent(ctx: RunContext[OperatorToolBox], description: str, sample_test: dict[str, Any]) -> str: +async def test_agent( + ctx: RunContext[OperatorToolBox], description: str, sample_test: dict[str, Any] +) -> str: result = await ctx.deps.test(description, sample_test) return result + # Synchronous run example def run_dataset_manager_agent_sync(): prompts = [ @@ -165,22 +182,22 @@ def run_dataset_manager_agent_sync(): "Execute operation on 'dataset4'.", # This should fail "Retrieve the results.", "Retrieve any failures.", - "Test my openAI compatible agent deployed at localhost:3000" + "Test my openAI compatible agent deployed at localhost:3000", ] - sample_test = { - "prompt": "Hello, how are you?", - "max_tokens": 5 - } + sample_test = {"prompt": "Hello, how are you?", "max_tokens": 5} for prompt in prompts: if "Test my" in prompt: - result = dataset_manager_agent.run_sync(prompt, deps=toolbox, sample_test=sample_test) + result = dataset_manager_agent.run_sync( + prompt, deps=toolbox, sample_test=sample_test + ) else: result = dataset_manager_agent.run_sync(prompt, deps=toolbox) print(f"Prompt: {prompt}") print(f"Response: {result.data}\n") + # Asynchronous run example async def run_dataset_manager_agent_async(): prompts = [ @@ -189,25 +206,25 @@ async def run_dataset_manager_agent_async(): "Execute operation on 'dataset4'.", # This should fail "Retrieve the results.", "Retrieve any failures.", - "Test my openAI compatible agent deployed at localhost:3000" + "Test my openAI compatible agent deployed at localhost:3000", ] - sample_test = { - "prompt": "Hello, how are you?", - "max_tokens": 5 - } + sample_test = {"prompt": "Hello, how are you?", "max_tokens": 5} for prompt in prompts: if "Test my" in prompt: - result = await dataset_manager_agent.run(prompt, deps=toolbox, sample_test=sample_test) + result = await dataset_manager_agent.run( + prompt, deps=toolbox, sample_test=sample_test + ) else: result = await dataset_manager_agent.run(prompt, deps=toolbox) print(f"Prompt: {prompt}") print(f"Response: {result.data}\n") + if __name__ == "__main__": # Run synchronous example run_dataset_manager_agent_sync() # Run asynchronous example - asyncio.run(run_dataset_manager_agent_async()) \ No newline at end of file + asyncio.run(run_dataset_manager_agent_async()) From a001a33f688942db3a014a64d4acff149869645d Mon Sep 17 00:00:00 2001 From: Praveenk8051 Date: Sun, 16 Feb 2025 16:11:46 +0100 Subject: [PATCH 5/6] refactor: update type hints in AgentSpecification for improved clarity and consistency --- agentic_security/probe_actor/operator.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/agentic_security/probe_actor/operator.py b/agentic_security/probe_actor/operator.py index 76890eb..265796a 100644 --- a/agentic_security/probe_actor/operator.py +++ b/agentic_security/probe_actor/operator.py @@ -1,25 +1,25 @@ import asyncio -from typing import Any, Optional -from pydantic import BaseModel, Field -from pydantic_ai import Agent, RunContext +import logging +from typing import Any + import httpx from httpx import LLMSpec -import logging +from pydantic import BaseModel, Field +from pydantic_ai import Agent, RunContext # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) - class AgentSpecification(BaseModel): - name: Optional[str] = Field(None, description="Name of the LLM/agent") - version: Optional[str] = Field(None, description="Version of the LLM/agent") - description: Optional[str] = Field(None, description="Description of the LLM/agent") - capabilities: Optional[list[str]] = Field(None, description="List of capabilities") - configuration: Optional[dict[str, Any]] = Field( + name: str | None = Field(None, description="Name of the LLM/agent") + version: str | None = Field(None, description="Version of the LLM/agent") + description: str | None = Field(None, description="Description of the LLM/agent") + capabilities: list[str] | None = Field(None, description="List of capabilities") + configuration: dict[str, Any] | None = Field( None, description="Configuration settings" ) - endpoint: Optional[str] = Field( + endpoint: str | None = Field( None, description="Endpoint URL of the deployed agent" ) From 121d56495e9a76a6ff288c90d60c5c4eae82c19f Mon Sep 17 00:00:00 2001 From: Praveenk8051 Date: Sun, 16 Feb 2025 16:13:21 +0100 Subject: [PATCH 6/6] style: streamline code formatting in operator.py for improved readability --- agentic_security/probe_actor/operator.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/agentic_security/probe_actor/operator.py b/agentic_security/probe_actor/operator.py index 265796a..c270dd1 100644 --- a/agentic_security/probe_actor/operator.py +++ b/agentic_security/probe_actor/operator.py @@ -11,6 +11,7 @@ from pydantic_ai import Agent, RunContext logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) + class AgentSpecification(BaseModel): name: str | None = Field(None, description="Name of the LLM/agent") version: str | None = Field(None, description="Version of the LLM/agent") @@ -19,9 +20,7 @@ class AgentSpecification(BaseModel): configuration: dict[str, Any] | None = Field( None, description="Configuration settings" ) - endpoint: str | None = Field( - None, description="Endpoint URL of the deployed agent" - ) + endpoint: str | None = Field(None, description="Endpoint URL of the deployed agent") class OperatorToolBox: