1
2
3
4
5
6
class Environment(str, Enum):
    """Deployment environment."""

    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"

1
2
3
4
5
class IdentityProvider(str, Enum):
    """Identity provider type."""

    SPIFFE = "spiffe"
    MTLS_STATIC = "mtls-static"  # For testing/development only

1
2
3
4
5
class AuthorizationProvider(str, Enum):
    """Authorization provider type."""

    OPA = "opa"
    ALLOW_ALL = "allow-all"  # Development only - rejected in production

1
2
3
4
5
class DefaultAction(str, Enum):
    """Default authorization action."""

    DENY = "deny"
    LOG_ONLY = "log-only"  # Development only - rejected in production

1
2
3
4
5
class PeerVerification(str, Enum):
    """Peer verification mode."""

    STRICT = "strict"
    LOG_ONLY = "log-only"  # Development only

1
2
3
4
5
class ProtocolType(str, Enum):
    """Communication protocol type."""

    A2A = "a2a"
    GRPC = "grpc"

1
2
3
4
5
6
7
class Capability(BaseModel):
    """Agent capability definition."""

    name: str
    description: str
    input_modes: list[str] = ["application/json"]
    output_modes: list[str] = ["application/json"]

1
2
3
4
5
6
from agentweave.config import Capability

cap = Capability(
    name="search_users",
    description="Search for users in the directory"
)

1
2
3
4
5
6
7
8
class AgentSettings(BaseModel):
    """Core agent settings."""

    name: str
    trust_domain: str
    description: str = ""
    environment: Environment = Environment.PRODUCTION
    capabilities: list[Capability] = []

1
2
3
4
5
6
7
8
9
10
from agentweave.config import AgentSettings, Environment, Capability

settings = AgentSettings(
    name="data-search-agent",
    trust_domain="agentweave.io",
    environment=Environment.PRODUCTION,
    capabilities=[
        Capability(name="search", description="Search the database")
    ]
)

1
2
3
4
5
6
class IdentityConfig(BaseModel):
    """Identity provider configuration."""

    provider: IdentityProvider = IdentityProvider.SPIFFE
    spiffe_endpoint: str = "unix:///run/spire/sockets/agent.sock"
    allowed_trust_domains: list[str] = []

1
2
3
4
5
6
7
from agentweave.config import IdentityConfig, IdentityProvider

identity = IdentityConfig(
    provider=IdentityProvider.SPIFFE,
    spiffe_endpoint="unix:///run/spire/sockets/agent.sock",
    allowed_trust_domains=["partner.io"]
)

1
2
3
4
5
class AuditConfig(BaseModel):
    """Audit logging configuration."""

    enabled: bool = True
    destination: str = "file:///var/log/agentweave/audit.log"

1
2
3
4
5
6
7
8
class AuthorizationConfig(BaseModel):
    """Authorization provider configuration."""

    provider: AuthorizationProvider = AuthorizationProvider.OPA
    opa_endpoint: str = "http://localhost:8181"
    policy_path: str = "agentweave/authz"
    default_action: DefaultAction = DefaultAction.DENY
    audit: AuditConfig = AuditConfig()

1
2
3
4
5
6
7
from agentweave.config import AuthorizationConfig, AuthorizationProvider, DefaultAction

authz = AuthorizationConfig(
    provider=AuthorizationProvider.OPA,
    opa_endpoint="http://localhost:8181",
    default_action=DefaultAction.DENY
)

1
2
3
4
5
class ConnectionPoolConfig(BaseModel):
    """Connection pool configuration."""

    max_connections: int = Field(default=100, ge=1, le=10000)
    idle_timeout_seconds: int = Field(default=60, ge=1)

1
2
3
4
5
class CircuitBreakerConfig(BaseModel):
    """Circuit breaker configuration."""

    failure_threshold: int = Field(default=5, ge=1)
    recovery_timeout_seconds: int = Field(default=30, ge=1)

1
2
3
4
5
6
class RetryConfig(BaseModel):
    """Retry policy configuration."""

    max_attempts: int = Field(default=3, ge=1, le=10)
    backoff_base_seconds: float = Field(default=1.0, ge=0.1)
    backoff_max_seconds: float = Field(default=30.0, ge=1.0)

1
2
3
4
5
6
7
8
class TransportConfig(BaseModel):
    """Transport layer configuration."""

    tls_min_version: Literal["1.2", "1.3"] = "1.3"
    peer_verification: PeerVerification = PeerVerification.STRICT
    connection_pool: ConnectionPoolConfig = ConnectionPoolConfig()
    circuit_breaker: CircuitBreakerConfig = CircuitBreakerConfig()
    retry: RetryConfig = RetryConfig()

1
2
3
4
5
6
class ServerConfig(BaseModel):
    """Server configuration."""

    host: str = "0.0.0.0"
    port: int = Field(default=8443, ge=1, le=65535)
    protocol: ProtocolType = ProtocolType.A2A

1
2
3
4
5
class MetricsConfig(BaseModel):
    """Metrics configuration."""

    enabled: bool = True
    port: int = Field(default=9090, ge=1, le=65535)

1
2
3
4
5
6
class TracingConfig(BaseModel):
    """Distributed tracing configuration."""

    enabled: bool = True
    exporter: Literal["otlp", "jaeger", "zipkin"] = "otlp"
    endpoint: str = "http://localhost:4317"

1
2
3
4
5
class LoggingConfig(BaseModel):
    """Logging configuration."""

    level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
    format: Literal["json", "text"] = "json"

1
2
3
4
5
6
class ObservabilityConfig(BaseModel):
    """Observability configuration."""

    metrics: MetricsConfig = MetricsConfig()
    tracing: TracingConfig = TracingConfig()
    logging: LoggingConfig = LoggingConfig()

1
2
3
4
5
6
7
8
9
class AgentConfig(BaseModel):
    """Complete agent configuration with strict security validation."""

    agent: AgentSettings
    identity: IdentityConfig = IdentityConfig()
    authorization: AuthorizationConfig = AuthorizationConfig()
    transport: TransportConfig = TransportConfig()
    server: ServerConfig = ServerConfig()
    observability: ObservabilityConfig = ObservabilityConfig()

1
def is_production(self) -> bool

1
2
3
config = AgentConfig.from_file("config.yaml")
if config.is_production():
    print("Running in production mode")

1
2
@classmethod
def from_file(cls, path: str | Path) -> AgentConfig

1
2
3
from agentweave.config import AgentConfig

config = AgentConfig.from_file("config.yaml")

1
2
@classmethod
def from_env(cls, prefix: str = "AGENTWEAVE_") -> AgentConfig

1
2
3
4
5
6
7
from agentweave.config import AgentConfig

# Load from environment
config = AgentConfig.from_env()

# Custom prefix
config = AgentConfig.from_env(prefix="MYAPP_")

1
2
3
4
if self.is_production() and self.authorization.default_action != DefaultAction.DENY:
    raise ConfigurationError(
        "authorization.default_action must be 'deny' in production environment"
    )

1
2
3
4
if self.is_production() and self.authorization.provider == AuthorizationProvider.ALLOW_ALL:
    raise ConfigurationError(
        "authorization.provider cannot be 'allow-all' in production environment"
    )

1
2
3
4
if self.is_production() and self.transport.peer_verification != PeerVerification.STRICT:
    raise ConfigurationError(
        "transport.peer_verification must be 'strict' in production environment"
    )

1
2
3
4
if self.transport.tls_min_version not in ("1.2", "1.3"):
    raise ConfigurationError(
        "transport.tls_min_version must be '1.2' or '1.3'"
    )

1
2
3
4
if self.is_production() and not self.authorization.audit.enabled:
    raise ConfigurationError(
        "authorization.audit.enabled must be true in production environment"
    )

1
2
3
4
5
6
7
8
9
10
11
from agentweave.config import AgentConfig
from agentweave import SecureAgent

# Load configuration
config = AgentConfig.from_file("config.yaml")

# Use with agent
class MyAgent(SecureAgent):
    pass

agent = MyAgent(config=config.agent)  # Pass AgentSettings