Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions src/mcp_cli/commands/resources/prompts.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,17 @@
from chuk_term.ui import output


def _prompt_message_text(content) -> str:
"""Best-effort extraction of displayable text from a prompt message's content."""
if isinstance(content, str):
return content
if isinstance(content, dict):
return content.get("text") or content.get("data") or ""
if isinstance(content, list):
return "\n".join(_prompt_message_text(c) for c in content)
return str(content) if content is not None else ""


class PromptsCommand(UnifiedCommand):
"""List and manage MCP prompts."""

Expand Down Expand Up @@ -89,6 +100,32 @@ async def execute(self, **kwargs) -> CommandResult:
error="No tool manager available. Please connect to a server first.",
)

# Get mode: fetch and render a single prompt, then stop.
get_name = kwargs.get("get")
if get_name:
result = await context.tool_manager.get_prompt(get_name)
messages = (
result.get("messages", []) if isinstance(result, dict) else []
)
if not messages:
return CommandResult(
success=False,
error=f"No prompt content returned for: {get_name}",
)
parts: list[str] = []
description = (
result.get("description") if isinstance(result, dict) else None
)
if description:
parts.append(description)
for msg in messages:
role = msg.get("role", "") if isinstance(msg, dict) else ""
text = _prompt_message_text(
msg.get("content") if isinstance(msg, dict) else msg
)
parts.append(f"[{role}] {text}" if role else text)
return CommandResult(success=True, output="\n".join(parts), data=result)

# Get prompts from tool manager
prompts = await context.tool_manager.list_prompts()

Expand Down
60 changes: 48 additions & 12 deletions src/mcp_cli/commands/resources/resources.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@


class ResourcesCommand(UnifiedCommand):
"""List available MCP resources."""
"""List MCP resources, or read the contents of one."""

@property
def name(self) -> str:
Expand All @@ -26,28 +26,30 @@ def aliases(self) -> list[str]:

@property
def description(self) -> str:
return "List available MCP resources"
return "List MCP resources, or read one with --read <uri>"

@property
def help_text(self) -> str:
return """
List available MCP resources from connected servers.
List MCP resources from connected servers, or read the contents of one.

Usage:
/resources [options] - List resources (chat mode)
resources [options] - List resources (interactive mode)
mcp-cli resources - List resources (CLI mode)

/resources [options] - List resources (chat mode)
resources [options] - List resources (interactive mode)
mcp-cli resources - List resources (CLI mode)
mcp-cli resources --read <uri>- Print the contents of a single resource

Options:
--server <index> - Show resources from specific server
--raw - Output as JSON
--uri <pattern> - Filter by URI pattern
--read <uri> - Read and print the contents of the given resource URI

Examples:
/resources - List all resources
/resources --server 0 - List resources from first server
resources --raw - Output as JSON
resources --uri "file://*" - Filter file resources
/resources - List all resources
/resources --server 0 - List resources from first server
resources --raw - Output as JSON
resources --read debug://mail_log- Read a specific resource's contents
"""

@property
Expand All @@ -72,6 +74,12 @@ def parameters(self) -> list[CommandParameter]:
required=False,
help="Filter by URI pattern",
),
CommandParameter(
name="read",
type=str,
required=False,
help="Read and print the contents of the given resource URI",
),
]

async def execute(self, **kwargs) -> CommandResult:
Expand All @@ -88,6 +96,29 @@ async def execute(self, **kwargs) -> CommandResult:
error="No tool manager available. Please connect to a server first.",
)

# Read mode: dump the contents of a single resource and stop.
read_uri = kwargs.get("read")
if read_uri:
content = await context.tool_manager.read_resource(read_uri)
contents = (
content.get("contents", []) if isinstance(content, dict) else []
)
if not contents:
return CommandResult(
success=False,
error=f"No content returned for resource: {read_uri}",
)
parts: list[str] = []
for item in contents:
if item.get("text") is not None:
parts.append(item["text"])
elif item.get("blob") is not None:
mime = item.get("mimeType", "application/octet-stream")
parts.append(f"<{mime}: {len(item['blob'])} base64 bytes>")
return CommandResult(
success=True, output="\n".join(parts), data=content
)

# Get resources from tool manager
resources = await context.tool_manager.list_resources()

Expand All @@ -104,7 +135,12 @@ async def execute(self, **kwargs) -> CommandResult:
# URI might be in extra or id field
uri = resource.id or resource.extra.get("uri", "unknown")
name = resource.name or "Unnamed"
type_val = resource.type or resource.extra.get("mime_type", "unknown")
type_val = (
resource.type
or resource.extra.get("mimeType")
or resource.extra.get("mime_type")
or "unknown"
)

table_data.append(
{
Expand Down
22 changes: 16 additions & 6 deletions src/mcp_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1201,12 +1201,17 @@ async def _servers_wrapper(**params):


# Resources command
@app.command("resources", help="List available resources")
@app.command("resources", help="List resources, or read one with --read <uri>")
def resources_command(
config_file: str = typer.Option(
"server_config.json", help="Configuration file path"
),
server: str | None = typer.Option(None, help="Server to connect to"),
read: str | None = typer.Option(
None,
"--read",
help="Read and print the contents of the given resource URI",
),
provider: str = typer.Option("openai", help="LLM provider name"),
model: str | None = typer.Option(None, help="Model name"),
disable_filesystem: bool = typer.Option(False, help="Disable filesystem access"),
Expand All @@ -1217,7 +1222,7 @@ def resources_command(
log_level: str = typer.Option("WARNING", "--log-level", help="Set log level"),
theme: str = typer.Option("default", "--theme", help="UI theme"),
) -> None:
"""Show all recorded resources."""
"""Show all recorded resources, or read one with --read <uri>."""
# Configure logging and theme for this command
_setup_command_logging(quiet, verbose, log_level, theme)

Expand All @@ -1229,7 +1234,7 @@ def resources_command(
from mcp_cli.adapters.cli import cli_execute

async def _resources_wrapper(**params):
return await cli_execute("resources")
return await cli_execute("resources", read=read)

run_command_sync(
_resources_wrapper,
Expand All @@ -1243,12 +1248,17 @@ async def _resources_wrapper(**params):


# Prompts command
@app.command("prompts", help="List available prompts")
@app.command("prompts", help="List prompts, or fetch one with --get <name>")
def prompts_command(
config_file: str = typer.Option(
"server_config.json", help="Configuration file path"
),
server: str | None = typer.Option(None, help="Server to connect to"),
get: str | None = typer.Option(
None,
"--get",
help="Fetch and print a specific prompt by name",
),
provider: str = typer.Option("openai", help="LLM provider name"),
model: str | None = typer.Option(None, help="Model name"),
disable_filesystem: bool = typer.Option(False, help="Disable filesystem access"),
Expand All @@ -1259,7 +1269,7 @@ def prompts_command(
log_level: str = typer.Option("WARNING", "--log-level", help="Set log level"),
theme: str = typer.Option("default", "--theme", help="UI theme"),
) -> None:
"""Show all prompt templates."""
"""Show all prompt templates, or fetch one with --get <name>."""
# Configure logging and theme for this command
_setup_command_logging(quiet, verbose, log_level, theme)

Expand All @@ -1271,7 +1281,7 @@ def prompts_command(
from mcp_cli.adapters.cli import cli_execute

async def _prompts_wrapper(**params):
return await cli_execute("prompts")
return await cli_execute("prompts", get=get)

run_command_sync(
_prompts_wrapper,
Expand Down
59 changes: 49 additions & 10 deletions src/mcp_cli/tools/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@
)
from mcp_cli.tools.filter import DisabledReason, ToolFilter
from mcp_cli.tools.models import (
PromptInfo,
ResourceInfo,
ServerInfo,
ToolCallResult,
ToolDefinitionInput,
Expand Down Expand Up @@ -1258,15 +1260,17 @@ def get_streams(self):
logger.error(f"Error getting streams: {e}")
return []

def list_resources(self):
"""List available resources from servers."""
async def list_resources(self) -> list[ResourceInfo]:
"""List available resources from servers as ResourceInfo objects."""
if not self.stream_manager:
return []

try:
if hasattr(self.stream_manager, "list_resources"):
return self.stream_manager.list_resources()
return []
if not hasattr(self.stream_manager, "list_resources"):
return []
# StreamManager.list_resources() is async and returns list[dict].
raw = await self.stream_manager.list_resources()
return [ResourceInfo.from_raw(item) for item in (raw or [])]
except Exception as e:
logger.error(f"Error listing resources: {e}")
return []
Expand Down Expand Up @@ -1300,15 +1304,50 @@ async def read_resource(
logger.error("Error reading resource %s from %s: %s", uri, server_name, e)
return {}

def list_prompts(self):
"""List available prompts from servers."""
async def get_prompt(
self,
name: str,
arguments: dict[str, Any] | None = None,
server_name: str | None = None,
) -> dict[str, Any]:
"""Get a prompt by name.

Args:
name: Prompt name to fetch.
arguments: Optional arguments used to render the prompt template.
server_name: Optional server name to target.

Returns:
Prompt content dict from the server.
"""
if not self.stream_manager:
return []
logger.debug("get_prompt: no stream_manager available")
return {}

try:
if hasattr(self.stream_manager, "list_prompts"):
return self.stream_manager.list_prompts()
result: dict[str, Any] = await self.stream_manager.get_prompt(
name, arguments, server_name
)
if not result:
logger.debug(
"get_prompt returned empty for %s (server=%s)", name, server_name
)
return result
except Exception as e:
logger.error("Error getting prompt %s from %s: %s", name, server_name, e)
return {}

async def list_prompts(self) -> list[PromptInfo]:
"""List available prompts from servers as PromptInfo objects."""
if not self.stream_manager:
return []

try:
if not hasattr(self.stream_manager, "list_prompts"):
return []
# StreamManager.list_prompts() is async and returns list[dict].
raw = await self.stream_manager.list_prompts()
return [PromptInfo.from_raw(item) for item in (raw or [])]
except Exception as e:
logger.error(f"Error listing prompts: {e}")
return []
54 changes: 52 additions & 2 deletions src/mcp_cli/tools/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -468,13 +468,63 @@ def from_raw(cls, raw: Any) -> "ResourceInfo":
in ``extra["value"]`` so it is never lost.
"""
if isinstance(raw, dict):
known = {k: raw.get(k) for k in ("id", "name", "type")}
extra = {k: v for k, v in raw.items() if k not in known}
# MCP resources are keyed by ``uri`` and typed by ``mimeType``; map
# those onto the canonical id/type fields so the UI doesn't need to
# know the wire shape. The original keys are still kept in ``extra``.
known = {
"id": raw.get("id") or raw.get("uri"),
"name": raw.get("name"),
"type": raw.get("type") or raw.get("mimeType"),
}
extra = {k: v for k, v in raw.items() if k not in ("id", "name", "type")}
return cls(**known, extra=extra)
# primitive - wrap it
return cls(extra={"value": raw})


class PromptInfo(BaseModel):
"""
Canonical representation of *one* prompt entry as returned by
``prompts.list``.

Mirrors :class:`ResourceInfo`: the common fields used in the UI are
normalised and **all additional keys** are preserved inside ``extra``.
"""

# Common attributes we frequently need in the UI
name: str | None = None
description: str | None = None
arguments: list[Any] = Field(default_factory=list)

# Anything else goes here …
extra: dict[str, Any] = Field(default_factory=dict)

model_config = {"frozen": False, "arbitrary_types_allowed": True}

# ------------------------------------------------------------------ #
# Factory helpers
# ------------------------------------------------------------------ #
@classmethod
def from_raw(cls, raw: Any) -> "PromptInfo":
"""
Convert a raw list item (dict | str | …) into a PromptInfo.

If *raw* is not a mapping we treat it as an opaque scalar and store it
in ``extra["value"]`` so it is never lost.
"""
if isinstance(raw, dict):
known = ("name", "description", "arguments")
extra = {k: v for k, v in raw.items() if k not in known}
return cls(
name=raw.get("name"),
description=raw.get("description"),
arguments=raw.get("arguments") or [],
extra=extra,
)
# primitive - wrap it
return cls(extra={"value": raw})


# ──────────────────────────────────────────────────────────────────────────────
# Transport and Server Configuration Models
# ──────────────────────────────────────────────────────────────────────────────
Expand Down
Loading