Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/health_check.py:9-30 (handler)

  The get_health tool handler, decorated with @mcp.tool. It checks the Prefect server's health by calling client.hello() and returns the status or error as TextContent.

  @mcp.tool
  async def get_health() -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get health status of the Prefect server.
      
      Returns:
          Health status information
      """
      try:
          # Test connection to Prefect by calling the health endpoint
          async with get_client() as client:
              health_status = await client.hello()
              
          return [types.TextContent(type="text", text=str(health_status))]
      
      except Exception as e:
          error_status = {
              "status": "unhealthy",
              "message": f"Error connecting to Prefect server: {str(e)}"
          }
          
          return [types.TextContent(type="text", text=str(error_status))]