Tool Availability Gating

import os
from praisonaiagents import Agent
from praisonaiagents import tool

@tool(availability=lambda: (bool(os.getenv("SERP_API_KEY")), "SERP_API_KEY not set"))
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

agent = Agent(name="Researcher", tools=[search_web])
agent.start("Research quantum computing")
# Tool hidden if the key is missing. Once the key is set, a subsequent transient
# probe failure within 30 s keeps the tool available (last-good).

from praisonaiagents import tool
import os

# Simple environment check
@tool(availability=lambda: (bool(os.getenv("API_KEY")), "API_KEY missing"))
def api_tool(query: str) -> str:
    return f"API result: {query}"

# Complex dependency check  
def check_docker_available():
    try:
        import docker
        client = docker.from_env()
        client.ping()  # Quick ping, not heavy I/O
        return True, ""
    except Exception as e:
        return False, f"Docker unavailable: {e}"

@tool(availability=check_docker_available)
def docker_command(cmd: str) -> str:
    """Run Docker commands."""
    return f"Docker: {cmd}"

from praisonaiagents import BaseTool
from praisonaiagents.tools.protocols import ToolAvailabilityProtocol

class CloudTool(BaseTool, ToolAvailabilityProtocol):
    name = "cloud_deploy"
    description = "Deploy to cloud services"

    def check_availability(self) -> tuple[bool, str]:
        # Check multiple dependencies
        if not os.getenv("AWS_ACCESS_KEY"):
            return False, "AWS credentials not configured"

        try:
            import boto3
            # Quick credential test (fast operation only)
            boto3.Session().get_credentials()
            return True, ""
        except Exception as e:
            return False, f"AWS SDK error: {e}"

    def run(self, service: str) -> str:
        return f"Deployed {service} to cloud"

from praisonaiagents.tools import list_available_tools, get_registry

# Get only available tools
available_tools = list_available_tools()
print(f"Available: {len(available_tools)} tools")

# Get all tools (including unavailable)
all_tools = get_registry().list_tools()
print(f"Total registered: {len(all_tools)} tools")

# Check specific tool
web_tool = get_registry().get("search_web")
if hasattr(web_tool, 'check_availability'):
    is_available, reason = web_tool.check_availability()
    print(f"Web tool available: {is_available}")
    if not is_available:
        print(f"Reason: {reason}")

# Result depends on when the last successful probe was:
def flaky_availability_check():
    raise ValueError("Network unreachable")

@tool(availability=flaky_availability_check)
def network_tool(host: str) -> str:
    return f"Ping {host}"

# Never succeeded / >30 s since last success:
# WARNING: Availability check failed for tool 'network_tool': Network unreachable
# → hidden from the LLM for 30 s

# Recent success then transient failure (within 30 s):
# DEBUG:   Availability check failed for tool 'network_tool' but serving last-good within grace window: Network unreachable
# → tool stays in the schema; no negative cached

from praisonaiagents import register_tool

def simple_function(text: str) -> str:
    return f"Processed: {text}"

# Plain functions are always available (no protocol support yet)
register_tool(simple_function)

import os
from praisonaiagents import tool

def check_environment(required_vars: list[str]):
    """Factory for environment-based availability checks."""
    def check():
        missing = [var for var in required_vars if not os.getenv(var)]
        if missing:
            return False, f"Missing environment variables: {', '.join(missing)}"
        return True, ""
    return check

@tool(availability=check_environment(["OPENAI_API_KEY", "PINECONE_API_KEY"]))
def ai_research(topic: str) -> str:
    """Research topics using AI and vector search."""
    return f"AI research on: {topic}"

@tool(availability=check_environment(["SLACK_TOKEN"]))  
def notify_team(message: str) -> str:
    """Send notifications to team Slack."""
    return f"Notified team: {message}"

from praisonaiagents import tool
import socket

def check_service_available(host: str, port: int):
    """Check if a network service is reachable."""
    def check():
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(2)  # Keep fast
                s.connect((host, port))
                return True, ""
        except Exception as e:
            return False, f"Service {host}:{port} unreachable: {e}"
    return check

@tool(availability=check_service_available("localhost", 5432))
def query_local_db(sql: str) -> str:
    """Query local PostgreSQL database."""
    return f"SQL result: {sql}"

@tool(availability=check_service_available("redis-server", 6379))
def cache_data(key: str, value: str) -> str:
    """Cache data in Redis."""
    return f"Cached {key}: {value}"

from praisonaiagents import Agent
from praisonaiagents.tools import list_available_tools

# Build agent with only available tools
available_tools = list_available_tools()
print(f"Loading agent with {len(available_tools)} available tools")

agent = Agent(
    name="AdaptiveAgent",
    instructions="Use whatever tools are available in the current environment.",
    tools=available_tools
)

# Agent automatically adapts to environment capabilities