diff --git a/agentic_security/agents/operator_crew.py b/agentic_security/agents/operator_crew.py index 5daf3cf..60b0e85 100644 --- a/agentic_security/agents/operator_crew.py +++ b/agentic_security/agents/operator_crew.py @@ -1,12 +1,12 @@ -import os import asyncio import logging -from typing import Any, List, Dict +import os +from typing import Any import httpx -from pydantic import BaseModel, Field, ConfigDict -from crewai import Agent, Task, Crew +from crewai import Agent, Crew, Task from crewai_tools import tool +from pydantic import BaseModel, ConfigDict, Field # Assuming LLMSpec is defined elsewhere; placeholder import from agentic_security.http_spec import LLMSpec @@ -23,8 +23,8 @@ 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( + capabilities: list[str] | None = Field(None, description="List of capabilities") + configuration: dict[str, Any] | None = Field( None, description="Configuration settings" ) endpoint: str | None = Field(None, description="Endpoint URL of the deployed agent") @@ -34,7 +34,7 @@ class AgentSpecification(BaseModel): # Define OperatorToolBox class (unchanged from original) class OperatorToolBox: - def __init__(self, spec: AgentSpecification, datasets: List[Dict[str, Any]]): + def __init__(self, spec: AgentSpecification, datasets: list[dict[str, Any]]): self.spec = spec self.datasets = datasets self.failures = [] @@ -43,7 +43,7 @@ class OperatorToolBox: def get_spec(self) -> AgentSpecification: return self.spec - def get_datasets(self) -> List[Dict[str, Any]]: + def get_datasets(self) -> list[dict[str, Any]]: return self.datasets def validate(self) -> bool: @@ -61,10 +61,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: @@ -248,9 +248,9 @@ async def run_crew(): os.environ["OPENAI_API_KEY"] = os.environ.get( "DEEPSEEK_API_KEY", "" ) # CrewAI uses OPENAI_API_KEY -os.environ["OPENAI_MODEL_NAME"] = ( - "deepseek:chat" # Specify DeepSeek model (adjust if needed) -) +os.environ[ + "OPENAI_MODEL_NAME" +] = "deepseek:chat" # Specify DeepSeek model (adjust if needed) if __name__ == "__main__": asyncio.run(run_crew()) diff --git a/agentic_security/agents/operator_pydantic.py b/agentic_security/agents/operator_pydantic.py index 9af09e6..7e2b670 100644 --- a/agentic_security/agents/operator_pydantic.py +++ b/agentic_security/agents/operator_pydantic.py @@ -1,10 +1,10 @@ import asyncio import logging -from typing import Any, List, Dict +from typing import Any import httpx -from pydantic import BaseModel, Field, ConfigDict -from pydantic_ai import Agent, Tool, RunContext +from pydantic import BaseModel, ConfigDict, Field +from pydantic_ai import Agent, RunContext, Tool # Assuming LLMSpec is defined elsewhere; placeholder import from agentic_security.http_spec import LLMSpec @@ -21,8 +21,8 @@ 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( + capabilities: list[str] | None = Field(None, description="List of capabilities") + configuration: dict[str, Any] | None = Field( None, description="Configuration settings" ) endpoint: str | None = Field(None, description="Endpoint URL of the deployed agent") @@ -32,7 +32,7 @@ class AgentSpecification(BaseModel): # Define OperatorToolBox class class OperatorToolBox: - def __init__(self, spec: AgentSpecification, datasets: List[Dict[str, Any]]): + def __init__(self, spec: AgentSpecification, datasets: list[dict[str, Any]]): self.spec = spec self.datasets = datasets self.failures = [] @@ -41,7 +41,7 @@ class OperatorToolBox: def get_spec(self) -> AgentSpecification: return self.spec - def get_datasets(self) -> List[Dict[str, Any]]: + def get_datasets(self) -> list[dict[str, Any]]: return self.datasets def validate(self) -> bool: @@ -59,10 +59,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: