feat(code): /tools slash command (#4649)

Deep Agents Code now supports `/tools`, which lists the built-in and MCP
tools available to the current agent.

---

Adds an interactive `/tools` slash command to the `deepagents-code` TUI
so users can see the tools available to the current agent, grouped into
built-in tools and tools from each MCP server. MCP servers that are
disabled, need login, or failed to load are shown separately instead of
disappearing from the list.

For managed sessions, built-in tools are enumerated off the UI thread
with a credential-free agent compile. For preconfigured local agents,
`/tools` inspects the active graph directly so custom tool sets remain
authoritative. When a custom or remote agent cannot be inspected, the
command explains that limitation rather than incorrectly reporting that
no tools are available.

MCP entries reuse the metadata loaded for the running agent instead of
starting discovery inside Textual's live event loop. If `/mcp`
configuration changes are waiting for a reconnect, `/tools` keeps using
the pre-change snapshot so its output continues to match the tools the
agent can currently invoke.

The catalog is rendered as a chat message, like `/tokens`, and is
discoverable through `/help`, command completion, and a startup tip.

Made by [Open
SWE](https://openswe.vercel.app/agents/626ca1ec-726e-58e8-7dae-4e85363c953d)

## References

- Plan:
https://openswe.vercel.app/agents/626ca1ec-726e-58e8-7dae-4e85363c953d/plan

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
2026-07-13 03:13:37 -04:00
committed by GitHub
parent 4f94a30c11
commit b1600a8da7
12 changed files with 1070 additions and 34 deletions
+2 -1
View File
@@ -8,7 +8,7 @@ Regenerate this file with `make commands-catalog` after changing command names,
aliases, descriptions, visibility, or hidden-command metadata.
## Public (32)
## Public (33)
| Command | Aliases | Description |
| --- | --- | --- |
@@ -41,6 +41,7 @@ aliases, descriptions, visibility, or hidden-command metadata.
| `/threads` | | Browse and resume past threads |
| `/timestamps` | | Show or hide message timestamps |
| `/tokens` | | Show token usage |
| `/tools` | | List the tools available to the agent |
| `/trace` | | Open this thread in LangSmith |
| `/update` | | Check for and install updates |
| `/version` | `/about` | Show version information |
+9
View File
@@ -22,6 +22,15 @@ FIREWORKS_MODEL_ID_PREFIXES: Final[tuple[str, ...]] = (
)
"""Model and router ID prefixes used for stripping and classification."""
MCP_REENABLED_PENDING_ERROR: Final[str] = "Re-enabled — press Ctrl+R to load."
"""User-facing reconnect guidance shown for an MCP server that was optimistically
re-enabled but whose agent has not yet reconnected.
Set as `MCPServerInfo.error` by `app._apply_optimistic_disabled_state` (alongside
`pending_reconnect=True`, which is what `/tools` actually keys off). Named here
so the producer and the tests asserting the message share one literal.
"""
SYSTEM_MESSAGE_PREFIX: Final[str] = "[SYSTEM]"
"""Prefix for synthetic human messages (e.g. interrupt cancellation notices).
+192 -2
View File
@@ -45,6 +45,7 @@ from deepagents_code import (
from deepagents_code._cli_context import CLIContext
from deepagents_code._constants import (
DEFAULT_AGENT_NAME as DEFAULT_ASSISTANT_ID,
MCP_REENABLED_PENDING_ERROR,
SYSTEM_MESSAGE_PREFIX,
)
from deepagents_code._git import (
@@ -361,6 +362,7 @@ if TYPE_CHECKING:
from deepagents_code.model_config import MissingProviderPackageError
from deepagents_code.resume_state import GoalStatus
from deepagents_code.skills.load import ExtendedSkillMetadata
from deepagents_code.tool_catalog import ToolCatalog, UnavailableServer
from deepagents_code.tui.textual_adapter import TextualUIAdapter
from deepagents_code.tui.widgets.approval import ApprovalMenu
from deepagents_code.tui.widgets.ask_user import AskUserMenu
@@ -8378,6 +8380,191 @@ class DeepAgentsApp(App):
await self._mount_message(UserMessage(command))
await self._mount_message(AppMessage(msg))
async def _handle_tools_command(self, command: str) -> None:
"""List the tools available to the agent as a chat message.
Managed sessions enumerate built-ins off the UI thread with a
credential-free agent compile; preconfigured local agents are inspected
directly so their custom tool set stays authoritative. MCP tools reuse
the metadata loaded for the running agent rather than re-discovering,
because discovery uses `asyncio.run`, which cannot run inside Textual's
live event loop.
Args:
command: The raw command text (displayed as a user message).
"""
from deepagents_code._constants import DEFAULT_AGENT_NAME
from deepagents_code.tool_catalog import (
build_catalog_from_server_info,
collect_built_in_tools,
collect_tools_from_agent,
)
await self._mount_message(UserMessage(command))
server_info = self._mcp_server_info_for_tools()
built_in = []
# Set to a human-readable reason when built-in tools cannot be listed:
# a remote/custom agent we cannot introspect, or a compile that raised.
# The agent still binds those tools; only enumeration failed. Left empty
# on success. Drives the notice logic below.
enumeration_failed_reason = ""
if self._server_kwargs is None:
try:
active_tools = (
collect_tools_from_agent(self._agent)
if self._agent is not None
else None
)
except Exception:
# `collect_tools_from_agent` is defensively written, but a remote
# graph proxy can do real work on attribute access and raise.
# Mirror the managed branch: log and degrade to the notice below
# rather than letting it escape as an unhandled handler error.
logger.exception("Failed to inspect agent tools for /tools")
active_tools = None
if active_tools is None:
enumeration_failed_reason = (
"Built-in tools cannot be enumerated for this custom or "
"remote agent"
)
else:
# The local graph's tool node holds built-in *and* MCP tools;
# MCP tools are rendered separately from `server_info`, so drop
# them here to avoid listing each MCP tool twice.
mcp_names = {
tool.name for server in server_info for tool in server.tools
}
built_in = [tool for tool in active_tools if tool.name not in mcp_names]
else:
enable_interpreter = bool(self._server_kwargs.get("enable_interpreter"))
try:
built_in = await asyncio.to_thread(
collect_built_in_tools,
assistant_id=self._assistant_id or DEFAULT_AGENT_NAME,
enable_interpreter=enable_interpreter,
)
except Exception:
logger.exception("Failed to enumerate built-in tools for /tools")
enumeration_failed_reason = "Could not enumerate built-in tools"
catalog = build_catalog_from_server_info(built_in, server_info)
has_mcp_info = any(group.source == "mcp" for group in catalog.groups) or bool(
catalog.unavailable
)
if enumeration_failed_reason and not has_mcp_info:
# No MCP tools or unavailable-server statuses to show: rendering "0
# tools available" would wrongly imply the agent has no tools at all,
# when in fact only the listing failed. Surface the reason on its own.
await self._mount_message(
AppMessage(
f"{enumeration_failed_reason}. The agent still has its "
"built-in tools; they just cannot be listed here.",
),
)
return
if enumeration_failed_reason:
await self._mount_message(
AppMessage(
f"{enumeration_failed_reason}; showing MCP information only."
),
)
await self._mount_message(AppMessage(self._render_tool_catalog(catalog)))
def _mcp_server_info_for_tools(self) -> list[MCPServerInfo]:
"""Return MCP metadata matching the tools bound to the running agent.
The `/mcp` viewer optimistically replaces a newly disabled server with
a tool-less cosmetic entry before reconnect. Until reconnect actually
rebuilds the agent, its original tools remain callable, so `/tools`
must use the saved pre-toggle entry instead.
Returns:
MCP server metadata for the active agent tool set.
"""
return [
self._mcp_optimistic_original_server_info.get(server.name, server)
for server in self._mcp_server_info or []
]
@staticmethod
def _render_tool_catalog(catalog: ToolCatalog) -> Content:
"""Render a tool catalog as chat `Content`.
Shows a count header, then each group's heading with its rows:
built-in groups render aligned `name description` rows, while MCP
groups render names only their descriptions are surfaced via `/mcp`,
noted by a pointer line whenever any MCP tools are present. Then any MCP
servers that loaded with no tools and a non-`ok` status, and finally a
discovery-error notice if the catalog carries one. Every display
string tool names/descriptions and MCP server names/statuses, some of
which are external is added as a plain-text span (via `Content.styled`
or a `(text, style)` tuple) that is never parsed as markup.
Args:
catalog: Collected tool groups, unavailable MCP servers, and any
discovery-error notice.
Returns:
Assembled `Content` ready to mount in an `AppMessage`.
"""
from deepagents_code.tool_catalog import unavailable_server_display
total = sum(len(group.tools) for group in catalog.groups)
noun = "tool" if total == 1 else "tools"
parts: list[str | Content | tuple[str, str | TStyle]] = [
Content.styled(f"{total} {noun} available", "bold"),
]
def _section(heading: str, rows: list[tuple[str, str]]) -> None:
"""Append a bold heading and left-aligned `label detail` rows."""
if not rows:
return
width = max(len(label) for label, _ in rows)
parts.extend(("\n\n", Content.styled(heading, "bold")))
for label, detail in rows:
row = f" {label.ljust(width)} {detail}".rstrip()
parts.extend(("\n", (row, "dim")))
has_mcp_tools = False
for group in catalog.groups:
if group.source == "mcp":
has_mcp_tools = has_mcp_tools or bool(group.tools)
rows = [(entry.name, "") for entry in group.tools]
else:
rows = [(entry.name, entry.description) for entry in group.tools]
_section(group.label, rows)
if has_mcp_tools:
parts.extend(
("\n\n", ("MCP tool descriptions are available in /mcp.", "dim"))
)
def _unavailable_row(server: UnavailableServer) -> tuple[str, str]:
"""Map an unavailable server to a `(name, detail)` row for `_section`.
Args:
server: An MCP server discovered with no usable tools.
Returns:
A `(name, detail)` pair for `_section` to align and render.
"""
label, detail = unavailable_server_display(server)
return (server.name, f"{label}: {detail}" if detail else label)
_section(
"Unavailable MCP servers",
[_unavailable_row(server) for server in catalog.unavailable],
)
if catalog.mcp_error:
parts.extend(("\n\n", (catalog.mcp_error, "dim")))
return Content.assemble(*parts)
def _goal_state_update(self) -> dict[str, Any]:
"""Build checkpoint state for TUI-owned goal/rubric metadata.
@@ -9684,7 +9871,7 @@ class DeepAgentsApp(App):
"/notifications, /reload, /restart, /rubric, "
"/skill:<name>, /remember, "
"/skill-creator, /theme, /scrollbar, /timestamps, /tokens, "
"/threads, /trace, "
"/tools, /threads, /trace, "
"/update, /auto-update, /install, /changelog, /docs, "
"/feedback, /help\n\n"
"Interactive Features:\n"
@@ -9926,6 +10113,8 @@ class DeepAgentsApp(App):
parts.append(model_name)
await self._mount_message(AppMessage(" · ".join(parts)))
elif cmd == "/tools":
await self._handle_tools_command(command)
elif cmd == "/remember" or cmd.startswith("/remember "):
# Convenience alias for /skill:remember — shorter and discoverable
# before skill loading completes.
@@ -15494,7 +15683,8 @@ class DeepAgentsApp(App):
name=entry.name,
transport=entry.transport,
status="disabled",
error="Re-enabled — press Ctrl+R to load.",
error=MCP_REENABLED_PENDING_ERROR,
pending_reconnect=True,
),
)
self._mcp_server_info = updated
@@ -226,6 +226,7 @@ def _print_unavailable_servers(servers: tuple[UnavailableServer, ...]) -> None:
if not servers:
return
from deepagents_code.config import console
from deepagents_code.tool_catalog import unavailable_server_display
name_width = max(len(server.name) for server in servers)
console.print()
@@ -234,11 +235,16 @@ def _print_unavailable_servers(servers: tuple[UnavailableServer, ...]) -> None:
)
for server in servers:
padded = server.name.ljust(name_width)
# `status: detail` (ASCII-only, no em-dash) so legacy consoles don't hit
# an encoding error; detail is discovery's own curated reason string.
detail = f": {server.detail}" if server.detail else ""
# `status: detail`, where detail is discovery's own curated reason string
# (ASCII on this CLI path, so legacy consoles don't hit an encoding
# error). `unavailable_server_display` collapses a disabled server to
# "disabled by user" with no detail; other statuses keep discovery's
# reason string. (The TUI's em-dash reconnect guidance is never produced
# by CLI discovery, so it cannot reach this column here.)
status, detail_text = unavailable_server_display(server)
detail = f": {detail_text}" if detail_text else ""
console.print(
f" {padded} {server.status}{detail}".rstrip(),
f" {padded} {status}{detail}".rstrip(),
style="dim",
markup=False,
highlight=False,
@@ -180,6 +180,12 @@ COMMANDS: tuple[SlashCommand, ...] = (
bypass_tier=BypassTier.QUEUED,
hidden_keywords="cost",
),
SlashCommand(
name="/tools",
description="List the tools available to the agent",
bypass_tier=BypassTier.QUEUED,
hidden_keywords="mcp functions capabilities builtin built-in",
),
SlashCommand(
name="/reload",
description="Reload environment and config",
+18 -1
View File
@@ -112,13 +112,24 @@ class MCPServerInfo:
error: str | None = None
"""Human-readable reason when `status != "ok"`."""
pending_reconnect: bool = False
"""`True` for a disabled entry that was just re-enabled in the TUI and is
awaiting a reconnect to load its tools.
Lets `/tools` (`tool_catalog.split_mcp_server_info`) preserve the reconnect
guidance held in `error` instead of collapsing it to the generic "disabled
by user" label — an explicit flag rather than a fragile match on the
guidance text. Only meaningful while `status == "disabled"`.
"""
def __post_init__(self) -> None:
"""Enforce the status/error/tools consistency invariant.
Raises:
ValueError: If any of: `status='ok'` with a non-`None` error;
non-`ok` status without an error message; non-`ok` status
carrying tools.
carrying tools; or `pending_reconnect` set without
`status='disabled'`.
"""
if self.status == "ok":
if self.error is not None:
@@ -140,6 +151,12 @@ class MCPServerInfo:
"cannot carry tools"
)
raise ValueError(msg)
if self.pending_reconnect and self.status != "disabled":
msg = (
f"MCPServerInfo {self.name!r}: pending_reconnect requires "
f"status='disabled' (got {self.status!r})"
)
raise ValueError(msg)
def needs_attention(self) -> bool:
"""Return whether this server is blocked on user login."""
+177 -22
View File
@@ -1,4 +1,7 @@
"""Enumerate the tools available to the agent for `dcode tools list`.
"""Enumerate the tools available to the agent.
Backs two entry points: the `dcode tools list` CLI command (`_run_tools_list`)
and the interactive `/tools` slash command (`app._handle_tools_command`).
The tool set is read from the *real* tool objects the agent binds rather than a
hand-maintained catalog, so names and descriptions never drift from what the
@@ -9,15 +12,15 @@ tool node; MCP tools are discovered via the same path the app and server use.
The collection functions here lazily import the heavy agent stack (agent
compilation, MCP discovery) inside their bodies. Only the fake-model base is
imported at module top, so importing this module is cheap relative to the agent
stack — and this module is itself imported lazily inside `_run_tools_list`,
never on the startup hot path. Those functions must only run on the
`dcode tools list` command path.
stack — and this module is itself imported lazily by both entry points
(`_run_tools_list` and `_handle_tools_command`), never on the startup hot path.
"""
from __future__ import annotations
import asyncio
import logging
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
@@ -25,9 +28,11 @@ from typing import TYPE_CHECKING, Any, Literal, cast
from deepagents_code._fake_models import _ToolBindingFakeModel
if TYPE_CHECKING:
from collections.abc import Sequence
from langgraph.prebuilt.tool_node import ToolNode
from deepagents_code.mcp_tools import MCPServerStatus
from deepagents_code.mcp_tools import MCPServerInfo, MCPServerStatus
logger = logging.getLogger(__name__)
@@ -91,6 +96,20 @@ class UnavailableServer:
the same text the interactive `/mcp` viewer shows. See `collect_mcp_catalog`.
"""
def __post_init__(self) -> None:
"""Enforce that an unavailable server is never `ok`.
An `ok` server exposes tools and belongs in a `ToolGroup`, never here;
rejecting it at construction keeps the documented non-`ok` invariant
from being silently violated by a future producer.
Raises:
ValueError: If `status` is `"ok"`.
"""
if self.status == "ok":
msg = "UnavailableServer.status must be a non-'ok' MCPServerStatus"
raise ValueError(msg)
@dataclass(frozen=True, slots=True)
class ToolCatalog:
@@ -114,6 +133,28 @@ class ToolCatalog:
"""
def unavailable_server_display(server: UnavailableServer) -> tuple[str, str]:
"""Return the `(status_label, detail)` display pair for an unavailable server.
Shared by the CLI (`client.commands.tools._print_unavailable_servers`) and
TUI (`app._render_tool_catalog`) renderers so both describe a server the same
way. A disabled server shows its reconnect guidance if present, else the
generic "disabled by user", with no separate detail; other statuses show the
status token plus discovery's reason string when present.
Args:
server: A server that loaded with no usable tools.
Returns:
`(status_label, detail)`: the primary status text and any secondary
detail (`""` when none). Each renderer lays these out itself, e.g. as
`status_label: detail`.
"""
if server.status == "disabled":
return (server.detail or "disabled by user", "")
return (server.status, server.detail)
class _CatalogModel(_ToolBindingFakeModel):
"""Offline placeholder model used only to compile the agent for enumeration.
@@ -160,6 +201,9 @@ def collect_built_in_tools(
Returns:
Built-in tools in bind order.
Raises:
RuntimeError: If the compiled graph does not expose its bound tools.
"""
from deepagents_code.agent import create_cli_agent
from deepagents_code.config import settings
@@ -180,17 +224,67 @@ def collect_built_in_tools(
enable_shell=True,
enable_interpreter=enable_interpreter,
)
# `agent.nodes["tools"].bound` reaches into langgraph's compiled graph; the
# "tools" node name and `ToolNode.tools_by_name` are langgraph conventions,
# not a documented contract. This real compile path is exercised by
# test_tool_catalog.py::TestCollectBuiltInTools, so a breaking langgraph
# change fails loudly there rather than silently emitting an empty list.
tool_node = cast("ToolNode", agent.nodes["tools"].bound)
tools_by_name = tool_node.tools_by_name
return [
ToolEntry(name=name, description=_first_line(tool.description))
for name, tool in tools_by_name.items()
]
tools = collect_tools_from_agent(agent)
if tools is None:
msg = "Compiled agent does not expose a LangGraph tool node"
raise RuntimeError(msg)
return tools
def collect_tools_from_agent(agent: object) -> list[ToolEntry] | None:
"""Read tools from a local compiled agent when its graph is inspectable.
LangGraph does not expose a public tool-enumeration API, so this reaches
through the compiled graph's conventional `nodes["tools"].bound` shape.
Returning `None` distinguishes an uninspectable graph (a remote agent, or a
local graph whose internals no longer match that convention) from a local
graph that validly binds zero tools (`[]`).
Args:
agent: Active local or remote agent object.
Returns:
Bound tools in graph order; `[]` for an inspectable local graph with no
tools; or `None` when the agent cannot be inspected locally.
"""
nodes = getattr(agent, "nodes", None)
if not isinstance(nodes, Mapping):
# No conventional node map: a remote agent or a non-graph object. Expected
# for remote agents, so debug rather than warning.
logger.debug("Agent %r has no inspectable node map", type(agent))
return None
if "tools" not in nodes:
# LangChain omits the tool node when an otherwise valid local agent
# binds no tools. The graph is still inspectable; its tool set is empty.
return []
node = nodes.get("tools")
tool_node = cast("ToolNode | None", getattr(node, "bound", None))
tools_by_name = getattr(tool_node, "tools_by_name", None)
if not isinstance(tools_by_name, Mapping):
# A "tools" node exists but does not expose the expected
# `bound.tools_by_name` mapping — a LangGraph internal-shape change, not
# a remote agent. Warn so this drift is visible in logs even though the
# user-facing notice attributes it to an uninspectable agent.
logger.warning(
"Agent 'tools' node is not introspectable (bound=%r); "
"LangGraph internals may have changed",
type(tool_node),
)
return None
tools: list[ToolEntry] = []
for name, tool in tools_by_name.items():
if not isinstance(name, str):
continue
description = getattr(tool, "description", None)
tools.append(
ToolEntry(
name=name,
description=_first_line(
description if isinstance(description, str) else None
),
)
)
return tools
def collect_mcp_catalog(
@@ -232,9 +326,34 @@ def collect_mcp_catalog(
logger.warning("MCP tool discovery failed for `tools list`", exc_info=True)
return [], [], "MCP discovery failed; showing built-in tools only."
groups, unavailable = split_mcp_server_info(server_info)
return groups, unavailable, None
def split_mcp_server_info(
server_info: Sequence[MCPServerInfo],
) -> tuple[list[ToolGroup], list[UnavailableServer]]:
"""Split loaded MCP server metadata into tool groups and unavailable servers.
Pure function shared by the CLI discovery path (`collect_mcp_catalog`) and
the interactive `/tools` command, which passes the app's already-loaded
`MCPServerInfo` list rather than re-discovering (Textual's running event
loop forbids the `asyncio.run` discovery path).
Servers that loaded but expose no tools are reported as `UnavailableServer`s
(errored, needing login, or disabled) rather than silently dropped —
surfacing exactly what a user debugging a missing tool needs to see.
Args:
server_info: Loaded MCP server metadata.
Returns:
`(groups, unavailable)`: per-server tool groups (only servers exposing
tools) and servers discovered with no tools and a non-`ok` status.
"""
groups: list[ToolGroup] = []
unavailable: list[UnavailableServer] = []
for server in server_info or []:
for server in server_info:
if server.tools:
entries = tuple(
ToolEntry(name=tool.name, description=_first_line(tool.description))
@@ -244,18 +363,54 @@ def collect_mcp_catalog(
elif server.status != "ok":
# A server that loaded but has no tools *and* is not "ok" is broken,
# unauthenticated, or disabled — report it so the omission is
# explained. `server.error` is discovery's own reason string (the
# same text the interactive `/mcp` viewer shows); it is not a stack
# trace, but for config-load failures it can include the local
# explained. A plainly-disabled server drops discovery's reason so
# the renderers show the generic "disabled by user" label; a
# just-re-enabled one (`pending_reconnect`) keeps its reconnect
# guidance so the renderer can distinguish it from a server the user
# left disabled. Other statuses retain discovery's reason string —
# not a stack trace, but config-load failures can include the local
# config file path — see `UnavailableServer.detail`.
detail = server.error or ""
if server.status == "disabled" and not server.pending_reconnect:
detail = ""
unavailable.append(
UnavailableServer(
name=server.name,
status=server.status,
detail=server.error or "",
detail=detail,
)
)
return groups, unavailable, None
return groups, unavailable
def build_catalog_from_server_info(
built_in: Sequence[ToolEntry],
server_info: Sequence[MCPServerInfo],
) -> ToolCatalog:
"""Assemble a `ToolCatalog` from pre-collected built-in tools and live MCP info.
The interactive `/tools` command entry point: it avoids the `asyncio.run`
MCP discovery reached via `collect_catalog` (the `asyncio.run` call itself
lives in `collect_mcp_catalog`), which cannot run inside Textual's running
event loop, by reusing the MCP metadata the app already loaded. `mcp_error`
is always `None` here because discovery is not attempted — any load failures
are already reflected per-server in `server_info` as non-`ok` `MCPServerInfo`
entries, which `split_mcp_server_info` surfaces as `UnavailableServer`s.
Args:
built_in: Built-in tools in bind order (from `collect_built_in_tools`).
server_info: The app's already-loaded MCP server metadata.
Returns:
A `ToolCatalog` with the built-in group first, then any MCP groups, plus
unavailable servers.
"""
groups: list[ToolGroup] = [
ToolGroup(label=BUILT_IN_GROUP, source="built-in", tools=tuple(built_in))
]
mcp_groups, unavailable = split_mcp_server_info(server_info)
groups.extend(mcp_groups)
return ToolCatalog(groups=tuple(groups), unavailable=tuple(unavailable))
async def _load_mcp_server_info(
@@ -16,6 +16,7 @@ _TIPS: dict[str, int] = {
"Use /threads -r to jump back to the thread you just left with /clear": 1,
"Use /offload when your conversation gets long": 2,
"Use /copy to copy the latest assistant message": 3,
"Use /tools to list the tools available to the agent": 1,
"Use /mcp to search your MCP servers and inspect tool parameters": 1,
"Use /mcp login <server> to authenticate MCP OAuth servers without leaving the TUI": 1, # noqa: E501
"Use /remember to save learnings from this conversation": 1,
@@ -496,10 +496,31 @@ class TestToolsList:
assert code == 0
assert "Unavailable MCP servers" in output
assert "offsvc" in output
assert "disabled" in output
assert "disabled by user" in output
# Empty detail → status stands alone, no trailing `: `.
assert "disabled:" not in output
def test_pending_reenable_renders_reconnect_guidance(self) -> None:
catalog = ToolCatalog(
groups=(),
unavailable=(
UnavailableServer(
name="notion",
status="disabled",
detail="Re-enabled — press Ctrl+R to load.",
),
),
)
args = argparse.Namespace(tools_command="list", output_format="text")
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code, output = _run_text(args)
assert code == 0
assert "Re-enabled — press Ctrl+R to load." in output
assert "disabled by user" not in output
def test_list_end_to_end_offline_renders_real_built_ins(self) -> None:
"""Real `collect_catalog` compiles the agent offline and renders it."""
args = argparse.Namespace(
+411
View File
@@ -11341,6 +11341,417 @@ class TestEditorSlashCommand:
mock.assert_awaited_once()
class TestToolsSlashCommand:
"""Tests for the `/tools` slash command."""
async def test_mounts_user_echo_then_catalog(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
from deepagents_code.tool_catalog import ToolEntry
app = DeepAgentsApp(agent=MagicMock())
app._assistant_id = "agent"
app._server_kwargs = {"enable_interpreter": False}
app._mcp_server_info = [
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="search_docs", description="Search"),),
status="ok",
),
]
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
return_value=[ToolEntry(name="read_file", description="Read a file")],
) as collect,
):
await app._handle_command("/tools")
collect.assert_called_once_with(assistant_id="agent", enable_interpreter=False)
assert mount.await_count == 2
first, second = (c.args[0] for c in mount.await_args_list)
assert isinstance(first, UserMessage)
assert isinstance(second, AppMessage)
rendered = second._content.plain
assert "read_file" in rendered
assert "search_docs" in rendered
assert "docs" in rendered
async def test_built_in_failure_still_shows_mcp(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
app = DeepAgentsApp(agent=MagicMock())
app._assistant_id = "agent"
app._server_kwargs = {}
app._mcp_server_info = [
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="search_docs", description="Search"),),
status="ok",
),
]
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
side_effect=RuntimeError("compile boom"),
),
):
await app._handle_command("/tools")
# User echo, the built-in failure notice, and the MCP-only catalog.
assert mount.await_count == 3
catalog_msg = mount.await_args_list[-1].args[0]
assert isinstance(catalog_msg, AppMessage)
assert "search_docs" in catalog_msg._content.plain
async def test_custom_local_agent_uses_its_bound_tools(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
tool_node = SimpleNamespace(
tools_by_name={
"custom_search": SimpleNamespace(description="Search custom data"),
"search_docs": SimpleNamespace(description="Search docs"),
}
)
agent = MagicMock()
agent.nodes = {"tools": SimpleNamespace(bound=tool_node)}
app = DeepAgentsApp(
agent=agent,
mcp_server_info=[
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="search_docs", description="Search docs"),),
status="ok",
)
],
)
app._server_kwargs = None
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch("deepagents_code.tool_catalog.collect_built_in_tools") as compile_,
):
await app._handle_command("/tools")
compile_.assert_not_called()
assert mount.await_count == 2
rendered = mount.await_args_list[-1].args[0]._content.plain
assert rendered.count("custom_search") == 1
assert "Search custom data" in rendered
assert rendered.count("search_docs") == 1
assert "Search docs" not in rendered
assert "MCP tool descriptions are available in /mcp." in rendered
async def test_tool_less_local_agent_reports_zero_tools(self) -> None:
agent = MagicMock()
agent.nodes = {"__start__": SimpleNamespace(), "model": SimpleNamespace()}
app = DeepAgentsApp(agent=agent)
app._server_kwargs = None
app._mcp_server_info = []
with patch.object(app, "_mount_message", new_callable=AsyncMock) as mount:
await app._handle_command("/tools")
assert mount.await_count == 2
rendered = mount.await_args_list[-1].args[0]._content.plain
assert "0 tools available" in rendered
assert "cannot be enumerated" not in rendered
assert "still has its built-in tools" not in rendered
async def test_remote_agent_reports_built_in_enumeration_unsupported(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
app = DeepAgentsApp(
agent=MagicMock(spec=[]),
mcp_server_info=[
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="search_docs", description="Search"),),
status="ok",
)
],
)
app._server_kwargs = None
with patch.object(app, "_mount_message", new_callable=AsyncMock) as mount:
await app._handle_command("/tools")
assert mount.await_count == 3
notice = mount.await_args_list[1].args[0]
assert "cannot be enumerated" in str(notice._content)
assert "search_docs" in mount.await_args_list[-1].args[0]._content.plain
async def test_remote_agent_reports_unavailable_mcp_server(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo
app = DeepAgentsApp(
agent=MagicMock(spec=[]),
mcp_server_info=[
MCPServerInfo(
name="notion",
transport="http",
status="unauthenticated",
error="Login required",
)
],
)
app._server_kwargs = None
with patch.object(app, "_mount_message", new_callable=AsyncMock) as mount:
await app._handle_command("/tools")
assert mount.await_count == 3
notice = str(mount.await_args_list[1].args[0]._content)
assert "cannot be enumerated" in notice
assert "showing MCP information only" in notice
rendered = mount.await_args_list[-1].args[0]._content.plain
assert "Unavailable MCP servers" in rendered
assert "notion" in rendered
assert "Login required" in rendered
async def test_pending_mcp_disable_keeps_active_tools_visible(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
original = MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="search_docs", description="Search"),),
status="ok",
)
app = DeepAgentsApp(agent=MagicMock(), mcp_server_info=[original])
app._server_kwargs = {"enable_interpreter": False}
app._apply_optimistic_disabled_state("docs", disabled=True)
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
return_value=[],
),
):
await app._handle_command("/tools")
rendered = mount.await_args_list[-1].args[0]._content.plain
assert "search_docs" in rendered
assert "Unavailable MCP servers" not in rendered
async def test_pending_mcp_reenable_keeps_reconnect_guidance(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo
app = DeepAgentsApp(
agent=MagicMock(),
mcp_server_info=[
MCPServerInfo(
name="notion",
transport="http",
status="disabled",
error="Disabled by user (F2 to re-enable).",
),
],
)
app._server_kwargs = {"enable_interpreter": False}
app._apply_optimistic_disabled_state("notion", disabled=False)
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
return_value=[],
),
):
await app._handle_command("/tools")
rendered = mount.await_args_list[-1].args[0]._content.plain
assert "notion" in rendered
assert "Re-enabled — press Ctrl+R to load." in rendered
assert "disabled by user" not in rendered
def test_render_tool_catalog_reports_unavailable(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo
from deepagents_code.tool_catalog import (
ToolEntry,
build_catalog_from_server_info,
)
servers = [
MCPServerInfo(
name="broken", transport="http", status="error", error="boom"
),
MCPServerInfo(
name="notion",
transport="http",
status="disabled",
error="Disabled by user (F2 to re-enable).",
),
]
catalog = build_catalog_from_server_info(
[ToolEntry(name="read_file", description="Read a file")], servers
)
rendered = DeepAgentsApp._render_tool_catalog(catalog).plain
assert "1 tool available" in rendered
assert "Unavailable MCP servers" in rendered
assert "broken" in rendered
assert "boom" in rendered
assert "notion" in rendered
assert "disabled by user" in rendered
assert "disabled:" not in rendered
assert "F2" not in rendered
async def test_forwards_enable_interpreter_true(self) -> None:
from deepagents_code.tool_catalog import ToolEntry
app = DeepAgentsApp(agent=MagicMock())
app._assistant_id = "agent"
app._server_kwargs = {"enable_interpreter": True}
app._mcp_server_info = []
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
return_value=[ToolEntry(name="js_eval", description="Run JS")],
) as collect,
):
await app._handle_command("/tools")
collect.assert_called_once_with(assistant_id="agent", enable_interpreter=True)
assert mount.await_count == 2
assert "js_eval" in mount.await_args_list[-1].args[0]._content.plain
async def test_built_in_failure_without_mcp_is_honest(self) -> None:
app = DeepAgentsApp(agent=MagicMock())
app._assistant_id = "agent"
app._server_kwargs = {}
app._mcp_server_info = []
with (
patch.object(app, "_mount_message", new_callable=AsyncMock) as mount,
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
side_effect=RuntimeError("compile boom"),
),
):
await app._handle_command("/tools")
# Only the user echo and an honest notice — no misleading "0 tools"
# catalog when the agent still binds tools that just could not be listed.
assert mount.await_count == 2
notice = str(mount.await_args_list[-1].args[0]._content)
assert "still has its built-in tools" in notice
assert "showing MCP tools only" not in notice
assert "0 tools available" not in notice
async def test_remote_agent_without_mcp_is_honest(self) -> None:
app = DeepAgentsApp(agent=MagicMock(spec=[]))
app._server_kwargs = None
app._mcp_server_info = []
with patch.object(app, "_mount_message", new_callable=AsyncMock) as mount:
await app._handle_command("/tools")
assert mount.await_count == 2
notice = str(mount.await_args_list[-1].args[0]._content)
assert "cannot be enumerated" in notice
assert "still has its built-in tools" in notice
assert "showing MCP tools only" not in notice
def test_render_escapes_markup(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
from deepagents_code.tool_catalog import (
ToolEntry,
build_catalog_from_server_info,
)
catalog = build_catalog_from_server_info(
[ToolEntry(name="[bold]evil[/bold]", description="desc [red]x[/red]")],
[
MCPServerInfo(
name="[i]srv",
transport="http",
tools=(MCPToolInfo(name="t", description="d"),),
status="ok",
)
],
)
# Rich markup in external names/descriptions must survive verbatim and
# never be parsed (an unclosed "[" would raise if it were).
rendered = DeepAgentsApp._render_tool_catalog(catalog).plain
assert "[bold]evil[/bold]" in rendered
assert "[red]x[/red]" in rendered
assert "[i]srv" in rendered
def test_render_empty_catalog_reports_zero_without_error(self) -> None:
from deepagents_code.tool_catalog import build_catalog_from_server_info
rendered = DeepAgentsApp._render_tool_catalog(
build_catalog_from_server_info([], [])
).plain
assert "0 tools available" in rendered
def test_render_plural_noun_and_column_alignment(self) -> None:
from deepagents_code.tool_catalog import (
ToolEntry,
build_catalog_from_server_info,
)
catalog = build_catalog_from_server_info(
[
ToolEntry(name="ls", description="list"),
ToolEntry(name="read_file", description="read"),
],
[],
)
rendered = DeepAgentsApp._render_tool_catalog(catalog).plain
assert "2 tools available" in rendered
# Shorter name padded to the widest name in the group so columns align.
assert f" {'ls'.ljust(len('read_file'))} list" in rendered
assert " read_file read" in rendered
# No MCP groups → the `/mcp` descriptions pointer must not appear.
assert "MCP tool descriptions are available in /mcp." not in rendered
def test_render_unavailable_without_detail_omits_colon(self) -> None:
from deepagents_code.mcp_tools import MCPServerInfo
from deepagents_code.tool_catalog import (
ToolEntry,
build_catalog_from_server_info,
)
# A non-`ok` server must carry an error, but it may be empty (""), which
# yields an empty `detail` — the branch that renders status with no
# colon. Use a non-`disabled` status: `disabled` takes its own
# "disabled by user" branch, so it would not exercise this path.
catalog = build_catalog_from_server_info(
[ToolEntry(name="read_file", description="Read a file")],
[
MCPServerInfo(
name="off",
transport="http",
status="awaiting_reconnect",
error="",
)
],
)
rendered = DeepAgentsApp._render_tool_catalog(catalog).plain
assert " off awaiting_reconnect" in rendered
assert "awaiting_reconnect:" not in rendered
def test_render_includes_mcp_error(self) -> None:
from deepagents_code.tool_catalog import (
BUILT_IN_GROUP,
ToolCatalog,
ToolGroup,
)
catalog = ToolCatalog(
groups=(ToolGroup(label=BUILT_IN_GROUP, source="built-in", tools=()),),
mcp_error="MCP discovery failed; showing built-in tools only.",
)
rendered = DeepAgentsApp._render_tool_catalog(catalog).plain
assert "MCP discovery failed" in rendered
class TestFetchThreadHistoryData:
"""Verify _fetch_thread_history_data handles server-mode resume scenarios."""
@@ -212,6 +212,22 @@ class TestMCPCommand:
assert "reconnect" in keywords
class TestToolsCommand:
"""Validate the `/tools` entry specifically."""
def test_tools_registered(self) -> None:
names = {cmd.name for cmd in COMMANDS}
assert "/tools" in names
def test_tools_classified_as_queue_bound(self) -> None:
assert "/tools" in QUEUE_BOUND
assert "/tools" not in HIDDEN_COMMANDS
def test_tools_hidden_keywords_cover_mcp(self) -> None:
tools_cmd = next(cmd for cmd in COMMANDS if cmd.name == "/tools")
assert "mcp" in tools_cmd.hidden_keywords.split()
class TestGoalCommand:
"""Validate the `/goal` entry specifically.
+206 -3
View File
@@ -6,6 +6,8 @@ import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, PropertyMock, patch
import pytest
from deepagents_code.config import Settings
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
from deepagents_code.tool_catalog import (
@@ -13,11 +15,15 @@ from deepagents_code.tool_catalog import (
ToolEntry,
ToolGroup,
UnavailableServer,
_CatalogModel,
_first_line,
_load_mcp_server_info,
build_catalog_from_server_info,
collect_built_in_tools,
collect_catalog,
collect_mcp_catalog,
collect_tools_from_agent,
split_mcp_server_info,
)
# Core tools the agent always binds, independent of optional integrations.
@@ -97,6 +103,76 @@ class TestCollectBuiltInTools:
create.assert_called_once()
assert create.call_args.kwargs["assistant_id"] == "custom-agent"
def test_raises_when_compiled_agent_not_inspectable(self) -> None:
# A compiled agent whose graph does not expose the conventional tool
# node must fail loudly (documented `Raises: RuntimeError`) rather than
# silently returning an empty list — `collect_tools_from_agent` returns
# `None`, which this function turns into the raise.
agent = SimpleNamespace()
with (
patch(
"deepagents_code.agent.create_cli_agent",
return_value=(agent, None),
),
pytest.raises(RuntimeError, match="does not expose"),
):
collect_built_in_tools()
class TestCollectToolsFromAgent:
"""Tests for inspecting the tool node of an already-running local graph."""
def test_reads_bound_tools(self) -> None:
tool_node = SimpleNamespace(
tools_by_name={
"custom_search": SimpleNamespace(
description="Search custom data\nAdditional details"
)
}
)
agent = SimpleNamespace(nodes={"tools": SimpleNamespace(bound=tool_node)})
assert collect_tools_from_agent(agent) == [
ToolEntry(name="custom_search", description="Search custom data")
]
def test_returns_empty_for_local_agent_without_tool_node(self) -> None:
from langchain.agents import create_agent
agent = create_agent(model=_CatalogModel(), tools=[])
assert collect_tools_from_agent(agent) == []
def test_returns_none_for_remote_agent(self) -> None:
agent = SimpleNamespace(url="https://example.test")
assert collect_tools_from_agent(agent) is None
def test_returns_none_when_tool_node_shape_unexpected(self) -> None:
# A "tools" node exists but its `bound` object lacks a `tools_by_name`
# mapping — a LangGraph internal-shape drift. Reported as uninspectable
# (`None`), not as a validly-empty tool set (`[]`).
agent = SimpleNamespace(
nodes={"tools": SimpleNamespace(bound=SimpleNamespace())}
)
assert collect_tools_from_agent(agent) is None
def test_skips_non_string_names_and_defaults_missing_description(self) -> None:
tool_node = SimpleNamespace(
tools_by_name={
"ok": SimpleNamespace(description="Fine"),
123: SimpleNamespace(description="dropped: non-str name"),
"no_desc": SimpleNamespace(description=None),
}
)
agent = SimpleNamespace(nodes={"tools": SimpleNamespace(bound=tool_node)})
assert collect_tools_from_agent(agent) == [
ToolEntry(name="ok", description="Fine"),
ToolEntry(name="no_desc", description=""),
]
class TestCollectMcpCatalog:
"""Tests for MCP discovery: grouping tools and surfacing broken servers."""
@@ -187,9 +263,7 @@ class TestCollectMcpCatalog:
assert groups == []
assert mcp_error is None
assert unavailable == [
UnavailableServer(
name="off", status="disabled", detail="turned off via /mcp"
),
UnavailableServer(name="off", status="disabled", detail=""),
UnavailableServer(
name="pending",
status="awaiting_reconnect",
@@ -211,6 +285,135 @@ class TestCollectMcpCatalog:
assert "/path/mcp.json" not in mcp_error
class TestSplitMcpServerInfo:
"""Tests for the pure server-info splitter shared by CLI and `/tools`."""
def test_ok_server_becomes_group(self) -> None:
servers = [
MCPServerInfo(
name="docs",
transport="http",
tools=(
MCPToolInfo(name="search_docs", description="Search the docs\nX"),
),
status="ok",
),
]
groups, unavailable = split_mcp_server_info(servers)
assert groups == [
ToolGroup(
label="docs",
source="mcp",
tools=(ToolEntry(name="search_docs", description="Search the docs"),),
)
]
assert unavailable == []
def test_non_ok_server_becomes_unavailable(self) -> None:
servers = [
MCPServerInfo(
name="broken", transport="http", status="error", error="boom"
),
]
groups, unavailable = split_mcp_server_info(servers)
assert groups == []
assert unavailable == [
UnavailableServer(name="broken", status="error", detail="boom")
]
def test_pending_reenable_guidance_is_preserved(self) -> None:
# `pending_reconnect` (not the guidance text) is what keeps the detail:
# a plainly-disabled server with the same text would be blanked.
servers = [
MCPServerInfo(
name="notion",
transport="http",
status="disabled",
error="Re-enabled — press Ctrl+R to load.",
pending_reconnect=True,
),
]
groups, unavailable = split_mcp_server_info(servers)
assert groups == []
assert unavailable == [
UnavailableServer(
name="notion",
status="disabled",
detail="Re-enabled — press Ctrl+R to load.",
)
]
def test_disabled_server_detail_blanked_without_pending_reconnect(self) -> None:
# Same guidance text, but no `pending_reconnect`: the detail is dropped
# so the renderers fall back to the generic "disabled by user" label.
servers = [
MCPServerInfo(
name="notion",
transport="http",
status="disabled",
error="Re-enabled — press Ctrl+R to load.",
),
]
_, unavailable = split_mcp_server_info(servers)
assert unavailable == [
UnavailableServer(name="notion", status="disabled", detail="")
]
def test_ok_server_without_tools_is_dropped(self) -> None:
servers = [MCPServerInfo(name="empty", transport="http", status="ok")]
groups, unavailable = split_mcp_server_info(servers)
assert groups == []
assert unavailable == []
class TestBuildCatalogFromServerInfo:
"""Tests for the TUI entry point that avoids `asyncio.run` discovery."""
def test_built_in_first_then_live_mcp_no_error(self) -> None:
built_in = [ToolEntry(name="read_file", description="Read a file")]
servers = [
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="search_docs", description="Search"),),
status="ok",
),
MCPServerInfo(
name="broken", transport="http", status="error", error="boom"
),
]
catalog = build_catalog_from_server_info(built_in, servers)
assert catalog.groups[0].label == BUILT_IN_GROUP
assert catalog.groups[0].source == "built-in"
assert catalog.groups[0].tools == (
ToolEntry(name="read_file", description="Read a file"),
)
assert catalog.groups[-1].label == "docs"
assert catalog.unavailable == (
UnavailableServer(name="broken", status="error", detail="boom"),
)
# Discovery is never attempted here, so there is no discovery error.
assert catalog.mcp_error is None
def test_does_not_run_mcp_discovery(self) -> None:
# The whole point of this path is to avoid `asyncio.run`-based discovery,
# which cannot run inside a live event loop.
with patch("deepagents_code.tool_catalog._load_mcp_server_info") as loader:
build_catalog_from_server_info([], [])
loader.assert_not_called()
def test_empty_inputs_yield_only_built_in_group(self) -> None:
catalog = build_catalog_from_server_info([], [])
assert len(catalog.groups) == 1
assert catalog.groups[0].label == BUILT_IN_GROUP
assert catalog.groups[0].tools == ()
assert catalog.unavailable == ()
class TestLoadMcpServerInfo:
"""Tests for `_load_mcp_server_info` session lifecycle and cwd handling."""