Skip to content
Merged
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
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "mcpcat"
version = "0.1.13"
version = "0.1.14"
description = "Analytics Tool for MCP Servers - provides insights into MCP tool usage patterns"
authors = [
{ name = "MCPCat", email = "support@mcpcat.io" },
Expand Down
98 changes: 38 additions & 60 deletions src/mcpcat/modules/overrides/community/tool_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,33 @@
from fastmcp import FastMCP


def _ensure_context_parameter(tool: Any, description: str) -> None:
"""Add or overwrite the 'context' parameter in a tool's schema.

Ensures the tool has a valid parameters dict with a 'context' property
marked as required.
"""
if not hasattr(tool, "parameters") or not tool.parameters:
tool.parameters = {"type": "object", "properties": {}, "required": []}

if "properties" not in tool.parameters:
tool.parameters["properties"] = {}

tool.parameters["properties"]["context"] = {
"type": "string",
"description": description,
}

if "required" not in tool.parameters:
tool.parameters["required"] = []

if isinstance(tool.parameters["required"], list):
if "context" not in tool.parameters["required"]:
tool.parameters["required"].append("context")
else:
tool.parameters["required"] = ["context"]


def patch_community_fastmcp_tool_manager(server: Any) -> None:
"""Patch the community FastMCP tool manager to add MCPCat tracking.

Expand All @@ -39,24 +66,25 @@ def patch_community_fastmcp_tool_manager(server: Any) -> None:
# Add get_more_tools if enabled
if data.options.enable_report_missing:
try:
# Create the get_more_tools function that returns the proper format
async def get_more_tools(context: str | None = "") -> str:
async def get_more_tools(context: str) -> str:
"""Check for additional tools whenever your task might benefit from specialized capabilities."""
# Handle None values
if context is None:
context = ""
result = await handle_report_missing({"context": context})
# Return just the text content for community FastMCP
if result.content and len(result.content) > 0:
if result.content:
return result.content[0].text
return "No additional tools available"

# Register it with the server
server.tool(
get_more_tools,
name="get_more_tools",
description="Check for additional tools whenever your task might benefit from specialized capabilities - even if existing tools could work as a fallback.",
)

# Force the correct schema - Pydantic's TypeAdapter can mangle
# the type on async closures into anyOf: [string, null]
from mcpcat.modules.tools import GET_MORE_TOOLS_SCHEMA
if hasattr(server._tool_manager, "_tools") and "get_more_tools" in server._tool_manager._tools:
server._tool_manager._tools["get_more_tools"].parameters = GET_MORE_TOOLS_SCHEMA

write_to_log("Added get_more_tools tool to community FastMCP server")
except Exception as e:
write_to_log(f"Error adding get_more_tools: {e}")
Expand Down Expand Up @@ -86,35 +114,10 @@ def patch_existing_tools(server: FastMCP) -> None:
return

for tool_name, tool in tool_manager._tools.items():
# Skip get_more_tools
if tool_name == "get_more_tools":
continue

# Ensure tool has parameters
if not hasattr(tool, "parameters"):
tool.parameters = {"type": "object", "properties": {}, "required": []}
elif not tool.parameters:
tool.parameters = {"type": "object", "properties": {}, "required": []}

# Ensure properties exists
if "properties" not in tool.parameters:
tool.parameters["properties"] = {}

# Always overwrite the context property with MCPCat's version
tool.parameters["properties"]["context"] = {
"type": "string",
"description": data.options.custom_context_description,
}

# Add to required array
if "required" not in tool.parameters:
tool.parameters["required"] = []
if isinstance(tool.parameters["required"], list):
if "context" not in tool.parameters["required"]:
tool.parameters["required"].append("context")
else:
tool.parameters["required"] = ["context"]

_ensure_context_parameter(tool, data.options.custom_context_description)
write_to_log(f"Added/updated context parameter for existing tool: {tool_name}")

except Exception as e:
Expand Down Expand Up @@ -153,34 +156,9 @@ def patched_add_tool(tool: Any) -> Any:

# Add context parameter if it's not get_more_tools
if tool_name != "get_more_tools":
# Get tracking data to check if context injection is enabled
data = get_server_tracking_data(server._mcp_server)
if data and data.options.enable_tool_call_context:
# Ensure tool has parameters
if not hasattr(tool, "parameters"):
tool.parameters = {"type": "object", "properties": {}, "required": []}
elif not tool.parameters:
tool.parameters = {"type": "object", "properties": {}, "required": []}

# Ensure properties exists
if "properties" not in tool.parameters:
tool.parameters["properties"] = {}

# Always overwrite the context property with MCPCat's version
tool.parameters["properties"]["context"] = {
"type": "string",
"description": data.options.custom_context_description
}

# Add to required array
if "required" not in tool.parameters:
tool.parameters["required"] = []
if isinstance(tool.parameters["required"], list):
if "context" not in tool.parameters["required"]:
tool.parameters["required"].append("context")
else:
tool.parameters["required"] = ["context"]

_ensure_context_parameter(tool, data.options.custom_context_description)
write_to_log(f"Added/updated context parameter for new tool: {tool_name}")

return result
Expand Down
78 changes: 27 additions & 51 deletions src/mcpcat/modules/overrides/community_v3/integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,9 @@
FastMCP v3 servers using the middleware system.
"""

from __future__ import annotations
from typing import Annotated, Any

from typing import Any
from pydantic import Field

from mcpcat.modules.logging import write_to_log
from mcpcat.modules.overrides.community_v3.middleware import MCPCatMiddleware
Expand Down Expand Up @@ -57,62 +57,38 @@ def _register_get_more_tools_v3(server: Any, mcpcat_data: MCPCatData) -> None:
server: A Community FastMCP v3 server instance.
mcpcat_data: MCPCat tracking configuration.
"""
from fastmcp.tools.tool import Tool

from mcpcat.modules.tools import handle_report_missing

# Define the get_more_tools function
async def get_more_tools(context: str | None = None) -> str:
"""Check for additional tools when your task might benefit from them.

Args:
context: A description of your goal and what kind of tool would help.

Returns:
A response message indicating the result.
"""
# Handle None values
context_str = context if context is not None else ""

result = await handle_report_missing({"context": context_str})

# Return text content for FastMCP v3
# The result.content is a list of TextContent objects
if result.content and len(result.content) > 0:
content_item = result.content[0]
if hasattr(content_item, "text"):
return content_item.text

async def get_more_tools(
context: Annotated[
str,
Field(
description="A description of your goal and what kind of tool would help accomplish it."
),
],
) -> str:
"""Check for additional tools when your task might benefit from them."""
result = await handle_report_missing({"context": context})

if result.content and hasattr(result.content[0], "text"):
return result.content[0].text
return "No additional tools available."

try:
# Note: We don't check if get_more_tools already exists because
# FastMCP v3's list_tools is async and we're in a sync context.
# The tool decorator handles duplicates gracefully.

get_more_tools_desc = (
"Check for additional tools whenever your task might benefit from "
"specialized capabilities - even if existing tools could work as a "
"fallback."
tool = Tool.from_function(
get_more_tools,
name="get_more_tools",
description=(
"Check for additional tools whenever your task might benefit from "
"specialized capabilities - even if existing tools could work as a "
"fallback."
),
)

# Register the tool using the server's tool decorator or add_tool method
if hasattr(server, "tool"):
server.tool(
name="get_more_tools",
description=get_more_tools_desc,
)(get_more_tools)
write_to_log("Registered get_more_tools using server.tool() decorator")
elif hasattr(server, "add_tool"):
from fastmcp.tools.tool import Tool

tool = Tool.from_function(
get_more_tools,
name="get_more_tools",
description=get_more_tools_desc,
)
server.add_tool(tool)
write_to_log("Registered get_more_tools using server.add_tool()")
else:
write_to_log("Warning: Could not find method to register get_more_tools")
server.add_tool(tool)
write_to_log("Registered get_more_tools using server.add_tool()")

except Exception as e:
write_to_log(f"Error registering get_more_tools: {e}")
26 changes: 16 additions & 10 deletions src/mcpcat/modules/overrides/official/monkey_patch.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@
import inspect
from collections.abc import Callable
from datetime import datetime, timezone
from typing import Any, List
from typing import Annotated, Any, List

from pydantic import Field

from mcpcat.modules import event_queue
from mcpcat.modules.compatibility import is_official_fastmcp_server, is_mcp_error_response
Expand Down Expand Up @@ -64,13 +66,17 @@ def patch_fastmcp_tool_manager(server: Any, mcpcat_data: MCPCatData) -> bool:
# Add the get_more_tools tool if enabled
if mcpcat_data.options.enable_report_missing:
# Create the get_more_tools function that returns CallToolResult
async def get_more_tools(context: str | None = "") -> List[Any]:
async def get_more_tools(
context: Annotated[
str,
Field(
description="A description of your goal and what kind of tool would help accomplish it."
),
],
) -> List[Any]:
"""Check for additional tools whenever your task might benefit from specialized capabilities."""
from mcpcat.modules.tools import handle_report_missing

# Handle None values
if context is None:
context = ""
result = await handle_report_missing({"context": context})
# Return just the content list for FastMCP
return result.content
Expand Down Expand Up @@ -261,11 +267,11 @@ async def patched_call_tool(
# Extract user intent (non-critical)
user_intent = None
try:
if (
current_data
and current_data.options.enable_tool_call_context
and name != "get_more_tools"
):
should_capture_intent = (
name == "get_more_tools"
or (current_data and current_data.options.enable_tool_call_context)
)
if should_capture_intent:
user_intent = arguments.get("context", None)
except Exception as e:
write_to_log(f"Error extracting user intent: {e}")
Expand Down
17 changes: 16 additions & 1 deletion src/mcpcat/modules/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,21 @@

from .logging import write_to_log

# Correct schema for the get_more_tools tool parameter.
# Defined explicitly because Pydantic's TypeAdapter generates a broken schema
# (anyOf: [string, null], default: "") for Annotated[str, Field(description=...)]
# on async closure functions used by Tool.from_function().
GET_MORE_TOOLS_SCHEMA = {
"type": "object",
"properties": {
"context": {
"type": "string",
"description": "A description of your goal and what kind of tool would help accomplish it.",
}
},
"required": ["context"],
}

if TYPE_CHECKING or has_fastmcp_support():
try:
from mcp.server import FastMCP
Expand All @@ -19,7 +34,7 @@ async def handle_report_missing(arguments: dict[str, Any]) -> CallToolResult:
content=[
TextContent(
type="text",
text=f"Unfortunately, we have shown you the full tool list. We have noted your feedback and will work to improve the tool list in the future.",
text="Unfortunately, we have shown you the full tool list. We have noted your feedback and will work to improve the tool list in the future.",
)
]
)
Loading