chore(typing): replace deprecated and legacy types

docs/tutorial/en/src/task_eval.py

@@ -123,7 +123,7 @@ async def __call__(
 # - A ``Task`` is a unit in the benchmark that includes all information for the agent to execute and evaluate (e.g., input/query and its ground truth).
 # - A ``Benchmark`` organizes multiple tasks for systematic evaluation.
 
-from typing import Generator
+from collections.abc import Generator
 from agentscope.evaluate import (
     Task,
     BenchmarkBase,
@@ -186,7 +186,7 @@ def __len__(self) -> int:
 
 import os
 import asyncio
-from typing import Callable
+from collections.abc import Callable
 from pydantic import BaseModel
 
 from agentscope.message import Msg

docs/tutorial/en/src/task_eval_openjudge.py

@@ -194,7 +194,7 @@ async def __call__(self, solution: SolutionOutput) -> MetricResult:
 
 # %%
 import os
-from typing import Generator
+from collections.abc import Generator
 from openjudge.graders.common.relevance import RelevanceGrader
 from openjudge.graders.common.correctness import CorrectnessGrader
 from agentscope.evaluate import (
@@ -284,7 +284,7 @@ def __len__(self) -> int:
 
 # %%
 
-from typing import Callable
+from collections.abc import Callable
 
 from agentscope.agent import ReActAgent
 from agentscope.evaluate import GeneralEvaluator

docs/tutorial/en/src/task_hook.py

@@ -62,7 +62,7 @@
 
 """
 import asyncio
-from typing import Any, Type
+from typing import Any
 
 from agentscope.agent import ReActAgentBase, AgentBase
 from agentscope.message import Msg

docs/tutorial/en/src/task_middleware.py

@@ -20,7 +20,7 @@
 
 """
 import asyncio
-from typing import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator, Callable
 
 from agentscope.message import TextBlock, ToolUseBlock
 from agentscope.tool import ToolResponse, Toolkit

docs/tutorial/en/src/task_tool.py

@@ -21,7 +21,8 @@
 import asyncio
 import inspect
 import json
-from typing import Any, AsyncGenerator
+from collections.abc import AsyncGenerator
+from typing import Any
 
 from pydantic import BaseModel, Field
 

docs/tutorial/en/src/task_tuner.py

@@ -75,7 +75,6 @@
 Below is an example workflow function using a ReAct agent to answer math questions:
 """
 
-from typing import Dict, Optional
 from agentscope.agent import ReActAgent
 from agentscope.formatter import OpenAIChatFormatter
 from agentscope.message import Msg
@@ -84,16 +83,16 @@
 
 
 async def example_workflow_function(
-    task: Dict,
+    task: dict,
     model: ChatModelBase,
-    auxiliary_models: Optional[Dict[str, ChatModelBase]] = None,
+    auxiliary_models: dict[str, ChatModelBase] | None = None,
 ) -> WorkflowOutput:
     """An example workflow function for tuning.
 
     Args:
-        task (`Dict`): The task information.
+        task (`dict`): The task information.
         model (`ChatModelBase`): The chat model used by the agent.
-        auxiliary_models (`Optional[Dict[str, ChatModelBase]]`): Additional
+        auxiliary_models (`dict[str, ChatModelBase] | None`): Additional
             chat models, generally used to simulate the behavior of other
             non-training agents in multi-agent scenarios.
 
@@ -152,16 +151,16 @@ async def example_workflow_function(
 
 
 async def example_judge_function(
-    task: Dict,
+    task: dict,
     response: Any,
-    auxiliary_models: Optional[Dict[str, ChatModelBase]] = None,
+    auxiliary_models: dict[str, ChatModelBase] | None = None,
 ) -> JudgeOutput:
     """A very simple judge function only for demonstration.
 
     Args:
-        task (`Dict`): The task information.
+        task (`dict`): The task information.
         response (`Any`): The response field from the WorkflowOutput.
-        auxiliary_models (`Optional[Dict[str, ChatModelBase]]`): Additional
+        auxiliary_models (`dict[str, ChatModelBase] | None`): Additional
             chat models for LLM-as-a-Judge purpose.
     Returns:
         `JudgeOutput`: The reward assigned by the judge.

docs/tutorial/zh_CN/src/task_eval.py

@@ -122,7 +122,7 @@ async def __call__(
 # - 一个 ``Task`` (任务) 是基准测试中的一个单元，包含智能体执行和评估所需的所有信息（例如，输入/查询及其标准答案）。
 # - 一个 ``Benchmark`` (基准测试) 组织多个任务进行系统性评估。
 
-from typing import Generator
+from collections.abc import Generator
 from agentscope.evaluate import (
     Task,
     BenchmarkBase,
@@ -184,7 +184,7 @@ def __len__(self) -> int:
 
 import os
 import asyncio
-from typing import Callable
+from collections.abc import Callable
 from pydantic import BaseModel
 
 from agentscope.message import Msg

docs/tutorial/zh_CN/src/task_eval_openjudge.py

@@ -201,7 +201,7 @@ async def __call__(self, solution: SolutionOutput) -> MetricResult:
 
 # %%
 import os
-from typing import Generator
+from collections.abc import Generator
 from openjudge.graders.common.relevance import RelevanceGrader
 from openjudge.graders.common.correctness import CorrectnessGrader
 from agentscope.evaluate import (
@@ -286,7 +286,7 @@ def __len__(self) -> int:
 # %%
 
 import asyncio
-from typing import Callable
+from collections.abc import Callable
 
 from agentscope.agent import ReActAgent
 from agentscope.evaluate import GeneralEvaluator

docs/tutorial/zh_CN/src/task_hook.py

@@ -60,7 +60,7 @@
 
 """
 import asyncio
-from typing import Any, Type
+from typing import Any
 
 from agentscope.agent import ReActAgentBase, AgentBase
 from agentscope.message import Msg

docs/tutorial/zh_CN/src/task_middleware.py

@@ -20,7 +20,7 @@
 
 """
 import asyncio
-from typing import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator, Callable
 
 from agentscope.message import TextBlock, ToolUseBlock
 from agentscope.tool import ToolResponse, Toolkit

docs/tutorial/zh_CN/src/task_tool.py

@@ -21,7 +21,8 @@
 import asyncio
 import inspect
 import json
-from typing import Any, AsyncGenerator
+from collections.abc import AsyncGenerator
+from typing import Any
 
 from pydantic import BaseModel, Field
 

docs/tutorial/zh_CN/src/task_tuner.py

@@ -75,7 +75,6 @@
 以下是一个用 ReAct 智能体回答数学问题的简单工作流函数示例：
 """
 
-from typing import Dict, Optional
 from agentscope.agent import ReActAgent
 from agentscope.formatter import OpenAIChatFormatter
 from agentscope.message import Msg
@@ -84,16 +83,16 @@
 
 
 async def example_workflow_function(
-    task: Dict,
+    task: dict,
     model: ChatModelBase,
-    auxiliary_models: Optional[Dict[str, ChatModelBase]] = None,
+    auxiliary_models: dict[str, ChatModelBase] | None = None,
 ) -> WorkflowOutput:
     """一个用于调优的工作流函数示例。
 
     Args:
-        task (`Dict`): 任务信息。
+        task (`dict`): 任务信息。
         model (`ChatModelBase`): 智能体使用的对话模型。
-        auxiliary_models (`Optional[Dict[str, ChatModelBase]]`):
+        auxiliary_models (`dict[str, ChatModelBase] | None`):
             用于辅助的额外对话模型，一般用于多智能体场景下模拟其他非训练智能体的行为。
 
     Returns:
@@ -151,16 +150,16 @@ async def example_workflow_function(
 
 
 async def example_judge_function(
-    task: Dict,
+    task: dict,
     response: Any,
-    auxiliary_models: Optional[Dict[str, ChatModelBase]] = None,
+    auxiliary_models: dict[str, ChatModelBase] | None = None,
 ) -> JudgeOutput:
     """仅用于演示的简单评判函数。
 
     Args:
-        task (`Dict`): 任务信息。
+        task (`dict`): 任务信息。
         response (`Any`): WorkflowOutput 的响应字段。
-        auxiliary_models (`Optional[Dict[str, ChatModelBase]]`):
+        auxiliary_models (`dict[str, ChatModelBase] | None`):
             用于 LLM-as-a-Judge 的辅助模型。
     Returns:
         `JudgeOutput`: 评判函数分配的奖励。

examples/agent/a2a_agent/setup_a2a_server.py

@@ -2,7 +2,8 @@
 """Set up an A2A server with a ReAct agent to handle the input query"""
 import os
 import uuid
-from typing import AsyncGenerator, Any
+from collections.abc import AsyncGenerator
+from typing import Any
 
 from agent_card import agent_card
 

examples/agent/a2ui_agent/samples/general_agent/setup_a2ui_server.py

@@ -3,7 +3,8 @@
 import os
 import uuid
 import copy
-from typing import AsyncGenerator, Any
+from collections.abc import AsyncGenerator
+from typing import Any
 
 from a2a.server.apps import A2AStarletteApplication
 from a2a.server.events import Event

examples/agent/browser_agent/browser_agent.py

@@ -10,7 +10,7 @@
 import json
 import inspect
 from functools import wraps
-from typing import Type, Optional, Any, Literal
+from typing import Any, Literal
 import asyncio
 import copy
 
@@ -104,7 +104,7 @@ def __init__(
         toolkit: Toolkit,
         sys_prompt: str = _BROWSER_AGENT_DEFAULT_SYS_PROMPT,
         max_iters: int = 50,
-        start_url: Optional[str] = "https://www.google.com",
+        start_url: str | None = "https://www.google.com",
         pure_reasoning_prompt: str = _BROWSER_AGENT_DEFAULT_PURE_REASONING_PROMPT,
         observe_reasoning_prompt: str = _BROWSER_AGENT_DEFAULT_OBSERVE_REASONING_PROMPT,
         task_decomposition_prompt: str = _BROWSER_AGENT_DEFAULT_TASK_DECOMPOSITION_PROMPT,
@@ -130,7 +130,7 @@ def __init__(
         self.iter_n = 0
         self.finish_function_name = "browser_generate_final_response"
         self.init_query = ""
-        self._required_structured_model: Type[BaseModel] | None = None
+        self._required_structured_model: type[BaseModel] | None = None
 
         super().__init__(
             name=name,
@@ -165,7 +165,7 @@ def __init__(
     async def reply(  # pylint: disable=R0912,R0915
         self,
         msg: Msg | list[Msg] | None = None,
-        structured_model: Type[BaseModel] | None = None,
+        structured_model: type[BaseModel] | None = None,
     ) -> Msg:
         """Process a message and return a response."""
         self.init_query = (
@@ -418,7 +418,7 @@ async def _summarize_mem(self) -> None:
 
     async def _build_observation(self) -> Msg:
         """Get a snapshot (and optional screenshot) before reasoning."""
-        image_data: Optional[str] = None
+        image_data: str | None = None
         if self._supports_multimodal():
             image_data = await self._get_screenshot()
         observe_msg = self.observe_by_chunk(image_data)
@@ -743,7 +743,7 @@ async def _memory_summarizing(self) -> None:
         for m in summarized_memory:
             await self.memory.add(m)
 
-    async def _get_screenshot(self) -> Optional[str]:
+    async def _get_screenshot(self) -> str | None:
         """
         Optionally take a screenshot of the current web page for multimodal prompts.
         Returns base64-encoded PNG data if available, else None.

examples/agent/browser_agent/build_in_helper/_video_understanding.py

@@ -13,7 +13,7 @@
 import uuid
 from base64 import b64encode
 from pathlib import Path
-from typing import Any, List, Optional
+from typing import Any
 
 from agentscope.message import (
     Base64Source,
@@ -137,7 +137,7 @@ def extract_frames(
     video_path: str,
     output_dir: str,
     max_frames: int = 16,
-) -> List[str]:
+) -> list[str]:
     """Extract representative frames using ffmpeg (no OpenCV dependency)."""
 
     if max_frames <= 0:
@@ -237,7 +237,7 @@ def extract_audio(video_path: str, audio_path: str) -> str:
     return audio_path
 
 
-def _probe_video_duration(video_path: str) -> Optional[float]:
+def _probe_video_duration(video_path: str) -> float | None:
     """Return the video duration in seconds using ffprobe, if available."""
 
     command = [
@@ -269,7 +269,7 @@ def _probe_video_duration(video_path: str) -> Optional[float]:
 
 
 def _build_multimodal_blocks(
-    frames: List[str],
+    frames: list[str],
     transcript: str,
     task: str,
 ) -> list:

examples/agent/deep_research_agent/deep_research_agent.py

@@ -4,7 +4,7 @@
 import os
 import json
 
-from typing import Type, Optional, Any, Tuple
+from typing import Any
 from datetime import datetime
 from copy import deepcopy
 import shortuuid
@@ -57,8 +57,8 @@ class SubTaskItem(BaseModel):
     """Subtask item of deep research agent."""
 
     objective: str
-    working_plan: Optional[str] = None
-    knowledge_gaps: Optional[str] = None
+    working_plan: str | None = None
+    knowledge_gaps: str | None = None
 
 
 class DeepResearchAgent(ReActAgent):
@@ -201,7 +201,7 @@ async def _ensure_mcp_initialized(self) -> None:
     async def reply(
         self,
         msg: Msg | list[Msg] | None = None,
-        structured_model: Type[BaseModel] | None = None,
+        structured_model: type[BaseModel] | None = None,
     ) -> Msg:
         """The reply method of the agent."""
         # Ensure MCP client is initialized before processing
@@ -435,7 +435,7 @@ async def _acting(self, tool_call: ToolUseBlock) -> Msg | None:
     async def get_model_output(
         self,
         msgs: list,
-        format_template: Type[BaseModel] = None,
+        format_template: type[BaseModel] = None,
         stream: bool = True,
     ) -> Any:
         """
@@ -478,7 +478,7 @@ async def call_specific_tool(
         self,
         func_name: str,
         params: dict = None,
-    ) -> Tuple[Msg, Msg]:
+    ) -> tuple[Msg, Msg]:
         """
         Call the specific tool in toolkit.
 
@@ -868,7 +868,7 @@ async def summarize_intermediate_results(self) -> ToolResponse:
     async def _generate_deepresearch_report(
         self,
         checklist: str,
-    ) -> Tuple[Msg, str]:
+    ) -> tuple[Msg, str]:
         """Collect and polish all draft reports into a final report.
 
         Args: