Conversation
* feat: 将 MessageSession 的 platform_id 改为 init=False，实例化时无需传入 Co-authored-by: aider (openai/gpt-5.2) <aider@aider.chat> * refactor: 将 isinstance 检查改为元组、将默认模型值设为空字符串、将类型注解改为 Any 并导入 * refactor: 为 _serialize_job 增加返回类型注解 dict * fix: 使用 cast 获取百度 AIP 的 msg 并对 psutil_addr 引入 type: ignore Co-authored-by: aider (openai/gpt-5.2) <aider@aider.chat> * refactor: 引入 _AddrWithPort 协议并替换 conn.laddr 的 cast Co-authored-by: aider (openai/gpt-5.2) <aider@aider.chat> * fix: 在构建 AstrBotMessage 时对 ctx.channel 可能为 None 进行兜底处理 Co-authored-by: aider (openai/gpt-5.2) <aider@aider.chat> --------- Co-authored-by: aider (openai/gpt-5.2) <aider@aider.chat>
* Fix TypeError when MCP schema type is a list Fixes crash in Gemini native tools with VRChat MCP. * Refactor: avoid modifying schema in place per feedback * Fix formatting and cleanup comments
Removed duplicate text and added a new image.
dosubot
bot
added
the
size:XL
dosubot
bot
added
the
area:platform
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - 我发现了 3 个问题，并给出了一些整体性的反馈：
- 新的
Response.to_json()辅助方法已经返回的是 Quart 的 response，因此像auth_middleware._unauthorized和srv_plug_route这样的地方应该直接返回Response().error(...).to_json()，而不是再用jsonify(...)包一层，否则会对 response 对象本身进行 JSON 编码，而不是对负载进行编码。 - 在
get_local_ip_addresses中，注释写的是应该过滤掉 IPv6 的 link-local 地址，但当前实现会把所有 IPv6 地址都 append 进去；建议在那里的逻辑中显式跳过ip.is_link_local，让行为和注释保持一致，同时避免后续再额外过滤。 - 将
MessageSession.platform_id改为field(init=False)意味着它在任何使用之前（例如在__post_init__或工厂函数中）必须先被赋值；否则__str__以及其他访问处可能会触发AttributeError，因此最好集中在一个地方完成它的初始化。
给 AI Agent 的提示
Please address the comments from this code review:
## Overall Comments
- 新的 `Response.to_json()` 辅助方法已经返回的是 Quart 的 response，因此像 `auth_middleware._unauthorized` 和 `srv_plug_route` 这样的地方应该直接返回 `Response().error(...).to_json()`，而不是再用 `jsonify(...)` 包一层，否则会对 response 对象本身进行 JSON 编码，而不是对负载进行编码。
- 在 `get_local_ip_addresses` 中，注释写的是应该过滤掉 IPv6 的 link-local 地址，但当前实现会把所有 IPv6 地址都 append 进去；建议在那里的逻辑中显式跳过 `ip.is_link_local`，让行为和注释保持一致，同时避免后续再额外过滤。
- 将 `MessageSession.platform_id` 改为 `field(init=False)` 意味着它在任何使用之前（例如在 `__post_init__` 或工厂函数中）必须先被赋值；否则 `__str__` 以及其他访问处可能会触发 `AttributeError`，因此最好集中在一个地方完成它的初始化。
## Individual Comments
### Comment 1
<location> `astrbot/dashboard/server.py:187-176` </location>
<code_context>
+ async def srv_plug_route(self, subpath: str, *args, **kwargs):
+ handler = self._plugin_route_map.get((f"/{subpath}", request.method))
+ if not handler:
+ return jsonify(Response().error("未找到该路由").to_json())
+
try:
</code_context>
<issue_to_address>
**issue (bug_risk):** 避免对已经 JSON 化的 Response 再次用 jsonify() 包裹。
`Response.to_json()` 已经返回了一个 Quart `Response`（内部是 `jsonify(asdict(self))`）。调用 `jsonify(Response().error(...).to_json())` 相当于对一个已经是 response 的对象再次调用 `jsonify`，这可能损坏响应体或引发类型错误。
相应地，应该直接返回这个 `Response`，例如：
```python
# in srv_plug_route
if not handler:
return Response().error("未找到该路由").to_json()
# in _unauthorized
@staticmethod
def _unauthorized(msg: str):
r = Response().error(msg).to_json()
r.status_code = 401
return r
```
这样可以保持抽象的一致性，并避免嵌套的 JSON 响应。
</issue_to_address>
### Comment 2
<location> `astrbot/core/utils/io.py:230-233` </location>
<code_context>
- network_ips.append(addr.address)
+ if addr.family == socket.AF_INET:
+ network_ips.append(ip_address(addr.address))
+ elif addr.family == socket.AF_INET6:
+ # 过滤掉 IPv6 的 link-local 地址（fe80:...）
+ # 用这个不如用::1
+ ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况
+ network_ips.append(ip)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 实际上 IPv6 的 link-local 地址并没有被过滤掉，这和注释不一致，并且可能在不经意间暴露给调用方。
当前对 IPv6 的处理仍然会 append link-local 地址：
```python
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
ip = ip_address(addr.address.split("%")[0])
network_ips.append(ip)
```
如果确实需要排除 link-local 地址，可以考虑：
```python
ip = ip_address(addr.address.split("%")[0])
if not ip.is_link_local:
network_ips.append(ip)
```
否则，需要更新注释，明确说明是有意保留 IPv6 link-local 地址的，这样调用方就可以据此依赖文档描述的行为。
建议的实现：
```python
if addr.family == socket.AF_INET:
network_ips.append(ip_address(addr.address))
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
# 用这个不如用::1
ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况
if not ip.is_link_local:
network_ips.append(ip)
```
如果在本文件（或代码库中其他位置）还有类似的 IPv6 处理模式但没有检查 `ip.is_link_local`，也应该一并加上相同的条件：先用 `ip_address(...split("%")[0])` 解析，再在 `not ip.is_link_local` 时 append，以保持行为一致。
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/server.py:231` </location>
<code_context>
- port = self.core_lifecycle.astrbot_config["dashboard"].get("port", 6185)
- host = self.core_lifecycle.astrbot_config["dashboard"].get("host", "0.0.0.0")
- enable = self.core_lifecycle.astrbot_config["dashboard"].get("enable", True)
+ def run(self) -> None:
+ cfg = self.config.get("dashboard", {})
+ _port: str = os.environ.get("DASHBOARD_PORT") or cfg.get("port", 6185)
</code_context>
<issue_to_address>
**issue (complexity):** 建议把 IPv6 绑定逻辑、本地主机端口检查和 IP 格式化拆分到几个小工具函数中，让 `run()` 及其相关方法保持简单，聚焦在高层行为上。
你可以保留现在所有的新行为（IPv6、双栈绑定、更丰富的日志），但通过更紧凑的封装，以及移除不必要的“host 感知”逻辑来减少心智负担。
### 1. 用一个 helper 隐藏 Hypercorn 绑定的复杂度
目前 `run()` 需要关心 `_build_bind()`、IPv4/IPv6，以及双栈的兼容处理。可以保留这些细节，但把它们挪到一个低层 helper 中，只返回简单字符串：
```python
def _get_binds(self, host: str, port: int) -> list[str]:
# normalize for bind construction
host = host.strip("[]")
def _fmt(h: str) -> str:
# only add [] when actually needed for IPv6
return f"[{h}]:{port}" if ":" in h else f"{h}:{port}"
binds = [_fmt(host)]
# dual-stack workaround stays here, not in run()
if host == "::" and platform.system() in ("Windows", "Darwin"):
binds.append(_fmt("0.0.0.0"))
return binds
```
这样 `run()` 可以简化为：
```python
config = HyperConfig()
config.bind = self._get_binds(host, port)
```
有了这个 helper，就可以去掉 `_build_bind()`，也不需要在更高层代码中传递 `IPv4Address | IPv6Address` 和 `ipaddress.ip_address()`。
### 2. 把 `check_port_in_use` 简化回“检查 localhost 上的端口”
`run()` 始终调用的是 `check_port_in_use("127.0.0.1", port)`，因此额外的 `host` 参数、协议簇检测以及 IPv6 分支并没有带来实质收益，反而增加了认知负担。
可以保留现有的行为（“检查本机上某端口是否被占用”），但把函数签名和实现简化：
```python
def check_port_in_use(self, port: int) -> bool:
"""Check if port is in use on localhost."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(2)
return sock.connect_ex(("127.0.0.1", port)) == 0
except Exception:
# be conservative on error
return True
```
在 `run()` 中的用法也保持简单：
```python
if self.check_port_in_use(port):
info = self.get_process_using_port(port)
raise RuntimeError(f"端口 {port} 已被占用\n{info}")
```
这样可以让函数聚焦在它真正的使用场景上，又不丢失任何行为。
### 3. 把 IP 相关格式化逻辑集中到一个小 helper 中
目前 `_print_access_urls()` 需要了解 `IPv4Address | IPv6Address`、`.is_loopback`、`.is_link_local`、`.version` 和 URL 格式化等细节。你可以继续让 `get_local_ip_addresses()` 返回 `ipaddress` 的对象，但把所有 IP 相关的分支集中在一个小 helper 中：
```python
def _iter_network_urls(self, port: int):
for ip in get_local_ip_addresses():
# filter once here, not in multiple places
if ip.is_loopback or ip.is_link_local:
continue
host = f"[{ip}]" if ip.version == 6 else str(ip)
yield f"http://{host}:{port}"
```
这样 `_print_access_urls()` 就更容易阅读：
```python
def _print_access_urls(self, host: str, port: int) -> None:
parts = [f"\n ✨✨✨\n AstrBot v{VERSION} WebUI 已启动\n\n"]
parts.append(f" ➜ 本地: http://localhost:{port}\n")
# only enumerate network URLs when binding all interfaces
if host in ("::", "0.0.0.0"):
for url in self._iter_network_urls(port):
parts.append(f" ➜ 网络: {url}\n")
parts.append(" ➜ 默认用户名和密码: astrbot\n ✨✨✨\n")
logger.info("".join(parts))
```
这样可以保持当前所有行为（IPv6 URL、过滤等），但把底层地址处理集中在一个地方，而不是分散在 `_print_access_urls()` 的多处。
---
这些改动可以在保留你新增的 IPv6 支持和改进结构的同时，减少必须理解 socket、操作系统差异和 `ipaddress` 类型的代码位置。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进 Review 质量。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- The new
Response.to_json()helper already returns a Quart response, so places likeauth_middleware._unauthorizedandsrv_plug_routeshould returnResponse().error(...).to_json()directly instead of wrapping it again withjsonify(...), which will otherwise JSON-encode a response object rather than the payload. - In
get_local_ip_addresses, the comment says IPv6 link-local addresses should be filtered out, but the implementation currently appends all IPv6 addresses; consider explicitly skippingip.is_link_localthere to keep the behavior and comment aligned and avoid later re-filtering. - Changing
MessageSession.platform_idtofield(init=False)means it must now be set before any use (e.g., in__post_init__or by factory functions); otherwise__str__and other accesses may hit anAttributeError, so it would be good to centralize its initialization.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new `Response.to_json()` helper already returns a Quart response, so places like `auth_middleware._unauthorized` and `srv_plug_route` should return `Response().error(...).to_json()` directly instead of wrapping it again with `jsonify(...)`, which will otherwise JSON-encode a response object rather than the payload.
- In `get_local_ip_addresses`, the comment says IPv6 link-local addresses should be filtered out, but the implementation currently appends all IPv6 addresses; consider explicitly skipping `ip.is_link_local` there to keep the behavior and comment aligned and avoid later re-filtering.
- Changing `MessageSession.platform_id` to `field(init=False)` means it must now be set before any use (e.g., in `__post_init__` or by factory functions); otherwise `__str__` and other accesses may hit an `AttributeError`, so it would be good to centralize its initialization.
## Individual Comments
### Comment 1
<location> `astrbot/dashboard/server.py:187-176` </location>
<code_context>
+ async def srv_plug_route(self, subpath: str, *args, **kwargs):
+ handler = self._plugin_route_map.get((f"/{subpath}", request.method))
+ if not handler:
+ return jsonify(Response().error("未找到该路由").to_json())
+
try:
</code_context>
<issue_to_address>
**issue (bug_risk):** Avoid double-wrapping Response JSON by calling jsonify() on an already-jsonified value.
`Response.to_json()` already returns a Quart `Response` (`jsonify(asdict(self))`). Calling `jsonify(Response().error(...).to_json())` therefore wraps a response in another `jsonify` call, which can corrupt the body or raise type errors.
Instead, return the `Response` directly, e.g.:
```python
# in srv_plug_route
if not handler:
return Response().error("未找到该路由").to_json()
# in _unauthorized
@staticmethod
def _unauthorized(msg: str):
r = Response().error(msg).to_json()
r.status_code = 401
return r
```
This keeps the abstraction consistent and avoids nested JSON responses.
</issue_to_address>
### Comment 2
<location> `astrbot/core/utils/io.py:230-233` </location>
<code_context>
- network_ips.append(addr.address)
+ if addr.family == socket.AF_INET:
+ network_ips.append(ip_address(addr.address))
+ elif addr.family == socket.AF_INET6:
+ # 过滤掉 IPv6 的 link-local 地址（fe80:...）
+ # 用这个不如用::1
+ ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况
+ network_ips.append(ip)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** IPv6 link-local addresses are not actually filtered out despite the comment and may leak into callers unintentionally.
Current IPv6 handling still appends link-local addresses:
```python
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
ip = ip_address(addr.address.split("%")[0])
network_ips.append(ip)
```
If link-local addresses should be excluded, consider:
```python
ip = ip_address(addr.address.split("%")[0])
if not ip.is_link_local:
network_ips.append(ip)
```
Otherwise, update the comment to make it clear that link-local IPv6 addresses are intentionally included, so callers can rely on the documented behavior.
Suggested implementation:
```python
if addr.family == socket.AF_INET:
network_ips.append(ip_address(addr.address))
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
# 用这个不如用::1
ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况
if not ip.is_link_local:
network_ips.append(ip)
```
If there are any other places in this file (or elsewhere in the codebase) that follow the same IPv6 pattern without checking `ip.is_link_local`, you should apply the same condition there to keep behavior consistent: parse with `ip_address(...split("%")[0])` and only append when `not ip.is_link_local`.
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/server.py:231` </location>
<code_context>
- port = self.core_lifecycle.astrbot_config["dashboard"].get("port", 6185)
- host = self.core_lifecycle.astrbot_config["dashboard"].get("host", "0.0.0.0")
- enable = self.core_lifecycle.astrbot_config["dashboard"].get("enable", True)
+ def run(self) -> None:
+ cfg = self.config.get("dashboard", {})
+ _port: str = os.environ.get("DASHBOARD_PORT") or cfg.get("port", 6185)
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating the IPv6 bind logic, localhost port check, and IP formatting into small helpers so `run()` and related methods stay straightforward and focused on high-level behavior.
You can keep all the new behavior (IPv6, dual-stack bind, richer logging) but reduce the mental surface area by tightening encapsulation and removing unused “host-awareness” where it doesn’t add value.
### 1. Hide Hypercorn bind complexity behind a single helper
`run()` currently has to know about `_build_bind()`, IPv4/IPv6, and the dual-stack workaround. You can keep all of that but move it into one low-level helper that returns plain strings:
```python
def _get_binds(self, host: str, port: int) -> list[str]:
# normalize for bind construction
host = host.strip("[]")
def _fmt(h: str) -> str:
# only add [] when actually needed for IPv6
return f"[{h}]:{port}" if ":" in h else f"{h}:{port}"
binds = [_fmt(host)]
# dual-stack workaround stays here, not in run()
if host == "::" and platform.system() in ("Windows", "Darwin"):
binds.append(_fmt("0.0.0.0"))
return binds
```
Then `run()` simplifies to:
```python
config = HyperConfig()
config.bind = self._get_binds(host, port)
```
With this in place, you can drop `_build_bind()` and avoid threading `IPv4Address | IPv6Address` and `ipaddress.ip_address()` through higher-level code.
### 2. Simplify `check_port_in_use` back to a “port on localhost” check
`run()` always calls `check_port_in_use("127.0.0.1", port)`, so the extra `host` parameter, family detection, and IPv6 branch don’t bring practical benefit but add cognitive load.
You can keep the behavior (“is this port occupied on the local machine”) and simplify the signature:
```python
def check_port_in_use(self, port: int) -> bool:
"""Check if port is in use on localhost."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(2)
return sock.connect_ex(("127.0.0.1", port)) == 0
except Exception:
# be conservative on error
return True
```
Usage in `run()` stays simple:
```python
if self.check_port_in_use(port):
info = self.get_process_using_port(port)
raise RuntimeError(f"端口 {port} 已被占用\n{info}")
```
This focuses the function on its actual use case without losing any behavior.
### 3. Localize IP-specific formatting into a small helper
Right now `_print_access_urls()` needs to know about `IPv4Address | IPv6Address`, `.is_loopback`, `.is_link_local`, `.version` and URL formatting. You can keep `get_local_ip_addresses()` returning `ipaddress` objects but isolate all IP-specific branching in a tiny helper:
```python
def _iter_network_urls(self, port: int):
for ip in get_local_ip_addresses():
# filter once here, not in multiple places
if ip.is_loopback or ip.is_link_local:
continue
host = f"[{ip}]" if ip.version == 6 else str(ip)
yield f"http://{host}:{port}"
```
Then `_print_access_urls()` becomes much easier to read:
```python
def _print_access_urls(self, host: str, port: int) -> None:
parts = [f"\n ✨✨✨\n AstrBot v{VERSION} WebUI 已启动\n\n"]
parts.append(f" ➜ 本地: http://localhost:{port}\n")
# only enumerate network URLs when binding all interfaces
if host in ("::", "0.0.0.0"):
for url in self._iter_network_urls(port):
parts.append(f" ➜ 网络: {url}\n")
parts.append(" ➜ 默认用户名和密码: astrbot\n ✨✨✨\n")
logger.info("".join(parts))
```
This keeps all current behavior (IPv6 URLs, filtering, etc.) but moves the low-level address handling into a single place instead of spreading it across `_print_access_urls()`.
---
These changes keep your new IPv6 support and improved structure while reducing the number of places that need to understand sockets, OS quirks, and `ipaddress` types.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
| @staticmethod | ||
| def _unauthorized(msg: str): | ||
| r = jsonify(Response().error(msg).to_json()) |
There was a problem hiding this comment.
issue (bug_risk): 避免对已经 JSON 化的 Response 再次用 jsonify() 包裹。
Response.to_json() 已经返回了一个 Quart Response（内部是 jsonify(asdict(self))）。调用 jsonify(Response().error(...).to_json()) 相当于对一个已经是 response 的对象再次调用 jsonify，这可能损坏响应体或引发类型错误。
相应地，应该直接返回这个 Response，例如：
# in srv_plug_route
if not handler:
return Response().error("未找到该路由").to_json()
# in _unauthorized
@staticmethod
def _unauthorized(msg: str):
r = Response().error(msg).to_json()
r.status_code = 401
return r这样可以保持抽象的一致性，并避免嵌套的 JSON 响应。
Original comment in English
issue (bug_risk): Avoid double-wrapping Response JSON by calling jsonify() on an already-jsonified value.
Response.to_json() already returns a Quart Response (jsonify(asdict(self))). Calling jsonify(Response().error(...).to_json()) therefore wraps a response in another jsonify call, which can corrupt the body or raise type errors.
Instead, return the Response directly, e.g.:
# in srv_plug_route
if not handler:
return Response().error("未找到该路由").to_json()
# in _unauthorized
@staticmethod
def _unauthorized(msg: str):
r = Response().error(msg).to_json()
r.status_code = 401
return rThis keeps the abstraction consistent and avoids nested JSON responses.
| elif addr.family == socket.AF_INET6: | ||
| # 过滤掉 IPv6 的 link-local 地址（fe80:...） | ||
| # 用这个不如用::1 | ||
| ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况 |
There was a problem hiding this comment.
suggestion (bug_risk): 实际上 IPv6 的 link-local 地址并没有被过滤掉，这和注释不一致，并且可能在不经意间暴露给调用方。
当前对 IPv6 的处理仍然会 append link-local 地址：
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
ip = ip_address(addr.address.split("%")[0])
network_ips.append(ip)如果确实需要排除 link-local 地址，可以考虑：
ip = ip_address(addr.address.split("%")[0])
if not ip.is_link_local:
network_ips.append(ip)否则，需要更新注释，明确说明是有意保留 IPv6 link-local 地址的，这样调用方就可以据此依赖文档描述的行为。
建议的实现：
if addr.family == socket.AF_INET:
network_ips.append(ip_address(addr.address))
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
# 用这个不如用::1
ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况
if not ip.is_link_local:
network_ips.append(ip)如果在本文件（或代码库中其他位置）还有类似的 IPv6 处理模式但没有检查 ip.is_link_local，也应该一并加上相同的条件：先用 ip_address(...split("%")[0]) 解析，再在 not ip.is_link_local 时 append，以保持行为一致。
Original comment in English
suggestion (bug_risk): IPv6 link-local addresses are not actually filtered out despite the comment and may leak into callers unintentionally.
Current IPv6 handling still appends link-local addresses:
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
ip = ip_address(addr.address.split("%")[0])
network_ips.append(ip)If link-local addresses should be excluded, consider:
ip = ip_address(addr.address.split("%")[0])
if not ip.is_link_local:
network_ips.append(ip)Otherwise, update the comment to make it clear that link-local IPv6 addresses are intentionally included, so callers can rely on the documented behavior.
Suggested implementation:
if addr.family == socket.AF_INET:
network_ips.append(ip_address(addr.address))
elif addr.family == socket.AF_INET6:
# 过滤掉 IPv6 的 link-local 地址（fe80:...）
# 用这个不如用::1
ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况
if not ip.is_link_local:
network_ips.append(ip)If there are any other places in this file (or elsewhere in the codebase) that follow the same IPv6 pattern without checking ip.is_link_local, you should apply the same condition there to keep behavior consistent: parse with ip_address(...split("%")[0]) and only append when not ip.is_link_local.
| port = self.core_lifecycle.astrbot_config["dashboard"].get("port", 6185) | ||
| host = self.core_lifecycle.astrbot_config["dashboard"].get("host", "0.0.0.0") | ||
| enable = self.core_lifecycle.astrbot_config["dashboard"].get("enable", True) | ||
| def run(self) -> None: |
There was a problem hiding this comment.
issue (complexity): 建议把 IPv6 绑定逻辑、本地主机端口检查和 IP 格式化拆分到几个小工具函数中，让 run() 及其相关方法保持简单，聚焦在高层行为上。
你可以保留现在所有的新行为（IPv6、双栈绑定、更丰富的日志），但通过更紧凑的封装，以及移除不必要的“host 感知”逻辑来减少心智负担。
1. 用一个 helper 隐藏 Hypercorn 绑定的复杂度
目前 run() 需要关心 _build_bind()、IPv4/IPv6，以及双栈的兼容处理。可以保留这些细节，但把它们挪到一个低层 helper 中，只返回简单字符串：
def _get_binds(self, host: str, port: int) -> list[str]:
# normalize for bind construction
host = host.strip("[]")
def _fmt(h: str) -> str:
# only add [] when actually needed for IPv6
return f"[{h}]:{port}" if ":" in h else f"{h}:{port}"
binds = [_fmt(host)]
# dual-stack workaround stays here, not in run()
if host == "::" and platform.system() in ("Windows", "Darwin"):
binds.append(_fmt("0.0.0.0"))
return binds这样 run() 可以简化为：
config = HyperConfig()
config.bind = self._get_binds(host, port)有了这个 helper，就可以去掉 _build_bind()，也不需要在更高层代码中传递 IPv4Address | IPv6Address 和 ipaddress.ip_address()。
2. 把 check_port_in_use 简化回“检查 localhost 上的端口”
run() 始终调用的是 check_port_in_use("127.0.0.1", port)，因此额外的 host 参数、协议簇检测以及 IPv6 分支并没有带来实质收益，反而增加了认知负担。
可以保留现有的行为（“检查本机上某端口是否被占用”），但把函数签名和实现简化：
def check_port_in_use(self, port: int) -> bool:
"""Check if port is in use on localhost."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(2)
return sock.connect_ex(("127.0.0.1", port)) == 0
except Exception:
# be conservative on error
return True在 run() 中的用法也保持简单：
if self.check_port_in_use(port):
info = self.get_process_using_port(port)
raise RuntimeError(f"端口 {port} 已被占用\n{info}")这样可以让函数聚焦在它真正的使用场景上，又不丢失任何行为。
3. 把 IP 相关格式化逻辑集中到一个小 helper 中
目前 _print_access_urls() 需要了解 IPv4Address | IPv6Address、.is_loopback、.is_link_local、.version 和 URL 格式化等细节。你可以继续让 get_local_ip_addresses() 返回 ipaddress 的对象，但把所有 IP 相关的分支集中在一个小 helper 中：
def _iter_network_urls(self, port: int):
for ip in get_local_ip_addresses():
# filter once here, not in multiple places
if ip.is_loopback or ip.is_link_local:
continue
host = f"[{ip}]" if ip.version == 6 else str(ip)
yield f"http://{host}:{port}"这样 _print_access_urls() 就更容易阅读：
def _print_access_urls(self, host: str, port: int) -> None:
parts = [f"\n ✨✨✨\n AstrBot v{VERSION} WebUI 已启动\n\n"]
parts.append(f" ➜ 本地: http://localhost:{port}\n")
# only enumerate network URLs when binding all interfaces
if host in ("::", "0.0.0.0"):
for url in self._iter_network_urls(port):
parts.append(f" ➜ 网络: {url}\n")
parts.append(" ➜ 默认用户名和密码: astrbot\n ✨✨✨\n")
logger.info("".join(parts))这样可以保持当前所有行为（IPv6 URL、过滤等），但把底层地址处理集中在一个地方，而不是分散在 _print_access_urls() 的多处。
这些改动可以在保留你新增的 IPv6 支持和改进结构的同时，减少必须理解 socket、操作系统差异和 ipaddress 类型的代码位置。
Original comment in English
issue (complexity): Consider encapsulating the IPv6 bind logic, localhost port check, and IP formatting into small helpers so run() and related methods stay straightforward and focused on high-level behavior.
You can keep all the new behavior (IPv6, dual-stack bind, richer logging) but reduce the mental surface area by tightening encapsulation and removing unused “host-awareness” where it doesn’t add value.
1. Hide Hypercorn bind complexity behind a single helper
run() currently has to know about _build_bind(), IPv4/IPv6, and the dual-stack workaround. You can keep all of that but move it into one low-level helper that returns plain strings:
def _get_binds(self, host: str, port: int) -> list[str]:
# normalize for bind construction
host = host.strip("[]")
def _fmt(h: str) -> str:
# only add [] when actually needed for IPv6
return f"[{h}]:{port}" if ":" in h else f"{h}:{port}"
binds = [_fmt(host)]
# dual-stack workaround stays here, not in run()
if host == "::" and platform.system() in ("Windows", "Darwin"):
binds.append(_fmt("0.0.0.0"))
return bindsThen run() simplifies to:
config = HyperConfig()
config.bind = self._get_binds(host, port)With this in place, you can drop _build_bind() and avoid threading IPv4Address | IPv6Address and ipaddress.ip_address() through higher-level code.
2. Simplify check_port_in_use back to a “port on localhost” check
run() always calls check_port_in_use("127.0.0.1", port), so the extra host parameter, family detection, and IPv6 branch don’t bring practical benefit but add cognitive load.
You can keep the behavior (“is this port occupied on the local machine”) and simplify the signature:
def check_port_in_use(self, port: int) -> bool:
"""Check if port is in use on localhost."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(2)
return sock.connect_ex(("127.0.0.1", port)) == 0
except Exception:
# be conservative on error
return TrueUsage in run() stays simple:
if self.check_port_in_use(port):
info = self.get_process_using_port(port)
raise RuntimeError(f"端口 {port} 已被占用\n{info}")This focuses the function on its actual use case without losing any behavior.
3. Localize IP-specific formatting into a small helper
Right now _print_access_urls() needs to know about IPv4Address | IPv6Address, .is_loopback, .is_link_local, .version and URL formatting. You can keep get_local_ip_addresses() returning ipaddress objects but isolate all IP-specific branching in a tiny helper:
def _iter_network_urls(self, port: int):
for ip in get_local_ip_addresses():
# filter once here, not in multiple places
if ip.is_loopback or ip.is_link_local:
continue
host = f"[{ip}]" if ip.version == 6 else str(ip)
yield f"http://{host}:{port}"Then _print_access_urls() becomes much easier to read:
def _print_access_urls(self, host: str, port: int) -> None:
parts = [f"\n ✨✨✨\n AstrBot v{VERSION} WebUI 已启动\n\n"]
parts.append(f" ➜ 本地: http://localhost:{port}\n")
# only enumerate network URLs when binding all interfaces
if host in ("::", "0.0.0.0"):
for url in self._iter_network_urls(port):
parts.append(f" ➜ 网络: {url}\n")
parts.append(" ➜ 默认用户名和密码: astrbot\n ✨✨✨\n")
logger.info("".join(parts))This keeps all current behavior (IPv6 URLs, filtering, etc.) but moves the low-level address handling into a single place instead of spreading it across _print_access_urls().
These changes keep your new IPv6 support and improved structure while reducing the number of places that need to understand sockets, OS quirks, and ipaddress types.
There was a problem hiding this comment.
Pull request overview
This pull request adds IPv6 support to AstrBot by changing the default host binding from 0.0.0.0 (IPv4-only) to :: (dual-stack IPv4/IPv6). The changes enable the application to listen on both IPv4 and IPv6 addresses simultaneously, improving accessibility for modern networks.
Changes:
- Updated default host configuration from
0.0.0.0to::across dashboard, platform adapters, and CLI - Enhanced network utilities to properly handle IPv6 addresses with type-safe IP address objects
- Added CLI options for host configuration and backend-only mode
- Improved code quality with better type hints and error handling
- Added systemd service file for easier deployment
Reviewed changes
Copilot reviewed 22 out of 23 changed files in this pull request and generated 12 comments.
Show a summary per file
| File | Description |
|---|---|
| dashboard/vite.config.ts | Changed dev server host to :: for IPv6 support and applied consistent quote style |
| astrbot/dashboard/server.py | Major refactoring: improved IPv6 handling, port checking, dual-stack binding, and plugin route indexing |
| astrbot/dashboard/routes/route.py | Added to_json() helper method to Response class |
| astrbot/core/utils/io.py | Enhanced IP address handling with IPv6 support and added public IP detection function |
| astrbot/core/config/default.py | Updated all default host values from 0.0.0.0 to :: |
| astrbot/core/platform/sources/* | Updated platform adapter default hosts to :: for consistent IPv6 support |
| astrbot/cli/commands/cmd_run.py | Added --host and --backend-only CLI options with environment variable support |
| astrbot/core/agent/tool.py | Fixed Gemini API compatibility for JSON schema types that are lists |
| astrbot/core/platform/message_session.py | Changed platform_id field to init=False for proper dataclass usage |
| scripts/astrbot.service | Added systemd service file for deployment |
| README.md | Updated footer with better formatting and mascot image |
|
|
||
| async def fetch(session: aiohttp.ClientSession, url: str): | ||
| try: | ||
| async with session.get(url, timeout=3) as resp: |
There was a problem hiding this comment.
The timeout parameter passed to session.get is an integer (3), but aiohttp expects a ClientTimeout object. This will raise a TypeError. Should use aiohttp.ClientTimeout(total=3) instead.
| async with session.get(url, timeout=3) as resp: | |
| async with session.get(url, timeout=aiohttp.ClientTimeout(total=3)) as resp: |
| async def get_public_ip_address() -> list[IPv4Address | IPv6Address]: | ||
| urls = [ | ||
| "https://api64.ipify.org", | ||
| "https://ident.me", | ||
| "https://ifconfig.me", | ||
| "https://icanhazip.com", | ||
| ] | ||
| found_ips: dict[int, IPv4Address | IPv6Address] = {} | ||
|
|
||
| async def fetch(session: aiohttp.ClientSession, url: str): | ||
| try: | ||
| async with session.get(url, timeout=3) as resp: | ||
| if resp.status == 200: | ||
| raw_ip = (await resp.text()).strip() | ||
| ip = ip_address(raw_ip) | ||
| if ip.version not in found_ips: | ||
| found_ips[ip.version] = ip | ||
| except Exception: | ||
| pass | ||
|
|
||
| async with aiohttp.ClientSession() as session: | ||
| tasks = [fetch(session, url) for url in urls] | ||
| await asyncio.gather(*tasks) | ||
|
|
||
| # 返回找到的所有 IP 对象列表 | ||
| return list(found_ips.values()) | ||
|
|
||
|
|
There was a problem hiding this comment.
The get_public_ip_address function is defined but never used anywhere in the codebase. If this is intended for future use, consider adding a TODO comment. Otherwise, this function should be removed to avoid dead code.
| async def get_public_ip_address() -> list[IPv4Address | IPv6Address]: | |
| urls = [ | |
| "https://api64.ipify.org", | |
| "https://ident.me", | |
| "https://ifconfig.me", | |
| "https://icanhazip.com", | |
| ] | |
| found_ips: dict[int, IPv4Address | IPv6Address] = {} | |
| async def fetch(session: aiohttp.ClientSession, url: str): | |
| try: | |
| async with session.get(url, timeout=3) as resp: | |
| if resp.status == 200: | |
| raw_ip = (await resp.text()).strip() | |
| ip = ip_address(raw_ip) | |
| if ip.version not in found_ips: | |
| found_ips[ip.version] = ip | |
| except Exception: | |
| pass | |
| async with aiohttp.ClientSession() as session: | |
| tasks = [fetch(session, url) for url in urls] | |
| await asyncio.gather(*tasks) | |
| # 返回找到的所有 IP 对象列表 | |
| return list(found_ips.values()) |
| return self | ||
|
|
||
| def to_json(self): | ||
| return jsonify(asdict(self)) |
There was a problem hiding this comment.
The new to_json method calls jsonify, but jsonify is already being called in the outer context (line 176, 187, 193). This results in double-wrapping with jsonify, which is incorrect. The to_json method should return asdict(self) directly, and let the caller use jsonify.
| return jsonify(asdict(self)) | |
| return asdict(self) |
| except Exception as e: | ||
| logger.warning(f"检查端口 {port} 时发生错误: {e!s}") | ||
| # 如果出现异常，保守起见认为端口可能被占用 | ||
| return await handler(*args, **kwargs) |
There was a problem hiding this comment.
The host detection using ":" in host to determine IPv6 is insufficient. IPv4-mapped IPv6 addresses like "::ffff:192.0.2.1" would be incorrectly classified. Consider using ipaddress.ip_address to properly determine the address family, similar to the _build_bind method implementation.
| local_ips: list[IPv4Address | IPv6Address] = get_local_ip_addresses() | ||
|
|
There was a problem hiding this comment.
The comment states that link-local addresses are filtered out in get_local_ip_addresses (第一次过滤在get_local_ip_addresses), but this is incorrect as no filtering actually happens there. This duplicate filtering here suggests the original filtering was intended but not implemented. Consider either implementing the filter in get_local_ip_addresses or updating this comment to be accurate.
| os.environ["DASHBOARD_PORT"] = port or "6185" | ||
| os.environ["DASHBOARD_HOST"] = host or "::" |
There was a problem hiding this comment.
The CLI default values are set unconditionally with or-logic (port or "6185", host or "::"), which means if the user explicitly passes an empty string as the host or port, it will be overridden with the default. This may not be the intended behavior. Consider checking for None explicitly instead.
| os.environ["DASHBOARD_PORT"] = port or "6185" | |
| os.environ["DASHBOARD_HOST"] = host or "::" | |
| os.environ["DASHBOARD_PORT"] = port if port is not None else "6185" | |
| os.environ["DASHBOARD_HOST"] = host if host is not None else "::" |
| # 过滤掉 IPv6 的 link-local 地址（fe80:...） | ||
| # 用这个不如用::1 | ||
| ip = ip_address(addr.address.split("%")[0]) # 处理带 zone index 的情况 | ||
| network_ips.append(ip) |
There was a problem hiding this comment.
The comment claims to filter out IPv6 link-local addresses (fe80:...), but no actual filtering is performed. All IPv6 addresses including link-local ones are added to the list. If filtering is intended, add a check like: if not ip.is_link_local: network_ips.append(ip)
| network_ips.append(ip) | |
| # 只保留非 link-local 的 IPv6 地址 | |
| if isinstance(ip, IPv6Address) and not ip.is_link_local: | |
| network_ips.append(ip) |
| 陪伴与能力从来不应该是对立面。我们希望创造的是一个既能理解情绪、给予陪伴，也能可靠完成工作的机器人。 | ||
| _私は、高性能ですから!_ | ||
|
|
||
| <img src="https://files.astrbot.app/watashiwa-koseino-desukara.gif" width="100"/> |
There was a problem hiding this comment.
Consider adding alternative text to the image tag for accessibility. For example: alt="AstrBot mascot"
| <img src="https://files.astrbot.app/watashiwa-koseino-desukara.gif" width="100"/> | |
| <img src="https://files.astrbot.app/watashiwa-koseino-desukara.gif" alt="AstrBot mascot" width="100"/> |
| @@ -1,15 +1,18 @@ | |||
| import asyncio | |||
| import ipaddress | |||
There was a problem hiding this comment.
Module 'ipaddress' is imported with both 'import' and 'import from'.
| import ipaddress |
| except Exception: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| pass | |
| except (aiohttp.ClientError, asyncio.TimeoutError, ValueError) as exc: | |
| # Ignore failures from individual endpoints but log for debugging purposes. | |
| logger.debug("Failed to fetch public IP from %s: %s", url, exc) |
Modifications / 改动点
Screenshots or Test Results / 运行截图或测试结果
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Summary by Sourcery
在整个控制面板和平台端点中新增 IPv6 优先的网络支持，同时重构控制面板的启动流程和路由结构，以提高结构清晰度和健壮性。
新功能：
错误修复：
增强改进：
构建：
文档：
Original summary in English
Summary by Sourcery
Add IPv6-first networking support across the dashboard and platform endpoints while refactoring dashboard startup and routing for better structure and robustness.
New Features:
Bug Fixes:
Enhancements:
Build:
Documentation: