refactor: FastMCP server migration refactor part 1

[nnennandukwe]

User description

The changes implement a migration from a manual MCP server implementation to FastMCP, consisting of:

1. Dependency Updates (pyproject.toml):

• Added fastmcp = "^2.10.0" and pydantic = "^2.8.0" dependencies

2. Server Rewrite (src/workshop_mcp/server.py):

• Replaced 236-line manual MCP server implementation with 79-line FastMCP-based server
• Converted from class-based WorkshopMCPServer to functional approach using decorators
• Simplified protocol handling by leveraging FastMCP's high-level API
• Preserved CLI entrypoint and tool behavior while adding structured output capabilities

---

PR Type

Enhancement

---

Description

• Migrated from manual MCP server to FastMCP framework

• Reduced server implementation from 236 to 79 lines

• Added structured output capabilities with Pydantic validation

• Preserved CLI entrypoint and tool behavior

---

Diagram Walkthrough

flowchart LR
  A["Manual MCP Server"] --> B["FastMCP Framework"]
  B --> C["Decorator-based Tools"]
  B --> D["Automatic Schema Generation"]
  B --> E["Structured Output"]
  F["236 Lines"] --> G["79 Lines"]

Loading

File Walkthrough

	Relevant files
Enhancement	server.py Complete server rewrite using FastMCP framework                    src/workshop_mcp/server.py Replaced 236-line manual MCP server with 79-line FastMCP implementation Converted from class-based WorkshopMCPServer to functional decorator approach Added structured output support with Pydantic field validation Simplified protocol handling using FastMCP's high-level API +61/-218
Dependencies	pyproject.toml Added FastMCP and Pydantic dependencies                                    pyproject.toml Added fastmcp = "^2.10.0" dependency for new framework Added pydantic = "^2.8.0" for structured data validation +2/-0

---

[qodo-code-review]

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

[qodo-code-review]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	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] +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
	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] 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
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] 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
More

[coditamar]

/improve
--pr_code_suggestions.commitable_code_suggestions=true

[qodo-code-review]

Suggestion: Wrap the mcp.run() call in sync_main with a try...except block to gracefully handle KeyboardInterrupt for shutdown and catch other potential exceptions. [general, importance: 7]

Suggested change

	def sync_main() -> None:
	"""
	Synchronous entry point for the MCP server.
	This function is used as the script entry point in pyproject.toml.
	"""
	try:
	asyncio.run(main())
	except KeyboardInterrupt:
	logger.info("Server stopped by user")
	sys.exit(0)
	except Exception as e:
	logger.error(f"Fatal error: {e}")
	sys.exit(1)
	"""Synchronous entry point for the FastMCP server (stdio by default)."""
	# FastMCP defaults to stdio transport, preserving original behavior
	mcp.run()
	def sync_main() -> None:
	"""Synchronous entry point for the FastMCP server (stdio by default)."""
	try:
	mcp.run()
	except KeyboardInterrupt:
	# Graceful shutdown
	return
	except Exception as e:
	# Emit a minimal error and non-zero exit for critical failures
	raise SystemExit(f"Server terminated due to error: {e}")