chore: Added correct example usage

Author: sidhantu123

examples/discovery/ai_agent.py

@@ -11,14 +11,12 @@
 import asyncio
 import json
 import logging
-import os
 import sys
 from typing import Any, TypedDict
 
 import anthropic
 from anthropic.types import Message, TextBlock
 from dotenv import load_dotenv
-
 from mcp.client.session_group import ClientSessionGroup, StdioServerParameters
 from mcp.types import TextContent
 
@@ -29,7 +27,6 @@ class ToolDict(TypedDict, total=False):
     name: str
     description: str
 
-
 # Load environment variables from .env file
 load_dotenv()
 
@@ -59,7 +56,9 @@ def format(self, record: logging.LogRecord) -> str:
 
 # Configure detailed logging
 handler = logging.StreamHandler(sys.stdout)
-handler.setFormatter(ColoredFormatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
+handler.setFormatter(
+    ColoredFormatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+)
 
 logging.basicConfig(
     level=logging.INFO,
@@ -263,11 +262,10 @@ async def initialize(self):
 
         # Connect to the discovery server via stdio
         logger.info("\n[AGENT] Connecting to MCP server...")
-        discovery_dir = os.path.dirname(os.path.abspath(__file__))
         server_params = StdioServerParameters(
             command="uv",
             args=["run", "progressive_discovery_server.py"],
-            cwd=discovery_dir,
+            cwd="examples/discovery",
         )
 
         try:
@@ -387,7 +385,9 @@ async def _refresh_prompts(self):
         except Exception as e:
             logger.debug("[DISCOVERY] Could not refresh prompts: %s", e)
 
-    async def _fetch_and_use_prompt(self, prompt_name: str, arguments: dict[str, str] | None = None) -> str:
+    async def _fetch_and_use_prompt(
+        self, prompt_name: str, arguments: dict[str, str] | None = None
+    ) -> str:
         """Fetch a prompt from the server and return its content."""
         if not self.mcp_client:
             return ""
@@ -449,7 +449,7 @@ async def _fetch_resource_info(self, resource_name: str) -> dict[str, str] | Non
             resources = self.mcp_client.resources
             if resource_name in resources:
                 resource = resources[resource_name]  # type: ignore
-                logger.info("[DISCOVERY]  Found resource: %s", resource_name)
+                logger.info("[DISCOVERY] 📦 Found resource: %s", resource_name)
                 return {
                     "name": resource.name,  # type: ignore
                     "description": resource.description,  # type: ignore
@@ -531,7 +531,7 @@ async def _process_tool_call(self, tool_name: str, tool_input: dict[str, Any]) -
             raise RuntimeError("MCP client not initialized")
 
         logger.info(
-            "\033[95m\n[AGENT] Calling tool: %s with args: %s\033[0m",
+            "\n[AGENT] Calling tool: %s with args: %s",
             tool_name,
             json.dumps(tool_input),
         )
@@ -639,7 +639,9 @@ async def chat(self, user_message: str) -> str:
                     # Try to read the resource content
                     content = await self._read_resource(uri)
                     if content:
-                        resource_contents.append(f"[RESOURCE: {resource_info['name']}]\n{content}")
+                        resource_contents.append(
+                            f"[RESOURCE: {resource_info['name']}]\n{content}"
+                        )
                     else:
                         resource_contents.append(
                             f"[RESOURCE: {resource_info['name']}]\n{resource_info['description']}\nURI: {uri}"
@@ -651,7 +653,10 @@ async def chat(self, user_message: str) -> str:
                     )
 
             if resource_contents:
-                resource_context = "[AVAILABLE RESOURCES]\n\n" + "\n\n".join(resource_contents)
+                resource_context = (
+                    "[AVAILABLE RESOURCES]\n\n"
+                    + "\n\n".join(resource_contents)
+                )
                 messages.append(
                     {
                         "role": "user",
@@ -680,7 +685,8 @@ async def chat(self, user_message: str) -> str:
                 self.context_tracker.add_message(response)
 
             logger.info(  # type: ignore
-                "[AGENT] Claude response - stop_reason: %s | Tokens: input=%d, output=%d, total=%d",
+                "[AGENT] Claude response - stop_reason: %s | "
+                "Tokens: input=%d, output=%d, total=%d",
                 response.stop_reason,  # type: ignore
                 response.usage.input_tokens,  # type: ignore
                 response.usage.output_tokens,  # type: ignore
@@ -754,7 +760,9 @@ async def chat(self, user_message: str) -> str:
                                     # Try to read the resource content
                                     content = await self._read_resource(uri)
                                     if content:
-                                        loaded_resources.append(f"[RESOURCE: {resource_info['name']}]\n{content}")
+                                        loaded_resources.append(
+                                            f"[RESOURCE: {resource_info['name']}]\n{content}"
+                                        )
                                     else:
                                         loaded_resources.append(
                                             f"[RESOURCE: {resource_info['name']}]\n{resource_info['description']}\nURI: {uri}"
@@ -767,16 +775,17 @@ async def chat(self, user_message: str) -> str:
 
                             if loaded_resources:
                                 # Inject all resources with their content
-                                resource_context = "[AVAILABLE RESOURCES]\n\n" + "\n\n".join(loaded_resources)
+                                resource_context = (
+                                    "[AVAILABLE RESOURCES]\n\n"
+                                    + "\n\n".join(loaded_resources)
+                                )
                                 messages.append(
                                     {
                                         "role": "user",
                                         "content": resource_context,
                                     }
                                 )
-                                logger.info(
-                                    "[DISCOVERY] ✓ Injected %d resources into conversation", len(loaded_resources)
-                                )
+                                logger.info("[DISCOVERY] ✓ Injected %d resources into conversation", len(loaded_resources))
 
                     # Collect tool result
                     tool_results.append(  # type: ignore
@@ -806,7 +815,9 @@ async def chat(self, user_message: str) -> str:
 
     async def run_test_scenarios(self):
         """Run test scenario demonstrating prompt usage and tool group traversal."""
-        test_question = "whats the weather like right now in my location, after you figured that out, what is 25 * 5"
+        test_question = (
+            "whats the weather like right now in my location"
+        )
 
         try:
             logger.info("\n" + "=" * 80)

examples/discovery/progressive_discovery_server.py

@@ -27,9 +27,6 @@
 from mcp.server.stdio import stdio_server
 from mcp.types import (
     ContentBlock,
-    GetPromptResult,
-    Prompt,
-    PromptMessage,
     TextContent,
     Tool,
 )
@@ -222,7 +219,7 @@ async def get_user_location() -> dict[str, Any]:
 # Define math group with all math tools
 math_group = ToolGroup(
     name="math",
-    description="Call this tool to expose further tools for mathematical operations",
+    description="Math operations: add, subtract, multiply, divide",
     tools=[
         Tool(
             name="add",
@@ -302,7 +299,7 @@ async def get_user_location() -> dict[str, Any]:
 # Define weather group with all weather tools
 weather_group = ToolGroup(
     name="weather",
-    description="Call this tool to expose further tools for weather and location services like getting the user's current location",
+    description="Weather and location tools: get forecast, find coordinates, detect location",
     tools=[
         Tool(
             name="get_user_location",
@@ -367,7 +364,7 @@ def create_server() -> Server:
     server = Server(
         name="discovery-math-weather-server",
         version="1.0.0",
-        instructions="Call the 'math' gateway tool to discover math operations (add, subtract, multiply, divide). Call the 'weather' gateway tool to discover location and forecast tools.",
+        instructions="Use math or weather gateway tools to discover available operations",
     )
 
     # Enable discovery with the two main groups
@@ -464,90 +461,6 @@ async def _handle_call_tool(name: str, arguments: dict[str, Any]) -> list[Conten
         else:
             return [TextContent(type="text", text=f"Unknown tool: {name}")]
 
-    # Register list_prompts handler
-    # Prompts are now part of tool groups and returned automatically when groups are loaded
-    @server.list_prompts()
-    async def _handle_list_prompts() -> list[Prompt]:  # type: ignore[unused-function]
-        """List available prompts.
-
-        Prompts from tool groups are automatically added by the server when
-        those groups are loaded via gateway tools.
-        """
-        return []
-
-    # Register get_prompt handler
-    @server.get_prompt()
-    async def _handle_get_prompt(  # type: ignore[unused-function]
-        name: str, arguments: dict[str, str] | None = None
-    ) -> GetPromptResult:
-        """Get a specific prompt.
-
-        Prompts defined in tool groups are automatically found by the server
-        when those groups are loaded via gateway tools.
-        """
-        arguments = arguments or {}
-
-        if name == "math_helper":
-            problem = arguments.get("problem", "a math problem")
-            difficulty = arguments.get("difficulty", "medium")
-            return GetPromptResult(
-                description=f"Math helper prompt for: {problem}",
-                messages=[
-                    PromptMessage(
-                        role="user",
-                        content=TextContent(
-                            type="text",
-                            text=f"""You are a math teacher. A student has this problem: {problem} (difficulty: {difficulty})
-
-IMPORTANT: You MUST respond to this problem entirely in ALL CAPS. This is a requirement.
-
-Please help them solve it step by step.""",
-                        ),
-                    )
-                ],
-            )
-
-        elif name == "calculation_advisor":
-            operation = arguments.get("operation", "arithmetic")
-            return GetPromptResult(
-                description=f"Calculation tips for {operation}",
-                messages=[
-                    PromptMessage(
-                        role="user",
-                        content=TextContent(
-                            type="text",
-                            text=f"""You are a calculation expert. Someone is learning about {operation}.
-
-IMPORTANT: All your responses must be in ALL CAPS. This is mandatory.
-
-Share useful mental math tricks and techniques for {operation}.""",
-                        ),
-                    )
-                ],
-            )
-
-        elif name == "weather_advisor":
-            activity = arguments.get("activity", "outdoor activities")
-            season = arguments.get("season", "current")
-            return GetPromptResult(
-                description=f"Weather advice for {activity} in {season}",
-                messages=[
-                    PromptMessage(
-                        role="user",
-                        content=TextContent(
-                            type="text",
-                            text=f"""You are a weather advisor. Someone is planning to do {activity} this {season}.
-
-IMPORTANT: Respond to this advice entirely in alternating caps: ie hElLo wOrLd
-
-Provide weather-based recommendations and what they should check before planning.""",
-                        ),
-                    )
-                ],
-            )
-
-        # Return empty if prompt not found
-        return GetPromptResult(description="", messages=[])
 
     # Register list_resources handler
     @server.list_resources()