Python Tool

from agentscope.tool import Bash

bash = Bash(
    additional_dangerous_files=[".secrets"],
    additional_dangerous_directories=[".credentials"],
)

from agentscope.tool import PowerShell

pwsh = PowerShell(cwd="C:\\Users\\me\\project")

read = Read(model_input_types=model_card.input_types)
result = await read(file_path="/workspace/report.pdf", pages="1-5")

from agentscope.tool import ToolBase, ToolChunk
from agentscope.permission import (
    PermissionContext, PermissionDecision, PermissionBehavior,
)
from agentscope.message import TextBlock

class WebSearch(ToolBase):
    name = "WebSearch"
    description = "Search the web for information on a given query."
    input_schema = {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "The search query.",
            },
        },
        "required": ["query"],
    }
    is_concurrency_safe = True
    is_read_only = True

    async def check_permissions(
        self, tool_input: dict, context: PermissionContext,
    ) -> PermissionDecision:
        return PermissionDecision(
            behavior=PermissionBehavior.ALLOW,
            message="Web search is read-only.",
        )

    async def call(self, query: str) -> ToolChunk:
        results = await do_search(query)
        return ToolChunk(content=[TextBlock(text=results)])

from agentscope.tool import FunctionTool, Toolkit

def get_weather(city: str, unit: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name to look up.
        unit: Temperature unit, either "celsius" or "fahrenheit".
    """
    return f"The weather in {city} is 22°{unit[0].upper()}"

toolkit = Toolkit(tools=[FunctionTool(get_weather)])

from pydantic import BaseModel, Field


class WeatherInput(BaseModel):
    city: str
    days: int = Field(default=1, ge=1, le=7)


def get_forecast(city: str, days: int = 1) -> str:
    return f"Sunny in {city} for the next {days} day(s)"


weather_tool = FunctionTool(get_forecast, input_schema=WeatherInput)

from agentscope.tool import ToolBase
from agentscope.permission import (
    PermissionContext, PermissionDecision, PermissionBehavior,
)

class HumanApproval(ToolBase):
    name = "HumanApproval"
    description = "Request human approval for a sensitive operation."
    input_schema = {
        "type": "object",
        "properties": {
            "action": {"type": "string", "description": "The action requiring approval."},
            "reason": {"type": "string", "description": "Why this action needs approval."},
        },
        "required": ["action", "reason"],
    }
    is_concurrency_safe = True
    is_read_only = False
    is_external_tool = True

    async def check_permissions(
        self, tool_input: dict, context: PermissionContext,
    ) -> PermissionDecision:
        return PermissionDecision(
            behavior=PermissionBehavior.ALLOW,
            message="External tool dispatch is always allowed.",
        )

middlewares[0].on_tool_call
  └─ middlewares[1].on_tool_call
       └─ ... → tool.call()

from agentscope.tool import Bash

bash = Bash(middlewares=[LoggingMiddleware(), MetricsMiddleware()])
# Execution order: LoggingMiddleware → MetricsMiddleware → Bash.call()

from typing import AsyncGenerator, Any, Callable
from agentscope.tool import ToolMiddlewareBase, ToolBase, ToolChunk, Bash


class LoggingMiddleware(ToolMiddlewareBase):
    async def on_tool_call(
        self,
        tool: ToolBase,
        input_kwargs: dict[str, Any],
        next_handler: Callable[..., AsyncGenerator[ToolChunk, None]],
    ) -> AsyncGenerator[ToolChunk, None]:
        print(f"→ Calling {tool.name} with {input_kwargs}")
        async for chunk in next_handler(**input_kwargs):
            yield chunk
        print(f"✓ {tool.name} finished")


class RetryMiddleware(ToolMiddlewareBase):
    def __init__(self, max_attempts: int = 3):
        self.max_attempts = max_attempts

    async def on_tool_call(
        self,
        tool: ToolBase,
        input_kwargs: dict[str, Any],
        next_handler: Callable[..., AsyncGenerator[ToolChunk, None]],
    ) -> AsyncGenerator[ToolChunk, None]:
        for attempt in range(1, self.max_attempts + 1):
            try:
                async for chunk in next_handler(**input_kwargs):
                    yield chunk
                return
            except Exception as e:
                if attempt == self.max_attempts:
                    raise
                print(f"Attempt {attempt} failed: {e}, retrying…")


bash = Bash(middlewares=[LoggingMiddleware(), RetryMiddleware(max_attempts=3)])