Skip to content

Conversation

@nnennandukwe
Copy link
Owner

@nnennandukwe nnennandukwe commented Oct 13, 2025

User description

Replace custom MCP server implementation with FastMCP for simplified server setup and tool registration. Migrate from manual protocol handling to decorator-based tool definition while maintaining the same keyword search functionality.


PR Type

Enhancement


Description

  • Migrated from custom MCP server implementation to FastMCP framework

  • Replaced manual protocol handling with decorator-based tool registration

  • Simplified server setup by removing ~170 lines of boilerplate code

  • Updated dependency from mcp to fastmcp package


Diagram Walkthrough

flowchart LR
  A["Custom MCP Server"] --> B["FastMCP Framework"]
  B --> C["Decorator-based Tools"]
  B --> D["Simplified Setup"]
  C --> E["keyword_search tool"]
Loading

File Walkthrough

Relevant files
Enhancement
server.py
Refactor server to use FastMCP with decorators                     

src/workshop_mcp/server.py

  • Replaced WorkshopMCPServer class with FastMCP instance and
    decorator-based tool
  • Removed manual protocol handlers (handle_list_tools, handle_call_tool)
  • Converted keyword_search to decorated async function with inline
    documentation
  • Simplified sync_main() to use mcp.run() instead of asyncio.run(main())
+49/-188
Dependencies
pyproject.toml
Switch MCP dependency to FastMCP package                                 

pyproject.toml

  • Updated dependency from mcp = "^1.0.0" to fastmcp = "^0.2.0"
+1/-1     

Replace custom MCP server implementation with FastMCP for simplified server setup and tool registration. Migrate from manual protocol handling to decorator-based tool definition while maintaining the same keyword search functionality.
@qodo-code-review
Copy link

qodo-code-review bot commented Oct 13, 2025

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 2 🔵🔵⚪⚪⚪
🧪 No relevant tests
🔒 No security concerns identified
⚡ Recommended focus areas for review

Logging/Observability

The previous implementation logged execution and errors; the new FastMCP version only emits an info message via Context and relies on implicit error handling. Consider whether additional logging or structured error returns are needed to preserve observability and backward compatibility for clients expecting error payloads.

await ctx.info(
    f"Executing keyword search for '{keyword}' in {len(root_paths)} paths"
)

# Delegate to the existing async tool implementation
result = await _keyword_search_tool.execute(keyword, root_paths)

# FastMCP automatically sends structured content for dict-like results
return result
Argument Validation

Input validation is delegated to type hints and Field constraints. Validate that FastMCP enforces list item type for root_paths and provides clear errors; if not, consider a runtime check or a Pydantic model to ensure each path is a non-empty string and to normalize/resolve paths if required.

root_paths: Annotated[
    list[str],
    Field(description="List of directory paths to search in", min_length=1),
],
ctx: Context,
Backward Compatibility

The old server returned JSON text blocks on success and structured JSON on errors; the new version returns a dict directly. Confirm that downstream clients can handle structured outputs and that the CLI/stdio behavior matches previous expectations, including error surface and tool metadata.

# Delegate to the existing async tool implementation
result = await _keyword_search_tool.execute(keyword, root_paths)

# FastMCP automatically sends structured content for dict-like results
return result

Below is a summary of compliance checks for this PR:

Security Compliance
🟢
No security concerns identified No security vulnerabilities detected by AI analysis. Human verification advised for critical code.
Ticket Compliance
🎫 No ticket provided
  • Create ticket/issue
Codebase Duplication Compliance
Codebase context is not defined

Follow the guide to enable codebase context checks.

Custom Compliance
No custom compliance provided

Follow the guide to enable custom compliance check.

Compliance status legend 🟢 - Fully Compliant
🟡 - Partial Compliant
🔴 - Not Compliant
⚪ - Requires Further Human Verification
🏷️ - Compliance label

@qodo-code-review
Copy link

qodo-code-review bot commented Oct 13, 2025

PR Code Suggestions ✨

Explore these optional code suggestions:

CategorySuggestion                                                                                                                                    Impact
General
Re-add essential server logging configuration

Re-add the logging.basicConfig call that was removed during refactoring. This
ensures that log messages from all application modules are captured, which is
essential for debugging and monitoring.

[src/workshop_mcp/server.py [20-23]](https://github.com/nnennandukwe/python-mcp-agent-workshop/pull/1/files#diff-7f72f26c48b307f814d7428fd520b2633b6ed7fdad9bbcbd12261b02590cc8e1R20-R23)

+import logging
+import sys
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[logging.StreamHandler(sys.stderr)],
+)
+
 # Instantiate FastMCP server
 mcp = FastMCP(
     name="workshop-mcp-server",
 )
  • Apply / Chat
Suggestion importance[1-10]: 8

__

Why: This suggestion correctly points out that removing the logging.basicConfig call silences logs from other modules like keyword_search.py, which is a significant regression in observability for debugging and monitoring. Reinstating the logging configuration is crucial for a production-ready server.

Medium
Remove redundant try-except block

Remove the redundant try...except...raise block from the keyword_search
function, as the FastMCP framework should handle exception logging and error
responses automatically.

[src/workshop_mcp/server.py [34-76]](https://github.com/nnennandukwe/python-mcp-agent-workshop/pull/2/files#diff-7f72f26c48b307f814d7428fd520b2633b6ed7fdad9bbcbd12261b02590cc8e1R34-R76)

 @mcp.tool()
 async def keyword_search(keyword: str, root_paths: List[str]) -> Dict[str, Any]:
     """
     Search for keyword occurrences across directory trees.
     
     Supports multiple text file formats (.py, .java, .js, .ts, .html, .css, 
     .json, .xml, .md, .txt, .yml, .yaml, etc.) and provides detailed 
     statistics about matches.
     
     Args:
         keyword: The keyword to search for (case-sensitive)
         root_paths: List of directory paths to search in
         
     Returns:
         Dictionary containing search results with file paths, occurrence counts,
         and summary statistics including:
         - keyword: The searched keyword
         - root_paths: List of searched directories
         - files: Dictionary mapping file paths to occurrence data
         - summary: Statistics including total files searched, matches found,
                   total occurrences, and most frequent file
                   
     Raises:
         ValueError: If keyword is empty or root_paths is empty
         FileNotFoundError: If any root path doesn't exist
     """
-    try:
-        logger.info(f"Executing keyword search for '{keyword}' in {len(root_paths)} paths")
-        
-        result = await keyword_search_tool.execute(keyword, root_paths)
-        
-        logger.info(
-            f"Search completed successfully: "
-            f"{result['summary']['total_files_searched']} files searched, "
-            f"{result['summary']['total_occurrences']} occurrences found"
-        )
-        
-        return result
-        
-    except Exception as e:
-        error_msg = f"Error executing keyword_search: {str(e)}"
-        logger.error(error_msg)
-        raise
+    logger.info(f"Executing keyword search for '{keyword}' in {len(root_paths)} paths")
+    
+    result = await keyword_search_tool.execute(keyword, root_paths)
+    
+    logger.info(
+        f"Search completed successfully: "
+        f"{result['summary']['total_files_searched']} files searched, "
+        f"{result['summary']['total_occurrences']} occurrences found"
+    )
+    
+    return result
  • Apply / Chat
Suggestion importance[1-10]: 6

__

Why: The suggestion correctly identifies that the try...except...raise block is likely redundant, as a framework like FastMCP is expected to handle exceptions in decorated tool functions, thus simplifying the code and preventing potential double logging.

Low
Improve input validation for keywords

Add strip_whitespace=True to the keyword field's Pydantic validation. This
prevents whitespace-only strings from passing validation, aligning it with the
business logic's requirements.

[src/workshop_mcp/server.py [45-48]](https://github.com/nnennandukwe/python-mcp-agent-workshop/pull/1/files#diff-7f72f26c48b307f814d7428fd520b2633b6ed7fdad9bbcbd12261b02590cc8e1R45-R48)

 keyword: Annotated[
     str,
-    Field(description="The keyword to search for (case-sensitive)", min_length=1),
+    Field(
+        description="The keyword to search for (case-sensitive)",
+        min_length=1,
+        strip_whitespace=True,
+    ),
 ],
  • Apply / Chat
Suggestion importance[1-10]: 6

__

Why: The suggestion correctly identifies that whitespace-only strings would pass the Pydantic validation and cause an error later. Adding strip_whitespace=True improves input validation by handling this case at the framework level, which is consistent with the PR's goal of using FastMCP for validation.

Low
Use built-in generic types

In the keyword_search function signature, replace List and Dict from the typing
module with the built-in list and dict types, as supported by Python 3.10+.

[src/workshop_mcp/server.py [35]](https://github.com/nnennandukwe/python-mcp-agent-workshop/pull/2/files#diff-7f72f26c48b307f814d7428fd520b2633b6ed7fdad9bbcbd12261b02590cc8e1R35-R35)

-from typing import Any, Dict, List
+from typing import Any
 
+...
+
+@mcp.tool()
+async def keyword_search(keyword: str, root_paths: list[str]) -> dict[str, Any]:
+

[To ensure code accuracy, apply this suggestion manually]

Suggestion importance[1-10]: 4

__

Why: The suggestion correctly recommends using modern built-in generics (list, dict) for type hints, which is appropriate given the project's Python 3.10+ requirement and improves code style.

Low
Possible issue
Add structured error handling for tools

Wrap the call to _keyword_search_tool.execute in a try...except block. This
allows catching specific exceptions like FileNotFoundError to send a more
informative error message to the user via ctx.error before re-raising the
exception.

[src/workshop_mcp/server.py [44-69]](https://github.com/nnennandukwe/python-mcp-agent-workshop/pull/1/files#diff-7f72f26c48b307f814d7428fd520b2633b6ed7fdad9bbcbd12261b02590cc8e1R44-R69)

 async def keyword_search(
     keyword: Annotated[
         str,
         Field(description="The keyword to search for (case-sensitive)", min_length=1),
     ],
     root_paths: Annotated[
         list[str],
         Field(description="List of directory paths to search in", min_length=1),
     ],
     ctx: Context,
 ) -> dict[str, Any]:
     """Execute an asynchronous keyword search across one or more root paths.
 
     Parameters are validated and documented via type annotations; FastMCP will
     generate the input schema automatically. Returning a dict enables structured
     output in addition to traditional content blocks.
     """
     await ctx.info(
         f"Executing keyword search for '{keyword}' in {len(root_paths)} paths"
     )
 
-    # Delegate to the existing async tool implementation
-    result = await _keyword_search_tool.execute(keyword, root_paths)
+    try:
+        # Delegate to the existing async tool implementation
+        result = await _keyword_search_tool.execute(keyword, root_paths)
 
-    # FastMCP automatically sends structured content for dict-like results
-    return result
+        # FastMCP automatically sends structured content for dict-like results
+        return result
+    except (FileNotFoundError, ValueError) as e:
+        await ctx.error(f"Failed to execute keyword search: {e}")
+        raise
  • Apply / Chat
Suggestion importance[1-10]: 7

__

Why: The suggestion correctly identifies that unhandled exceptions from the tool's logic lead to generic framework errors. Adding a try...except block to log a specific error with ctx.error before re-raising improves user experience by providing more context, which is a valuable enhancement.

Medium
Avoid forced sys.exit on errors

In sync_main, remove the sys.exit() calls from the exception handlers. Instead,
let the function exit normally after a KeyboardInterrupt and re-raise other
exceptions after logging them.

[src/workshop_mcp/server.py [79-93]](https://github.com/nnennandukwe/python-mcp-agent-workshop/pull/2/files#diff-7f72f26c48b307f814d7428fd520b2633b6ed7fdad9bbcbd12261b02590cc8e1R79-R93)

 def sync_main() -> None:
     """
     Synchronous entry point for the MCP server.
     
     This function is used as the script entry point in pyproject.toml.
     """
+    logger.info("Starting Workshop MCP Server with FastMCP")
     try:
-        logger.info("Starting Workshop MCP Server with FastMCP")
         mcp.run()
     except KeyboardInterrupt:
         logger.info("Server stopped by user")
-        sys.exit(0)
     except Exception as e:
-        logger.error(f"Fatal error: {e}")
-        sys.exit(1)
+        logger.exception("Fatal error while running FastMCP server")
+        raise
  • Apply / Chat
Suggestion importance[1-10]: 7

__

Why: This is a valuable suggestion for robust error handling, as removing sys.exit() and re-raising exceptions allows for better debugging and makes the application more composable.

Medium
  • More

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants