ci: Switch Type checker to use ty from astral

.github/workflows/linter.yaml

@@ -41,14 +41,10 @@ jobs:
         id: ruff-format
         run: uv run ruff format --check
         continue-on-error: true
-      - name: Run MyPy Type Checker
-        id: mypy
+      - name: Run ty Type Checker
+        id: ty
         continue-on-error: true
-        run: uv run mypy src
-      - name: Run Pyright (Pylance equivalent)
-        id: pyright
-        continue-on-error: true
-        run: uv run pyright src
+        run: uv run ty check --output-format=github
       - name: Run JSCPD for copy-paste detection
         id: jscpd
         continue-on-error: true
@@ -60,15 +56,13 @@ jobs:
         env:
           RUFF_LINT: ${{ steps.ruff-lint.outcome }}
           RUFF_FORMAT: ${{ steps.ruff-format.outcome }}
-          MYPY: ${{ steps.mypy.outcome }}
-          PYRIGHT: ${{ steps.pyright.outcome }}
+          TY: ${{ steps.ty.outcome }}
           JSCPD: ${{ steps.jscpd.outcome }}
         run: |-
           failed=()
           [[ "$RUFF_LINT"   == "failure" ]] && failed+=("Ruff Linter")
           [[ "$RUFF_FORMAT" == "failure" ]] && failed+=("Ruff Formatter")
-          [[ "$MYPY"        == "failure" ]] && failed+=("MyPy")
-          [[ "$PYRIGHT"     == "failure" ]] && failed+=("Pyright")
+          [[ "$TY"          == "failure" ]] && failed+=("ty")
           [[ "$JSCPD"       == "failure" ]] && failed+=("JSCPD")
           if (( ${#failed[@]} )); then
             joined=$(IFS=', '; echo "${failed[*]}")

docs/ai/coding_conventions.md

@@ -3,7 +3,7 @@
 Non-negotiable rules for code quality and style.
 
 1. **Python Types**: All Python code MUST include type hints. All function definitions MUST include return types.
-2. **Type Safety**: All code MUST pass `mypy` and `pyright` checks.
+2. **Type Safety**: All code MUST pass `ty` checks.
 3. **Formatting & Linting**: All code MUST be formatted with `ruff`.
 
 #### Examples:

docs/ai/mandatory_checks.md

@@ -10,8 +10,7 @@ Exact shell commands required to test the project and fix formatting issues.
 
 2. **Type Checking**:
    ```bash
-   uv run mypy src
-   uv run pyright src
+   uv run ty check
    ```
 
 3. **Testing**:

pyproject.toml

@@ -106,7 +106,6 @@ style = "pep440"
 [dependency-groups]
 dev = [
   "fastapi>=0.115.2",
-  "mypy>=1.15.0",
   "PyJWT>=2.0.0",
   "pytest>=8.3.5",
   "pytest-asyncio>=0.26.0",
@@ -122,8 +121,8 @@ dev = [
   "trio",
   "uvicorn>=0.35.0",
   "pytest-timeout>=2.4.0",
-  "pyright",
   "a2a-sdk[all]",
+  "ty>=0.0.34",
 ]
 
 [[tool.uv.index]]
@@ -135,20 +134,12 @@ explicit = true
 [tool.uv.sources]
 a2a-sdk = { workspace = true }
 
-[tool.mypy]
-plugins = ["pydantic.mypy"]
-exclude = ["src/a2a/types/a2a_pb2\\.py", "src/a2a/types/a2a_pb2_grpc\\.py"]
-disable_error_code = [
-  "import-not-found",
-  "annotation-unchecked",
-  "import-untyped",
-]
+[tool.ty]
 
-[[tool.mypy.overrides]]
-module = "examples.*"
-follow_imports = "skip"
+[tool.ty.environment]
+python-version = "3.10"
 
-[tool.pyright]
+[tool.ty.src]
 include = ["src"]
 exclude = [
   "**/__pycache__",
@@ -161,8 +152,6 @@ exclude = [
   "src/a2a/compat/v0_3/*_pb2*.py",
   "src/a2a/compat/v0_3/proto_utils.py",
 ]
-venvPath = "."
-venv = ".venv"
 
 [tool.coverage.run]
 branch = true

scripts/lint.sh

@@ -15,7 +15,7 @@ echo -e "${BLUE}${BOLD}=== A2A Python Fixed-and-Lint Suite ===${NC}"
 echo -e "Fixing formatting and linting issues, then verifying types...\n"
 
 # 1. Ruff Linter (with fix)
-echo -e "${YELLOW}${BOLD}--- [1/4] Running Ruff Linter (fix) ---${NC}"
+echo -e "${YELLOW}${BOLD}--- [1/3] Running Ruff Linter (fix) ---${NC}"
 if uv run ruff check --fix; then
     echo -e "${GREEN}✓ Ruff Linter passed (and fixed what it could)${NC}"
 else
@@ -24,29 +24,20 @@ else
 fi
 
 # 2. Ruff Formatter
-echo -e "\n${YELLOW}${BOLD}--- [2/4] Running Ruff Formatter (apply) ---${NC}"
+echo -e "\n${YELLOW}${BOLD}--- [2/3] Running Ruff Formatter (apply) ---${NC}"
 if uv run ruff format; then
     echo -e "${GREEN}✓ Ruff Formatter applied${NC}"
 else
     echo -e "${RED}✗ Ruff Formatter failed${NC}"
     FAILED=1
 fi
 
-# 3. MyPy Type Checker
-echo -e "\n${YELLOW}${BOLD}--- [3/4] Running MyPy Type Checker ---${NC}"
-if uv run mypy src; then
-    echo -e "${GREEN}✓ MyPy passed${NC}"
+# 3. ty Type Checker
+echo -e "\n${YELLOW}${BOLD}--- [3/3] Running ty Type Checker ---${NC}"
+if uv run ty check; then
+    echo -e "${GREEN}✓ ty passed${NC}"
 else
-    echo -e "${RED}✗ MyPy failed${NC}"
-    FAILED=1
-fi
-
-# 4. Pyright Type Checker
-echo -e "\n${YELLOW}${BOLD}--- [4/4] Running Pyright ---${NC}"
-if uv run pyright; then
-    echo -e "${GREEN}✓ Pyright passed${NC}"
-else
-    echo -e "${RED}✗ Pyright failed${NC}"
+    echo -e "${RED}✗ ty failed${NC}"
     FAILED=1
 fi
 

src/a2a/client/optionals.py

@@ -12,5 +12,8 @@
         class Channel:  # type: ignore[no-redef]
             """Stub class for type hinting when grpc.aio is not available."""
 
+            async def close(self, grace: float | None = None) -> None:
+                """Stub for Channel.close."""
+
     else:
         Channel = None  # At runtime, pd will be None if the import failed.

src/a2a/client/transports/grpc.py

@@ -301,7 +301,9 @@ async def close(self) -> None:
     def _get_grpc_metadata(
         self, context: ClientCallContext | None
     ) -> list[tuple[str, str]]:
-        metadata = [(VERSION_HEADER.lower(), PROTOCOL_VERSION_CURRENT)]
+        metadata: list[tuple[str, str]] = [
+            (VERSION_HEADER.lower(), PROTOCOL_VERSION_CURRENT)
+        ]
         if context and context.service_parameters:
             for key, value in context.service_parameters.items():
                 metadata.append((key.lower(), value))

src/a2a/compat/v0_3/grpc_transport.py

@@ -359,7 +359,9 @@ def _get_grpc_metadata(
         self, context: ClientCallContext | None = None
     ) -> list[tuple[str, str]]:
         """Creates gRPC metadata for extensions."""
-        metadata = [(VERSION_HEADER.lower(), PROTOCOL_VERSION_0_3)]
+        metadata: list[tuple[str, str]] = [
+            (VERSION_HEADER.lower(), PROTOCOL_VERSION_0_3)
+        ]
 
         if context and context.service_parameters:
             params = dict(context.service_parameters)

src/a2a/compat/v0_3/jsonrpc_transport.py

@@ -398,17 +398,15 @@ async def close(self) -> None:
         """Closes the httpx client."""
         await self.httpx_client.aclose()
 
-    def _create_jsonrpc_error(
-        self, error_dict: dict[str, Any]
-    ) -> A2AClientError:
+    def _create_jsonrpc_error(self, error_dict: dict[str, Any]) -> Exception:
         """Raises a specific error based on jsonrpc error code."""
         code = error_dict.get('code')
         message = error_dict.get('message', 'Unknown Error')
 
         if isinstance(code, int):
             error_class = _JSON_RPC_ERROR_CODE_TO_A2A_ERROR.get(code)
             if error_class:
-                return error_class(message)  # type: ignore[return-value]
+                return error_class(message)
 
         return A2AClientError(message)
 

src/a2a/compat/v0_3/request_handler.py

@@ -1,5 +1,4 @@
 import logging
-import typing
 
 from collections.abc import AsyncIterable
 
@@ -148,10 +147,7 @@ async def on_list_task_push_notification_configs(
             v03_resp.root,
             types_v03.ListTaskPushNotificationConfigSuccessResponse,
         ):
-            return typing.cast(
-                'list[types_v03.TaskPushNotificationConfig]',
-                v03_resp.root.result,
-            )
+            return v03_resp.root.result
         return []
 
     async def on_delete_task_push_notification_config(

src/a2a/server/request_handlers/__init__.py

@@ -23,7 +23,7 @@
 try:
     from a2a.server.request_handlers.grpc_handler import (
         DefaultGrpcServerCallContextBuilder,
-        GrpcHandler,  # type: ignore
+        GrpcHandler,
         GrpcServerCallContextBuilder,
     )
 except ImportError as e:
@@ -33,7 +33,7 @@
         _original_error,
     )
 
-    class GrpcHandler:  # type: ignore
+    class GrpcHandler:
         """Placeholder for GrpcHandler when dependencies are not installed."""
 
         def __init__(self, *args, **kwargs):

src/a2a/server/request_handlers/grpc_handler.py

@@ -82,7 +82,7 @@ def _get_metadata_value(
     ]
 
 
-_ERROR_CODE_MAP = {
+_ERROR_CODE_MAP: dict[type[A2AError], grpc.StatusCode] = {
     types.InvalidRequestError: grpc.StatusCode.INVALID_ARGUMENT,
     types.MethodNotFoundError: grpc.StatusCode.NOT_FOUND,
     types.InvalidParamsError: grpc.StatusCode.INVALID_ARGUMENT,

src/a2a/server/routes/jsonrpc_dispatcher.py

@@ -22,9 +22,7 @@
     MethodNotFoundError,
 )
 from a2a.server.request_handlers.request_handler import RequestHandler
-from a2a.server.request_handlers.response_helpers import (
-    build_error_response,
-)
+from a2a.server.request_handlers.response_helpers import build_error_response
 from a2a.server.routes.common import (
     DefaultServerCallContextBuilder,
     ServerCallContextBuilder,
@@ -67,8 +65,8 @@
         # Starlette v0.48.0
         from starlette.status import HTTP_413_CONTENT_TOO_LARGE
     except ImportError:
-        from starlette.status import (  # type: ignore[no-redef]
-            HTTP_413_REQUEST_ENTITY_TOO_LARGE as HTTP_413_CONTENT_TOO_LARGE,
+        from starlette.status import (
+            HTTP_413_REQUEST_ENTITY_TOO_LARGE as HTTP_413_CONTENT_TOO_LARGE,  # type: ignore[no-redef]
         )
 
     _package_starlette_installed = True
@@ -597,7 +595,7 @@ async def event_generator(
                         'data': json.dumps(error_response),
                     }
 
-            return EventSourceResponse(event_generator(handler_result))
+            return EventSourceResponse(event_generator(handler_result))  # ty:ignore[invalid-argument-type]
 
         # handler_result is a dict (JSON-RPC response)
         return JSONResponse(handler_result)

src/a2a/server/tasks/__init__.py

@@ -26,7 +26,7 @@
 
 try:
     from a2a.server.tasks.database_task_store import (
-        DatabaseTaskStore,  # type: ignore
+        DatabaseTaskStore,
     )
 except ImportError as e:
     _original_error = e
@@ -36,7 +36,7 @@
         e,
     )
 
-    class DatabaseTaskStore:  # type: ignore
+    class DatabaseTaskStore:
         """Placeholder for DatabaseTaskStore when dependencies are not installed."""
 
         def __init__(self, *args, **kwargs):
@@ -48,7 +48,7 @@ def __init__(self, *args, **kwargs):
 
 try:
     from a2a.server.tasks.database_push_notification_config_store import (
-        DatabasePushNotificationConfigStore,  # type: ignore
+        DatabasePushNotificationConfigStore,
     )
 except ImportError as e:
     _original_error = e
@@ -58,7 +58,7 @@ def __init__(self, *args, **kwargs):
         e,
     )
 
-    class DatabasePushNotificationConfigStore:  # type: ignore
+    class DatabasePushNotificationConfigStore:
         """Placeholder for DatabasePushNotificationConfigStore when dependencies are not installed."""
 
         def __init__(self, *args, **kwargs):

src/a2a/server/tasks/database_push_notification_config_store.py

@@ -114,7 +114,7 @@ def __init__(  # noqa: PLR0913
         self.create_table = create_table
         self._initialized = False
         self.owner_resolver = owner_resolver
-        self.config_model = (
+        self.config_model = (  # ty:ignore[invalid-assignment]
             PushNotificationConfigModel
             if table_name == 'push_notification_configs'
             else create_push_notification_config_model(table_name)
@@ -380,10 +380,10 @@ async def delete_info(
 
             result = await session.execute(stmt)
 
-            if result.rowcount > 0:  # type: ignore[attr-defined]
+            if result.rowcount > 0:  # ty:ignore[unresolved-attribute]
                 logger.info(
                     'Deleted %s push notification config(s) for task %s, owner %s.',
-                    result.rowcount,  # type: ignore[attr-defined]
+                    result.rowcount,  # ty:ignore[unresolved-attribute]
                     task_id,
                     owner,
                 )

src/a2a/server/tasks/database_task_store.py

@@ -89,7 +89,7 @@ def __init__(  # noqa: PLR0913
         self.core_to_model_conversion = core_to_model_conversion
         self.model_to_core_conversion = model_to_core_conversion
 
-        self.task_model = (
+        self.task_model = (  # ty:ignore[invalid-assignment]
             TaskModel
             if table_name == 'tasks'
             else create_task_model(table_name)
@@ -328,7 +328,7 @@ async def delete(self, task_id: str, context: ServerCallContext) -> None:
             result = await session.execute(stmt)
             # Commit is automatic when using session.begin()
 
-            if result.rowcount > 0:  # type: ignore[attr-defined]
+            if result.rowcount > 0:  # ty:ignore[unresolved-attribute]
                 logger.info(
                     'Task %s deleted successfully for owner %s.', task_id, owner
                 )

src/a2a/utils/signing.py

@@ -75,7 +75,7 @@ def agent_card_signer(agent_card: AgentCard) -> AgentCard:
             payload=payload_dict,
             key=signing_key,
             algorithm=protected_header.get('alg', 'HS256'),
-            headers=dict(protected_header),
+            headers=dict(protected_header),  # ty:ignore[no-matching-overload]
         )
 
         # The result of jwt.encode is a compact serialization: HEADER.PAYLOAD.SIGNATURE

src/a2a/utils/telemetry.py

@@ -140,7 +140,7 @@ def __getattr__(self, name: str) -> Any:
     _SpanKind = _NoOp()  # type: ignore
     StatusCode = _NoOp()  # type: ignore
 
-SpanKind = _SpanKind  # type: ignore
+SpanKind = _SpanKind
 __all__ = ['SpanKind']
 
 
@@ -204,7 +204,7 @@ def trace_function(  # noqa: PLR0915
             attribute_extractor=attribute_extractor,
         )
 
-    actual_span_name = span_name or f'{func.__module__}.{func.__name__}'
+    actual_span_name = span_name or f'{func.__module__}.{func.__name__}'  # ty:ignore[unresolved-attribute]
 
     is_async_func = inspect.iscoroutinefunction(func)